diff --git a/web/hwlab-cloud-web/src/api/workbench-events.ts b/web/hwlab-cloud-web/src/api/workbench-events.ts index 5d0d6431..419edc25 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.ts @@ -73,7 +73,15 @@ export interface WorkbenchSyncReplayResponse { scope?: Record | null; cursor?: Record | null; events?: WorkbenchRealtimeEvent[]; - delta?: WorkbenchRealtimeEvent[]; + delta?: WorkbenchRealtimeEvent[] | { + sessions?: Record[]; + messages?: Record[]; + parts?: Record[]; + turns?: Record[]; + traceEvents?: Record[]; + checkpoints?: Record[]; + [key: string]: unknown; + }; families?: Record | null; authority?: Record | null; [key: string]: unknown; diff --git a/web/hwlab-cloud-web/src/stores/workbench-colada.test.ts b/web/hwlab-cloud-web/src/stores/workbench-colada.test.ts index b886da60..128e7b1c 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-colada.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-colada.test.ts @@ -7,10 +7,12 @@ import { createApp } from "vue"; import { createPinia, setActivePinia } from "pinia"; import { PiniaColada, useQueryCache } from "@pinia/colada"; -import { createWorkbenchServerState, type WorkbenchServerState } from "./workbench-server-state"; +import { createWorkbenchServerState, reduceWorkbenchServerState, selectActiveMessages, selectTurnStatusAuthority, type WorkbenchServerState } from "./workbench-server-state"; import { workbenchColadaKeys } from "./workbench-colada-keys"; import { useWorkbenchColadaReducer } from "./workbench-colada-reducer"; import { workbenchSyncReplayDiagnostic, workbenchSyncReplayEvents } from "./workbench-realtime-authority"; +import { reduceWorkbenchRealtimeEvent } from "./workbench-event-reducer"; +import type { TurnStatusAuthority } from "./workbench-session"; const storeDir = path.dirname(fileURLToPath(import.meta.url)); @@ -102,3 +104,45 @@ test("workbench sync replay diagnostic exposes authority and terminal seal metad valuesRedacted: true }); }); + +test("workbench sync replay projects durable message and turn facts into observer state", () => { + const sessionId = "ses_sync_cross_page"; + const traceId = "trc_sync_cross_page"; + const payload = { + contractVersion: "workbench-sync-v1", + realtimeAuthority: "workbench-realtime-authority-v2", + cursor: { outboxSeq: 31 }, + events: [], + delta: { + messages: [ + { messageId: "msg_sync_cross_page_user", sessionId, traceId, turnId: traceId, role: "user", status: "sent", text: "run it", projectedSeq: 10, updatedAt: "2026-07-08T12:40:00.000Z", valuesRedacted: true }, + { messageId: "msg_sync_cross_page_agent", sessionId, traceId, turnId: traceId, role: "agent", status: "running", text: "", projectedSeq: 11, updatedAt: "2026-07-08T12:40:01.000Z", valuesRedacted: true } + ], + turns: [ + { turnId: traceId, sessionId, traceId, messageId: "msg_sync_cross_page_agent", status: "running", running: true, terminal: false, projectedSeq: 11, updatedAt: "2026-07-08T12:40:01.000Z", valuesRedacted: true } + ] + } + }; + + const events = workbenchSyncReplayEvents(payload); + assert.deepEqual(events.map((event) => event.type), ["message.snapshot", "message.snapshot", "turn.snapshot"]); + assert.equal(events.every((event) => reduceWorkbenchRealtimeEvent(event, `workbench.${event.type}`).action.type !== "ignore"), true); + + let state = createWorkbenchServerState(); + state = reduceWorkbenchServerState(state, { type: "session.detail", session: { sessionId, status: "running", messages: [] } }); + for (const event of events) { + const action = reduceWorkbenchRealtimeEvent(event, `workbench.${event.type}`).action; + if (action.type === "message.snapshot") { + state = reduceWorkbenchServerState(state, { type: "message.snapshot", sessionId: action.realtimeEvent.sessionId, message: action.realtimeEvent.message }); + } else if (action.type === "turn.snapshot") { + state = reduceWorkbenchServerState(state, { type: "turn.status", turn: action.turn as TurnStatusAuthority }); + } + } + + const messages = selectActiveMessages(state, sessionId); + assert.equal(messages.length, 2); + assert.equal(messages[0]?.role, "user"); + assert.equal(messages[1]?.role, "agent"); + assert.equal(messages[1]?.traceId, traceId); + assert.equal(selectTurnStatusAuthority(state)[traceId]?.status, "running"); +}); diff --git a/web/hwlab-cloud-web/src/stores/workbench-realtime-authority.ts b/web/hwlab-cloud-web/src/stores/workbench-realtime-authority.ts index aaf0781a..075931b7 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-realtime-authority.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-realtime-authority.ts @@ -2,6 +2,7 @@ // Responsibility: Frontend authority gates for Workbench realtime events and sync replay batches. import type { WorkbenchRealtimeEvent, WorkbenchSyncReplayResponse } from "@/api/workbench-events"; +import type { ChatMessage } from "@/types"; export const WORKBENCH_REALTIME_AUTHORITY_VERSION = "workbench-realtime-authority-v2"; export const WORKBENCH_SYNC_CONTRACT_VERSION = "workbench-sync-v1"; @@ -83,7 +84,7 @@ export function workbenchRealtimeEntityAuthority(event: WorkbenchRealtimeEvent): export function workbenchSyncReplayEvents(payload: WorkbenchSyncReplayResponse | null | undefined): WorkbenchRealtimeEvent[] { const events = arrayOfRecords(payload?.events); - const delta = arrayOfRecords(payload?.delta); + const delta = workbenchSyncReplayDeltaEvents(payload); const seen = new Set(); const output: WorkbenchRealtimeEvent[] = []; for (const value of [...events, ...delta]) { @@ -96,6 +97,16 @@ export function workbenchSyncReplayEvents(payload: WorkbenchSyncReplayResponse | return output; } +function workbenchSyncReplayDeltaEvents(payload: WorkbenchSyncReplayResponse | null | undefined): WorkbenchRealtimeEvent[] { + const explicitEvents = arrayOfRecords(payload?.delta); + const delta = recordValue(payload?.delta); + const replayEvents = explicitEvents as WorkbenchRealtimeEvent[]; + if (!delta || Array.isArray(payload?.delta)) return replayEvents; + const messages = arrayOfRecords(delta.messages).map((message) => syncReplayMessageSnapshotEvent(message, payload)).filter((event): event is WorkbenchRealtimeEvent => Boolean(event)); + const turns = arrayOfRecords(delta.turns).map((turn) => syncReplayTurnSnapshotEvent(turn, payload)).filter((event): event is WorkbenchRealtimeEvent => Boolean(event)); + return [...replayEvents, ...messages, ...turns]; +} + export function workbenchSyncReplayDiagnostic(payload: WorkbenchSyncReplayResponse | null | undefined, events: WorkbenchRealtimeEvent[], input: { reason: string; sinceOutboxSeq?: number | null }): Record { const eventRows = Array.isArray(events) ? events : []; const cursor = recordValue(payload?.cursor); @@ -142,6 +153,75 @@ function syncEventKey(event: WorkbenchRealtimeEvent): string { return [event.type ?? "~", event.traceId ?? "~", event.sessionId ?? "~", event.cursor?.outboxSeq ?? event.outboxSeq ?? "~", event.cursor?.traceSeq ?? event.traceSeq ?? "~"].join("|"); } +function syncReplayMessageSnapshotEvent(message: Record, payload: WorkbenchSyncReplayResponse | null | undefined): WorkbenchRealtimeEvent | null { + const sessionId = textValue(message.sessionId); + if (!sessionId) return null; + const traceId = textValue(message.traceId); + const role = textValue(message.role) ?? "agent"; + const id = textValue(message.messageId ?? message.id) ?? [traceId, role].filter(Boolean).join(":"); + if (!id) return null; + const entity = syncReplayEntity("messages", id, message, payload); + return { + type: "message.snapshot", + contractVersion: stringValue(payload?.contractVersion), + realtimeAuthority: stringValue(payload?.realtimeAuthority), + sessionId, + traceId, + reason: "sync-delta", + cursor: syncReplayCursor(message, payload), + entity, + projectionRevision: entity.projectionRevision, + message: message as ChatMessage, + valuesRedacted: true + }; +} + +function syncReplayTurnSnapshotEvent(turn: Record, payload: WorkbenchSyncReplayResponse | null | undefined): WorkbenchRealtimeEvent | null { + const traceId = textValue(turn.traceId); + const turnId = textValue(turn.turnId ?? traceId); + if (!traceId && !turnId) return null; + const entity = syncReplayEntity("turns", turnId ?? traceId ?? "turn", turn, payload); + return { + type: "turn.snapshot", + contractVersion: stringValue(payload?.contractVersion), + realtimeAuthority: stringValue(payload?.realtimeAuthority), + sessionId: textValue(turn.sessionId), + traceId, + reason: "sync-delta", + cursor: syncReplayCursor(turn, payload), + entity, + projectionRevision: entity.projectionRevision, + turn, + valuesRedacted: true + }; +} + +function syncReplayEntity(family: string, id: string, fact: Record, payload: WorkbenchSyncReplayResponse | null | undefined): NonNullable { + const traceSeq = finiteNumber(fact.projectedSeq ?? fact.sourceSeq ?? fact.seq); + const outboxSeq = finiteNumber(payload?.cursor?.outboxSeq ?? fact.outboxSeq); + const version = finiteNumber(fact.projectionRevision ?? fact.projectedSeq ?? fact.sourceSeq ?? fact.seq ?? outboxSeq) ?? 0; + const projectionRevision = textValue(fact.projectionRevision ?? fact.projectedSeq ?? fact.sourceSeq ?? fact.seq ?? version) ?? "0"; + return { + family, + id, + version, + entityVersion: version, + outboxSeq, + traceSeq, + projectionRevision, + serverCommittedAt: textValue(fact.updatedAt ?? fact.createdAt ?? fact.occurredAt), + authority: "workbench-sync-delta", + detailProjection: false + }; +} + +function syncReplayCursor(fact: Record, payload: WorkbenchSyncReplayResponse | null | undefined): NonNullable { + return { + traceSeq: finiteNumber(fact.projectedSeq ?? fact.sourceSeq ?? fact.seq), + outboxSeq: finiteNumber(payload?.cursor?.outboxSeq ?? fact.outboxSeq) + }; +} + function workbenchRealtimeEventHasTerminalSeal(event: WorkbenchRealtimeEvent): boolean { const turn = recordValue(event.turn); if (turn?.terminal === true) return true; @@ -184,6 +264,11 @@ function stringValue(value: unknown): string | null { return text || null; } +function textValue(value: unknown): string | null { + const text = typeof value === "string" ? value.trim() : value === null || value === undefined ? "" : String(value).trim(); + return text || null; +} + function finiteNumber(value: unknown): number | null { const number = Number(value); return Number.isFinite(number) && number >= 0 ? Math.trunc(number) : null; diff --git a/web/hwlab-cloud-web/src/stores/workbench-server-state.test.ts b/web/hwlab-cloud-web/src/stores/workbench-server-state.test.ts index e297c98a..f02ce844 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-server-state.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-server-state.test.ts @@ -123,6 +123,20 @@ test("server state reducer upserts terminal authority when the session page has assert.equal(terminal.text, "final answer from turn authority"); }); +test("server state reducer appends authoritative agent snapshot after durable user snapshot", () => { + const sessionId = "ses_state_cross_page_snapshot"; + const traceId = "trc_state_cross_page_snapshot"; + let state = createWorkbenchServerState(); + state = reduceWorkbenchServerState(state, { type: "session.detail", session: sessionRecord({ sessionId, status: "running", messages: [] }) }); + state = reduceWorkbenchServerState(state, { type: "message.snapshot", sessionId, message: userMessage({ id: "msg_state_cross_page_user", text: "run it", sessionId, traceId }) }); + state = reduceWorkbenchServerState(state, { type: "message.snapshot", sessionId, message: agentMessage({ id: "msg_state_cross_page_agent", status: "running", sessionId, traceId }) }); + + const messages = selectActiveMessages(state, sessionId); + assert.equal(messages.length, 2); + assert.equal(messages[0]?.role, "user"); + assert.equal(messages[1]?.role, "agent"); +}); + test("server state reducer keeps sealed terminal messages when a stale partial snapshot arrives", () => { const sessionId = "ses_state_partial_snapshot"; const traceId = "trc_state_partial_snapshot"; diff --git a/web/hwlab-cloud-web/src/stores/workbench-server-state.ts b/web/hwlab-cloud-web/src/stores/workbench-server-state.ts index 99852af7..bf77b27f 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-server-state.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-server-state.ts @@ -187,7 +187,7 @@ function reduceMessageSnapshot(state: WorkbenchServerState, sessionId: string | if (!id || !message) return state; const existing = state.messagesBySessionId[id] ?? state.sessionsById[id]?.messages ?? []; const index = existing.findIndex((item) => messageMatchesSnapshot(item, message)); - if (index < 0 && message.role === "agent" && existing.length > 0) return state; + if (index < 0 && message.role === "agent" && existing.length > 0 && !canAppendAuthoritativeAgentSnapshot(existing, message)) return state; const merged = index >= 0 ? existing.map((item, itemIndex) => itemIndex === index ? mergeMessageSnapshot(item, message) : item) : [...existing, message]; @@ -220,6 +220,16 @@ function messageMatchesSnapshot(existing: ChatMessage, incoming: ChatMessage): b return Boolean(existing.traceId && incoming.traceId && existing.traceId === incoming.traceId && existing.role === incoming.role); } +function canAppendAuthoritativeAgentSnapshot(existing: ChatMessage[], incoming: ChatMessage): boolean { + const incomingId = incoming.messageId ?? incoming.id; + const incomingTraceId = incoming.traceId ?? incoming.runnerTrace?.traceId; + if (!incomingId || !incomingTraceId) return false; + return existing.some((message) => { + const existingTraceId = message.traceId ?? message.runnerTrace?.traceId; + return existingTraceId === incomingTraceId; + }); +} + function mergeMessageSnapshot(existing: ChatMessage, incoming: ChatMessage): ChatMessage { const normalizedIncoming = normalizeUnsealedTerminalMessage(incoming); const merged = {