fix(workbench): stream terminal message snapshots (#1996)

internal/cloud/server-workbench-http.ts

@@ -177,6 +177,16 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
                 cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq },
                 projectionStatus: { terminal: row.terminal, sealed: row.sealed }
               });
+              await writeMessageRealtimeSnapshotSafe({
+                writeEvent,
+                readModel,
+                actor: auth.actor,
+                sessionId: rowSessionId,
+                threadId: streamThreadId,
+                traceId: rowTraceId,
+                reason: "outbox-terminal",
+                cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq }
+              });
               if (rowTraceId) void writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: rowTraceId, reason: "terminal" });
             } else {
               writeEvent("workbench.trace.event", {
@@ -352,6 +362,37 @@ function handleWorkbenchRealtimeFailure(request, response, url, options = {}, er
   if (!response.destroyed) sendJson(response, classified.statusCode, workbenchError(classified.code, classified.message, { route: "/v1/workbench/events", errorName: error?.name ?? "Error" }));
 }
 
+async function writeMessageRealtimeSnapshotSafe({ writeEvent, readModel, actor, sessionId, threadId = null, traceId = null, reason = "message", cursor = null } = {}) {
+  const safeSession = safeSessionId(sessionId);
+  if (!safeSession || typeof readModel?.queryFacts !== "function") return;
+  try {
+    const result = await queryFactsByRouteId(readModel, safeSession, { families: WORKBENCH_SESSION_MESSAGE_PAGE_FAMILIES });
+    if (result.error) return;
+    const session = visibleFactSessions(result.facts, actor)[0] ?? null;
+    if (!session) return;
+    const messages = factMessagesForSession(session, result.facts);
+    const message = [...messages].reverse().find((item) => {
+      if (!isAssistantLikeRole(item?.role)) return false;
+      return !traceId || item.traceId === traceId;
+    }) ?? null;
+    if (!message) return;
+    writeEvent("workbench.message.snapshot", {
+      type: "message.snapshot",
+      sessionId: factSessionId(session),
+      threadId: threadId ?? session.threadId ?? null,
+      traceId: message.traceId ?? traceId ?? null,
+      reason,
+      cursor,
+      message
+    });
+  } catch (error) {
+    writeEvent("workbench.error", {
+      type: "error",
+      error: { code: "workbench_message_snapshot_failed", message: error?.message ?? "Workbench realtime message snapshot failed." }
+    });
+  }
+}
+
 function realtimeEventCreatedAt(payload) {
   const candidates = [payload?.event?.createdAt, payload?.snapshot?.lastEvent?.createdAt, payload?.turn?.updatedAt];
   for (const value of candidates) {

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

@@ -1,7 +1,7 @@
 // SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
 // Responsibility: Workbench SSE client. Realtime events accelerate UI projection; REST snapshots remain gap-fill authority.
 
-import type { ProjectionDiagnostic, TraceEvent } from "@/types";
+import type { ChatMessage, ProjectionDiagnostic, TraceEvent } from "@/types";
 import { recordWorkbenchSseLifecycle } from "@/utils/workbench-performance";
 
 export interface WorkbenchRealtimeTraceSnapshot {
@@ -29,6 +29,7 @@ export interface WorkbenchRealtimeEvent {
   traceId?: string | null;
   event?: TraceEvent | null;
   snapshot?: WorkbenchRealtimeTraceSnapshot | null;
+  message?: ChatMessage | null;
   turn?: Record<string, unknown> | null;
   reason?: string | null;
   error?: { code?: string; message?: string; [key: string]: unknown } | null;
@@ -54,6 +55,7 @@ const WORKBENCH_EVENT_NAMES = [
   "workbench.trace.snapshot",
   "workbench.trace.event",
   "workbench.trace.unavailable",
+  "workbench.message.snapshot",
   "workbench.turn.snapshot",
   "workbench.heartbeat",
   "workbench.error"

web/hwlab-cloud-web/src/stores/workbench-server-state.ts

@@ -17,6 +17,7 @@ export type WorkbenchServerAction =
   | { type: "session.list"; sessions: WorkbenchSessionRecord[] }
   | { type: "session.detail"; session: WorkbenchSessionRecord | null | undefined }
   | { type: "session.messages"; sessionId: string | null | undefined; messages: ChatMessage[] }
+  | { type: "message.snapshot"; sessionId: string | null | undefined; message: ChatMessage | null | undefined }
   | { type: "session.status"; session: SessionStatusAuthority }
   | { type: "session.forget"; sessionId: string }
   | { type: "turn.status"; turn: TurnStatusAuthority }
@@ -42,6 +43,8 @@ export function reduceWorkbenchServerState(state: WorkbenchServerState, action:
       return reduceSessionDetail(state, action.session);
     case "session.messages":
       return reduceSessionMessages(state, action.sessionId, action.messages);
+    case "message.snapshot":
+      return reduceMessageSnapshot(state, action.sessionId, action.message);
     case "session.status":
       return { ...state, sessionStatusById: { ...state.sessionStatusById, [action.session.sessionId]: action.session } };
     case "session.forget":
@@ -140,6 +143,34 @@ function reduceSessionMessages(state: WorkbenchServerState, sessionId: string |
   };
 }
 
+function reduceMessageSnapshot(state: WorkbenchServerState, sessionId: string | null | undefined, message: ChatMessage | null | undefined): WorkbenchServerState {
+  const id = sessionId || message?.sessionId;
+  if (!id || !message) return state;
+  const existing = state.messagesBySessionId[id] ?? state.sessionsById[id]?.messages ?? [];
+  const index = existing.findIndex((item) => messageMatchesSnapshot(item, message));
+  const merged = index >= 0
+    ? existing.map((item, itemIndex) => itemIndex === index ? mergeMessageSnapshot(item, message) : item)
+    : [...existing, message];
+  return reduceSessionMessages(state, id, merged);
+}
+
+function messageMatchesSnapshot(existing: ChatMessage, incoming: ChatMessage): boolean {
+  const existingId = existing.messageId ?? existing.id;
+  const incomingId = incoming.messageId ?? incoming.id;
+  if (existingId && incomingId && existingId === incomingId) return true;
+  return Boolean(existing.traceId && incoming.traceId && existing.traceId === incoming.traceId && existing.role === incoming.role);
+}
+
+function mergeMessageSnapshot(existing: ChatMessage, incoming: ChatMessage): ChatMessage {
+  return {
+    ...existing,
+    ...incoming,
+    runnerTrace: existing.runnerTrace ?? incoming.runnerTrace ?? null,
+    traceAutoLifecycle: existing.traceAutoLifecycle,
+    updatedAt: incoming.updatedAt ?? existing.updatedAt
+  };
+}
+
 function mergeSessionRecord(existing: WorkbenchSessionRecord | undefined, incoming: WorkbenchSessionRecord): WorkbenchSessionRecord {
   if (!existing) return incoming;
   const incomingMessages = Array.isArray(incoming.messages) ? incoming.messages : undefined;

web/hwlab-cloud-web/src/stores/workbench.ts

@@ -1178,6 +1178,10 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
       applyRealtimeTraceEvent(event.traceId, event.event, event.snapshot, event);
       return;
     }
+    if (event.type === "message.snapshot" && event.message) {
+      applyRealtimeMessageSnapshot(event);
+      return;
+    }
     if (event.type === "turn.snapshot" && event.turn) {
       applyRealtimeTurnSnapshot(event.turn);
       return;
@@ -1192,6 +1196,15 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
     }
   }
 
+  function applyRealtimeMessageSnapshot(event: WorkbenchRealtimeEvent): void {
+    const message = event.message ? normalizeChatMessage(event.message) : null;
+    const sessionId = normalizeWorkbenchSessionId(firstNonEmptyString(event.sessionId, message?.sessionId));
+    if (!message || !sessionId) return;
+    const activeId = activeSessionId.value;
+    if (activeId && sessionId !== activeId) return;
+    reduceServerState({ type: "message.snapshot", sessionId, message });
+  }
+
   function applyRealtimeTraceSnapshot(traceId: string | null | undefined, snapshot: WorkbenchRealtimeEvent["snapshot"]): void {
     const id = firstNonEmptyString(traceId, snapshot?.traceId);
     if (!id || !snapshot) return;