Merge pull request #2437 from pikasTech/issue-1600-cross-page-projection

fix: converge Workbench sync replay projection

web/hwlab-cloud-web/src/api/workbench-events.ts

@@ -73,7 +73,15 @@ export interface WorkbenchSyncReplayResponse {
   scope?: Record<string, unknown> | null;
   cursor?: Record<string, unknown> | null;
   events?: WorkbenchRealtimeEvent[];
-  delta?: WorkbenchRealtimeEvent[];
+  delta?: WorkbenchRealtimeEvent[] | {
+    sessions?: Record<string, unknown>[];
+    messages?: Record<string, unknown>[];
+    parts?: Record<string, unknown>[];
+    turns?: Record<string, unknown>[];
+    traceEvents?: Record<string, unknown>[];
+    checkpoints?: Record<string, unknown>[];
+    [key: string]: unknown;
+  };
   families?: Record<string, unknown> | null;
   authority?: Record<string, unknown> | null;
   [key: string]: unknown;

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");
+});

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<string>();
   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<string, unknown> {
   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<string, unknown>, 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<string, unknown>, 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<string, unknown>, payload: WorkbenchSyncReplayResponse | null | undefined): NonNullable<WorkbenchRealtimeEvent["entity"]> {
+  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<string, unknown>, payload: WorkbenchSyncReplayResponse | null | undefined): NonNullable<WorkbenchRealtimeEvent["cursor"]> {
+  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;

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";

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 = {