workbench consume facts-only realtime snapshots

internal/cloud/server-workbench-http.test.ts

@@ -1522,7 +1522,7 @@ test("workbench trace events API blocks historical projectedSeq collisions inste
   }
 });
 
-test("workbench realtime stream accepts session authority without project or workspace", async () => {
+test("workbench realtime stream surfaces facts blocker instead of legacy trace fallback", async () => {
   const traceStore = createCodeAgentTraceStore();
   const results = createCodeAgentChatResultStore();
   const traceId = "trc_workbench_realtime";
@@ -1538,8 +1538,8 @@ test("workbench realtime stream accepts session authority without project or wor
     updatedAt: "2026-06-17T02:00:00.000Z",
     session: { sessionStatus: "running", lastTraceId: traceId, messages: [{ role: "user", text: "stream", traceId }] }
   };
-  traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" });
-  results.set(traceId, { status: "running", traceId, ownerUserId: ACTOR.id, sessionId: session.id, threadId: session.threadId });
+  traceStore.append(traceId, { type: "assistant", status: "completed", label: "assistant:completed", terminal: true, message: "legacy trace answer" });
+  results.set(traceId, { status: "completed", traceId, ownerUserId: ACTOR.id, sessionId: session.id, threadId: session.threadId, finalResponse: { text: "legacy result answer" } });
   const accessController = {
     store: {
       async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
@@ -1554,18 +1554,24 @@ test("workbench realtime stream accepts session authority without project or wor
   try {
     const { port } = server.address();
     const eventsPromise = getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}`, 4);
-    setTimeout(() => traceStore.append(traceId, { type: "backend", status: "running", label: "backend:live" }), 50);
     const events = await eventsPromise;
-    assert.deepEqual(events.slice(0, 3).map((event) => event.event), [
+    assert.deepEqual(events.map((event) => event.event), [
       "workbench.connected",
+      "workbench.error",
       "workbench.trace.snapshot",
       "workbench.turn.snapshot"
     ]);
     assert.equal(events[0].data.filters.sessionId, session.id);
-    assert.equal(events[1].data.traceId, traceId);
-    assert.equal(events[2].data.turn.traceId, traceId);
-    assert.equal(events[3].event, "workbench.trace.event");
-    assert.equal(events[3].data.event.label, "backend:live");
+    assert.equal(events[1].data.error.code, "workbench_facts_session_missing");
+    assert.equal(events[2].data.traceId, traceId);
+    assert.equal(events[2].data.snapshot.status, "unknown");
+    assert.equal(events[2].data.snapshot.eventCount, 0);
+    assert.equal(events[3].data.turn.traceId, traceId);
+    assert.equal(events[3].data.turn.status, "unknown");
+    assert.equal(events[3].data.turn.terminal, false);
+    assert.equal(events[3].data.turn.finalResponse, null);
+    assert.equal(JSON.stringify(events).includes("legacy result answer"), false);
+    assert.equal(JSON.stringify(events).includes("legacy trace answer"), false);
   } finally {
     await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
   }

internal/cloud/server-workbench-http.ts

@@ -2221,14 +2221,16 @@ async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId,
   }
   const sessionId = safeSessionId(context.result?.sessionId ?? context.result?.session?.sessionId ?? context.session?.id) ?? null;
   const threadId = safeOpaqueId(context.result?.threadId ?? context.result?.session?.threadId ?? context.session?.threadId) ?? (textValue(context.session?.threadId) || null);
+  const realtimeTraceSnapshot = context.factsReadBlocker ? blockedRealtimeTraceSnapshot(context) : traceSnapshotForRealtime(context.trace);
+  const traceSeq = context.factsReadBlocker ? 0 : traceSnapshotLastSeq(context.trace);
   writeEvent("workbench.trace.snapshot", {
     type: "trace.snapshot",
     reason,
     sessionId,
     threadId,
     traceId,
-    snapshot: { ...traceSnapshotForRealtime(context.trace), sessionId, threadId },
-    cursor: { traceSeq: traceSnapshotLastSeq(context.trace) }
+    snapshot: { ...realtimeTraceSnapshot, sessionId, threadId },
+    cursor: { traceSeq }
   });
   writeEvent("workbench.turn.snapshot", {
     type: "turn.snapshot",
@@ -2237,21 +2239,12 @@ async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId,
     threadId,
     traceId,
     turn: realtimeTurnSnapshot(context),
-    cursor: { traceSeq: traceSnapshotLastSeq(context.trace) }
+    cursor: { traceSeq }
   });
 }
 
 function realtimeTurnSnapshot(context) {
-  if (context?.factsReadBlocker) {
-    return {
-      ...turnSnapshot(context),
-      projectionStatus: "blocked",
-      projectionHealth: "degraded",
-      blocker: context.factsReadBlocker,
-      projection: blockedTraceEventProjection(context.projection ?? {}, context.factsReadBlocker),
-      valuesRedacted: true
-    };
-  }
+  if (context?.factsReadBlocker) return blockedRealtimeTurnSnapshot(context, context.factsReadBlocker);
   const facts = context?.facts;
   const traceId = safeTraceId(context?.traceId);
   if (traceId && facts && context?.factSession) {
@@ -2259,7 +2252,76 @@ function realtimeTurnSnapshot(context) {
     const projected = factTurnSnapshot({ turn: factTurn, session: context.factSession, facts, traceId, turnId: context.turnId });
     if (projected) return projected;
   }
-  return turnSnapshot(context);
+  const blocker = traceEventsReadModelBlocker("workbench_facts_turn_missing", "Workbench durable facts did not contain a turn snapshot; realtime must surface the read-model gap instead of falling back to legacy trace projection.", { traceId, session: context?.factSession, route: "/v1/workbench/events" });
+  return blockedRealtimeTurnSnapshot(context, blocker);
+}
+
+function blockedRealtimeTraceSnapshot(context) {
+  const traceId = safeTraceId(context?.traceId ?? context?.factsReadBlocker?.traceId) ?? null;
+  const blocker = context?.factsReadBlocker ?? traceEventsReadModelBlocker("workbench_facts_unavailable", "Workbench durable facts are unavailable for realtime trace snapshot.", { traceId, route: "/v1/workbench/events" });
+  const projection = blockedTraceEventProjection({}, blocker);
+  return {
+    traceId,
+    status: "unknown",
+    traceStatus: "unknown",
+    events: [],
+    eventCount: 0,
+    fullTraceLoaded: true,
+    hasMore: false,
+    nextProjectedSeq: 0,
+    updatedAt: null,
+    projection,
+    projectionStatus: projection.projectionStatus,
+    projectionHealth: projection.projectionHealth,
+    staleMs: projection.staleMs ?? null,
+    blocker,
+    valuesRedacted: true,
+    secretMaterialStored: false
+  };
+}
+
+function blockedRealtimeTurnSnapshot(context, blocker) {
+  const traceId = safeTraceId(context?.traceId ?? blocker?.traceId) ?? null;
+  const turnId = safeTurnId(context?.turnId ?? blocker?.turnId) || traceId;
+  const sessionId = safeSessionId(blocker?.sessionId ?? factSessionId(context?.factSession) ?? context?.session?.id) ?? null;
+  const threadId = safeOpaqueId(context?.factSession?.threadId ?? context?.session?.threadId) ?? (textValue(context?.session?.threadId) || null);
+  const projection = blockedTraceEventProjection({}, blocker);
+  return {
+    turnId,
+    traceId,
+    status: "unknown",
+    running: false,
+    terminal: false,
+    sessionId,
+    threadId,
+    userMessageId: null,
+    assistantMessageId: null,
+    assistantText: null,
+    finalResponse: null,
+    timing: null,
+    startedAt: null,
+    lastEventAt: null,
+    finishedAt: null,
+    durationMs: null,
+    agentRun: null,
+    trace: {
+      traceId,
+      status: "unknown",
+      eventCount: 0,
+      updatedAt: null
+    },
+    urls: {
+      self: turnId ? `/v1/workbench/turns/${encodeURIComponent(turnId)}` : null,
+      traceEvents: traceId ? `/v1/workbench/traces/${encodeURIComponent(traceId)}/events` : null
+    },
+    projectionStatus: projection.projectionStatus,
+    projectionHealth: projection.projectionHealth,
+    staleMs: projection.staleMs ?? null,
+    blocker,
+    projection,
+    valuesRedacted: true,
+    secretMaterialStored: false
+  };
 }
 
 async function writeTraceRealtimeSnapshotSafe(input) {
@@ -2305,6 +2367,15 @@ async function visibleTraceContext(options, actor, traceId) {
     }
   }
   const factSession = factArray(facts.sessions).find((item) => factSessionId(item) === factSessionIdValue) ?? null;
+  if (!factsReadBlocker) {
+    if (!factSessionIdValue) {
+      factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_session_id_missing", "Workbench realtime could not resolve a durable session id; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" });
+    } else if (typeof readModel.queryFacts !== "function") {
+      factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_unavailable", "Workbench durable facts query is unavailable; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" });
+    } else if (!factSession) {
+      factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_session_missing", "Workbench durable facts did not contain the trace session; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" });
+    }
+  }
   return { visible: true, turnId: projection.turnId, traceId, status: projection.status, projection, result, session, trace, facts, factSession, factsReadBlocker };
 }
 

web/hwlab-cloud-web/scripts/check.ts

@@ -64,6 +64,8 @@ const appSource = readCloudWebAppSource(rootDir);
 const workbenchStoreSource = readWeb("src/stores/workbench.ts");
 const traceTimelineSource = readWeb("src/components/agent/TraceTimeline.vue");
 const messageTraceDebugSource = readWeb("src/components/workbench/MessageTraceDebugPanel.vue");
+const conversationPanelSource = readWeb("src/components/workbench/ConversationPanel.vue");
+const serverWorkbenchHttpSource = fs.readFileSync(path.resolve(rootDir, "..", "..", "internal/cloud/server-workbench-http.ts"), "utf8");
 
 assert.match(html, /<div id="app"><\/div>/u, "index.html must expose Vue mount root");
 assert.match(html, /src="\/src\/main\.ts"/u, "index.html must load Vue TS entry");
@@ -102,6 +104,11 @@ assertIncludes(appSource, "activityRef", "Code Agent inactivity-timeout activity
 assertIncludes(workbenchStoreSource, "connectWorkbenchEvents", "Workbench must use the unified SSE realtime entry");
 assert.doesNotMatch(workbenchStoreSource, /scheduleRealtimeGapHydration|hydrateRealtimeGap/u, "Workbench realtime consumer must not repair projection through REST gap fill");
 assert.doesNotMatch(workbenchStoreSource, /subscribeToTrace|TRACE_POLL_INTERVAL_MS/u, "Workbench store must not reintroduce active trace polling");
+assert.doesNotMatch(workbenchStoreSource, /message\.runnerTrace(?:\?\.|\.)status/u, "Workbench message lifecycle must not be inferred from runnerTrace.status");
+assert.doesNotMatch(conversationPanelSource, /message\.runnerTrace(?:\?\.|\.)status/u, "ConversationPanel must not override message completion from runnerTrace.status");
+assert.doesNotMatch(serverWorkbenchHttpSource, /return\s+turnSnapshot\(context\)|\.\.\.turnSnapshot\(context\)|blockedTraceEventProjection\(context\.projection/u, "Realtime turn snapshots must not fall back to legacy trace/result projection");
+assertIncludes(serverWorkbenchHttpSource, "blockedRealtimeTurnSnapshot", "Realtime read-model gaps must produce blocked diagnostic turn snapshots");
+assertIncludes(serverWorkbenchHttpSource, "workbench_facts_session_missing", "Realtime read-model gaps must expose a facts-missing blocker");
 assertIncludes(appSource, "/v1/workbench/events", "Workbench realtime client must use the RESTful same-origin events endpoint");
 assertIncludes(appSource, "/v1/workbench/traces/", "trace hydration must use Workbench read-model trace API");
 assert.doesNotMatch(appSource, /\/v1\/agent\/(?:turns|traces|chat\/result)\//u, "Cloud Web must not call legacy Code Agent read-through turn/trace/result APIs");

web/hwlab-cloud-web/src/components/workbench/ConversationPanel.vue

@@ -36,8 +36,7 @@ const scrollSignature = computed(() => workbench.activeMessages.map((message) =>
   message.status,
   visibleMessageText(message).length,
   message.updatedAt ?? "",
-  message.runnerTrace?.eventCount ?? message.runnerTrace?.events?.length ?? 0,
-  message.runnerTrace?.status ?? ""
+  message.runnerTrace?.eventCount ?? message.runnerTrace?.events?.length ?? 0
 ].join(":")));
 
 watch(scrollSignature, async () => {
@@ -71,7 +70,7 @@ function isTerminalAgentMessage(message: ChatMessage): boolean {
 
 function isCompletedAgentMessage(message: ChatMessage): boolean {
   if (message.role !== "agent") return false;
-  return [message.status, message.runnerTrace?.status].some((value) => String(value ?? "").trim().toLowerCase().replace(/_/gu, "-") === "completed");
+  return String(message.status ?? "").trim().toLowerCase().replace(/_/gu, "-") === "completed";
 }
 
 function hasSealedCompletedText(message: ChatMessage): boolean {
@@ -238,10 +237,7 @@ function traceAutoExpanded(message: ChatMessage): boolean | null {
 }
 
 function traceForDisplay(message: ChatMessage): ChatMessage["runnerTrace"] {
-  if (!message.runnerTrace) return message.runnerTrace;
-  const status = firstNonEmptyString(message.status, message.runnerTrace.status);
-  if (!status || status === message.runnerTrace.status) return message.runnerTrace;
-  return { ...message.runnerTrace, status };
+  return message.runnerTrace;
 }
 
 function messageDurationMeta(message: ChatMessage): { text: string; label: string } | null {

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

@@ -582,8 +582,8 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     const turn = turnStatusAuthority.value[id];
     if (turn?.terminal === true || isTerminalMessageStatus(turn?.status)) return false;
     const message = [...messages.value].reverse().find((item) => firstNonEmptyString(item.traceId, item.runnerTrace?.traceId) === id) ?? null;
-    if (isTerminalMessageStatus(message?.status) || isTerminalMessageStatus(message?.runnerTrace?.status)) return false;
-    return isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status) || isTraceActiveStatus(message?.runnerTrace?.status);
+    if (isTerminalMessageStatus(message?.status)) return false;
+    return isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status);
   }
 
   async function refreshSessionMessageProjectionPage(sessionId: string | null | undefined): Promise<void> {
@@ -665,7 +665,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
 
   function messageHasActiveTurnStatusEvidence(message: ChatMessage, traceId: string): boolean {
     if (currentRequest.value?.traceId === traceId) return true;
-    return isTraceActiveStatus(message.status) || isTraceActiveStatus(message.runnerTrace?.status);
+    return isTraceActiveStatus(message.status);
   }
 
   function traceNeedsTurnStatusRefresh(traceId: string | null | undefined, source: ChatMessage[] = messages.value): boolean {
@@ -674,8 +674,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     const turn = turnStatusAuthority.value[id];
     if (turn?.terminal === true || isTerminalMessageStatus(turn?.status)) return false;
     const message = [...source].reverse().find((item) => firstNonEmptyString(item.traceId, item.runnerTrace?.traceId) === id) ?? null;
-    if (isTerminalMessageStatus(message?.status) || isTerminalMessageStatus(message?.runnerTrace?.status)) return false;
-    if (message?.runnerTrace && traceSnapshotHasTerminalEvidence(message.runnerTrace)) return false;
+    if (isTerminalMessageStatus(message?.status)) return false;
     return true;
   }
 
@@ -986,12 +985,12 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
     markWorkbenchTraceEventsReceived({ traceId, events, transport: "rest_gap" });
     updateSessionMessages(ownerSessionId, (source) => source.map((message) => {
       if (!messageMatchesTraceAuthority(message, traceId, authoritySessionId, ownerSessionId)) return message;
-      const displayStatus = firstNonEmptyString(message.status, message.runnerTrace?.status, result.status, result.runnerTrace?.status);
+      const traceDetailStatus = firstNonEmptyString(result.traceStatus, result.runnerTrace?.traceStatus, result.runnerTrace?.status);
       const { traceId: _resultTraceId, sessionId: _resultSessionId, threadId: _resultThreadId, ...resultTraceRest } = result.runnerTrace ?? {};
       const nextTrace = {
         ...resultTraceRest,
         traceId: optionalString(result.traceId, traceId),
-        status: displayStatus ?? undefined,
+        status: traceDetailStatus ?? undefined,
         sessionId: optionalString(authoritySessionId, message.runnerTrace?.sessionId),
         threadId: optionalString(result.threadId, message.runnerTrace?.threadId),
         events,
@@ -1422,8 +1421,8 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
 
   function messageHasSealedTerminalResult(message: ChatMessage | null): boolean {
     if (!message) return false;
-    if (isTerminalMessageStatus(message.status) || isTerminalMessageStatus(message.runnerTrace?.status)) return true;
-    return message.runnerTrace ? traceSnapshotHasTerminalEvidence(message.runnerTrace) : false;
+    if (!isTerminalMessageStatus(message.status)) return false;
+    return Boolean(firstNonEmptyString(message.text, finalResponseText((message as Record<string, unknown>).finalResponse)));
   }
 
   function failTrace(traceId: string, message: string): void {
@@ -1499,7 +1498,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
 
   function isRunningAgentMessageForTiming(message: ChatMessage): boolean {
     if (message.role !== "agent") return false;
-    const status = normalizedStatusText(firstNonEmptyString(message.status, message.runnerTrace?.status));
+    const status = normalizedStatusText(message.status);
     if (status && isTerminalMessageStatus(status)) return false;
     return status === null || isTraceActiveStatus(status) || message.traceAutoLifecycle === "running";
   }
@@ -1713,7 +1712,7 @@ function activeTraceIdFromMessages(messages: ChatMessage[], turnStatusAuthority:
   for (const message of [...messages].reverse()) {
     if (message.role !== "agent") continue;
     if (messageHasCompletedFinalResponse(message)) continue;
-    if (isTerminalMessageStatus(message.status) || isTerminalMessageStatus(message.runnerTrace?.status)) continue;
+    if (isTerminalMessageStatus(message.status)) continue;
     const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);
     if (!traceId) continue;
     const turn = turnStatusAuthority[traceId];
@@ -2095,7 +2094,7 @@ function traceHasCompletedFinalResponse(traceId: string | null | undefined, sour
 
 function messageHasCompletedFinalResponse(message: ChatMessage | null | undefined): boolean {
   if (!message || message.role !== "agent") return false;
-  if (normalizedStatusText(message.status) !== "completed" && normalizedStatusText(message.runnerTrace?.status) !== "completed") return false;
+  if (normalizedStatusText(message.status) !== "completed") return false;
   return Boolean(firstNonEmptyString(message.text, finalResponseText((message as Record<string, unknown>).finalResponse)));
 }