fix(workbench): prioritize terminal turn sealing

web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts

@@ -138,10 +138,20 @@ test("workbench active terminal paths seal final response from turn authority",
   const realtimeTurnBlock = source.slice(source.indexOf("function applyRealtimeTurnSnapshot"), source.indexOf("async function refreshTerminalTraceFromRest"));
   const terminalRestBlock = source.slice(source.indexOf("async function refreshTerminalTraceFromRest"), source.indexOf("function installRealtimeVisibilityHandler"));
   const completeBlock = source.slice(source.indexOf("function completeTrace"), source.indexOf("async function hydrateTerminalMessageDiagnostics"));
+  const loadBlock = source.slice(source.indexOf("async function loadWorkbenchSession"), source.indexOf("async function sealRestoredActiveTurnMessages"));
+  const restoreSealBlock = source.slice(source.indexOf("async function sealRestoredActiveTurnMessages"), source.indexOf("function reattachRestoredActiveTrace"));
 
   assert.match(projectBlock, /terminalAuthorityMessageFromTurnResult\(result\)/u);
+  assert.match(projectBlock, /type:\s*"message\.upsert"/u);
   assert.match(projectBlock, /\.\.\.terminalPatch/u);
+  assert.match(source, /createKeyedSingleflight<ApiResult<AgentChatResultResponse>>/u);
+  assert.match(source, /traceHydrationSingleflight\.run/u);
   assert.match(realtimeTurnBlock, /!terminalTurn[\s\S]*refreshRealtimeSessionMessages/u);
   assert.match(terminalRestBlock, /traceProjectionIsTerminalSealed\(id, ownerMessagesBeforeProjection\)[\s\S]*return/u);
   assert.match(completeBlock, /projectTurnAuthorityToMessages\(traceId, result, "complete-trace"\)/u);
+  assert.match(completeBlock, /traceProjectionIsTerminalSealed\(traceId, serverState\.value\.messagesBySessionId\[ownerSessionId\] \?\? \[\]\)[\s\S]*scheduleSessionListRefresh/u);
+  assert.match(loadBlock, /const messageLimit = sessionMessageProjectionWindowLimit\(\);/u);
+  assert.doesNotMatch(loadBlock, /limit:\s*100/u);
+  assert.doesNotMatch(restoreSealBlock, /Promise\.all/u);
+  assert.doesNotMatch(restoreSealBlock, /force:\s*true/u);
 });

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

@@ -78,6 +78,22 @@ test("server state reducer seals completed message when final response exists",
   assert.equal(selectSessionStatusAuthority(state)[sessionId]?.status, "completed");
 });
 
+test("server state reducer upserts terminal authority when the session page has no agent row", () => {
+  const sessionId = "ses_state_terminal_upsert_missing_agent";
+  const traceId = "trc_state_terminal_upsert_missing_agent";
+  let state = createWorkbenchServerState();
+  state = reduceWorkbenchServerState(state, { type: "session.detail", session: sessionRecord({ sessionId, status: "running", messages: [] }) });
+  state = reduceWorkbenchServerState(state, { type: "session.messages", sessionId, messages: [userMessage({ id: "msg_state_terminal_upsert_user", text: "run it", sessionId, traceId })] });
+  state = reduceWorkbenchServerState(state, { type: "message.upsert", sessionId, message: agentMessage({ id: "msg_state_terminal_upsert_agent", status: "completed", text: "final answer from turn authority", finalResponse: { text: "final answer from turn authority" }, sessionId, traceId }) });
+
+  const messages = selectActiveMessages(state, sessionId);
+  assert.equal(messages.length, 2);
+  const terminal = messages.find((message) => message.role === "agent" && message.traceId === traceId);
+  assert.ok(terminal);
+  assert.equal(terminal.status, "completed");
+  assert.equal(terminal.text, "final answer from turn authority");
+});
+
 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

@@ -18,6 +18,7 @@ export type WorkbenchServerAction =
   | { 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: "message.upsert"; sessionId: string | null | undefined; message: ChatMessage | null | undefined }
   | { type: "session.status"; session: SessionStatusAuthority }
   | { type: "session.forget"; sessionId: string }
   | { type: "session.cleanupDropped"; keepSessionIds: string[] }
@@ -46,6 +47,8 @@ export function reduceWorkbenchServerState(state: WorkbenchServerState, action:
       return reduceSessionMessages(state, action.sessionId, action.messages);
     case "message.snapshot":
       return reduceMessageSnapshot(state, action.sessionId, action.message);
+    case "message.upsert":
+      return reduceMessageUpsert(state, action.sessionId, action.message);
     case "session.status":
       return { ...state, sessionStatusById: { ...state.sessionStatusById, [action.session.sessionId]: mergeSessionStatusAuthority(state.sessionStatusById[action.session.sessionId], action.session) } };
     case "session.forget":
@@ -188,6 +191,23 @@ function reduceMessageSnapshot(state: WorkbenchServerState, sessionId: string |
   return reduceSessionMessages(state, id, merged);
 }
 
+function reduceMessageUpsert(state: WorkbenchServerState, sessionId: string | null | undefined, message: ChatMessage | null | undefined): WorkbenchServerState {
+  const id = sessionId || message?.sessionId;
+  if (!id || !message) return state;
+  const existingSession = state.sessionsById[id];
+  const existing = state.messagesBySessionId[id] ?? existingSession?.messages ?? [];
+  const index = existing.findIndex((item) => messageMatchesSnapshot(item, message));
+  const mergedMessages = index >= 0
+    ? existing.map((item, itemIndex) => itemIndex === index ? mergeMessageSnapshot(item, message) : item)
+    : [...existing, message];
+  const session = existingSession ? { ...existingSession, messages: mergedMessages, messageCount: mergedMessages.length } : null;
+  return {
+    ...state,
+    sessionsById: session ? { ...state.sessionsById, [id]: session } : state.sessionsById,
+    messagesBySessionId: { ...state.messagesBySessionId, [id]: mergedMessages }
+  };
+}
+
 function messageMatchesSnapshot(existing: ChatMessage, incoming: ChatMessage): boolean {
   const existingId = existing.messageId ?? existing.id;
   const incomingId = incoming.messageId ?? incoming.id;

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

@@ -5,6 +5,7 @@ import { computed, nextTick, ref } from "vue";
 import { defineStore } from "pinia";
 import { api } from "@/api";
 import { workbenchRuntimePolicy } from "@/config/workbench-runtime-policy";
+import { createKeyedSingleflight } from "@/utils/scheduler/keyed-singleflight";
 import { createWorkbenchHealthProbeCache } from "@/utils/workbench-health";
 import { agentErrorFromProjection, normalizeApiErrorRecord, normalizeErrorDiagnostic, normalizeProjectionDiagnostic, projectionDiagnosticFromApiFailure, projectionDiagnosticFromFailure } from "@/utils/workbench-error-runtime";
 import { readWorkbenchJson, readWorkbenchNumber, readWorkbenchString, removeWorkbenchStorageKey, writeWorkbenchJson, writeWorkbenchString } from "@/utils/workbench-storage-runtime";
@@ -88,6 +89,8 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   const workbenchColadaReducer = useWorkbenchColadaReducer();
   const workbenchColadaQueries = useWorkbenchColadaQueries();
   const workbenchColadaMutations = useWorkbenchColadaMutations();
+  const turnStatusReadSingleflight = createKeyedSingleflight<ApiResult<AgentChatResultResponse>>();
+  const traceHydrationSingleflight = createKeyedSingleflight<void>();
   const providerProfile = ref<ProviderProfile>(readString("hwlab.workbench.providerProfile.v1", "codex"));
   const providerOptions = ref<ProviderProfileOption[]>(defaultProviderProfileOptions(providerProfile.value));
   const recentDrafts = ref<DraftEntry[]>(readRecentDrafts());
@@ -445,6 +448,14 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     return [...source].reverse().find((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === id) ?? null;
   }
 
+  function traceTerminalBodyIsVisible(traceId: string | null | undefined, sessionId: string | null | undefined = null): boolean {
+    const id = firstNonEmptyString(traceId);
+    if (!id) return false;
+    const ownerSessionId = normalizeWorkbenchSessionId(sessionId) ?? traceOwnerSessionId(id, null);
+    const source = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value;
+    return traceProjectionIsTerminalSealed(id, source);
+  }
+
   function appendActiveMessages(...items: ChatMessage[]): void {
     replaceActiveMessages([...messages.value, ...items]);
   }
@@ -471,7 +482,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
 
   function fetchWorkbenchTurnStatus(traceId: string, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId), options: { force?: boolean } = {}): Promise<ApiResult<AgentChatResultResponse>> {
     const activitySource = useActivityTimeout ? () => activityRef.value : null;
-    return workbenchColadaQueries.fetchTurn(traceId, { timeoutMs: 8000, activityRef: activitySource, minIntervalMs: runtimePolicy.workbenchTurnStatusMinRefreshMs, force: options.force });
+    return turnStatusReadSingleflight.run(traceId, () => workbenchColadaQueries.fetchTurn(traceId, { timeoutMs: 8000, activityRef: activitySource, minIntervalMs: runtimePolicy.workbenchTurnStatusMinRefreshMs, force: options.force }), { reason: options.force ? "force-turn-status" : "turn-status" });
   }
 
   function fetchWorkbenchTraceEvents(traceId: string, afterProjectedSeq: number, useActivityTimeout = shouldUseActivityTimeoutForTrace(traceId), options: { force?: boolean } = {}): Promise<ApiResult<AgentChatResultResponse>> {
@@ -576,6 +587,10 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   async function refreshTurnStatusByTraceId(traceId: string | null | undefined, options: { force?: boolean } = {}): Promise<void> {
     const id = firstNonEmptyString(traceId);
     if (!id) return;
+    if (traceTerminalBodyIsVisible(id)) {
+      recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_turn_skip", source: "turn-status", valuesRedacted: true } });
+      return;
+    }
     const response = await fetchWorkbenchTurnStatus(id, shouldUseActivityTimeoutForTrace(id), { force: options.force });
     if (response.ok && response.data) {
       applyTurnStatusSnapshot(id, response.data);
@@ -624,23 +639,13 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     const authoritySessionId = traceResultSessionId(result);
     const ownerSessionId = traceOwnerSessionId(traceId, authoritySessionId);
     if (!ownerSessionId) return false;
-    let matched = false;
-    let sealed = false;
-    updateSessionMessages(ownerSessionId, (source) => {
-      const projected = source.map((message) => {
-        if (!messageMatchesTraceAuthority(message, traceId, authoritySessionId, ownerSessionId)) return message;
-        matched = true;
-        const wasSealed = messageHasSealedTerminalResult(message);
-        const next = projectTurnAuthorityMessage(message, result);
-        if (!wasSealed && messageHasSealedTerminalResult(next)) sealed = true;
-        return next;
-      });
-      if (matched) return projected;
-      const terminalMessage = terminalAuthorityMessageFromTurnResult(result);
-      if (!terminalMessage) return projected;
-      sealed = true;
-      return [...projected, terminalMessage];
-    });
+    const source = serverState.value.messagesBySessionId[ownerSessionId] ?? serverState.value.sessionsById[ownerSessionId]?.messages ?? [];
+    const existing = [...source].reverse().find((message) => messageMatchesTraceAuthority(message, traceId, authoritySessionId, ownerSessionId)) ?? null;
+    const next = existing ? projectTurnAuthorityMessage(existing, result) : terminalAuthorityMessageFromTurnResult(result);
+    if (!next) return false;
+    const message = next.sessionId ? next : { ...next, sessionId: ownerSessionId };
+    reduceServerState({ type: "message.upsert", sessionId: ownerSessionId, message });
+    const sealed = !messageHasSealedTerminalResult(existing) && messageHasSealedTerminalResult(message);
     if (sealed) {
       recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-authority", sessionId: ownerSessionId, traceId, outcome: "ok", diagnostic: { code: "terminal_direct_seal", reason, source: "turn-authority", valuesRedacted: true } });
     }
@@ -793,10 +798,24 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   }
 
   async function hydrateTraceEventsForMessageNow(message: ChatMessage, options: { force?: boolean } = {}): Promise<void> {
+    const traceId = message.traceId ?? message.runnerTrace?.traceId;
+    if (!traceId) return;
+    if (traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId)) {
+      recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: message.sessionId ?? null, traceId, outcome: "ok", diagnostic: { code: "terminal_low_priority_trace_skip", source: "trace-hydration", valuesRedacted: true } });
+      return;
+    }
+    await traceHydrationSingleflight.run(traceId, async () => {
+      if (traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId)) return;
+      await hydrateTraceEventsForMessagePages(message, options);
+    }, { reason: options.force ? "force-trace-hydration" : "trace-hydration" });
+  }
+
+  async function hydrateTraceEventsForMessagePages(message: ChatMessage, options: { force?: boolean } = {}): Promise<void> {
     const traceId = message.traceId ?? message.runnerTrace?.traceId;
     if (!traceId) return;
     let afterProjectedSeq = traceHydrationProjectedSeq(message.runnerTrace);
     for (let page = 0; page < runtimePolicy.traceHydrationMaxPages; page += 1) {
+      if (traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId)) return;
       const result = await fetchTraceHydrationPage(traceId, afterProjectedSeq, { force: options.force });
       if (!result.ok || !result.data) {
         if (shouldSuppressTransientWorkbenchReadFailure(result)) return;
@@ -1037,6 +1056,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   async function refreshActiveTraceFromRest(traceId: string, reason: string): Promise<void> {
     const id = firstNonEmptyString(traceId);
     if (!id) return;
+    if (traceTerminalBodyIsVisible(id)) {
+      clearActiveTraceRestGapFill(id);
+      recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_rest_gap_skip", reason, source: "active-rest-gap", valuesRedacted: true } });
+      return;
+    }
     const ownerBefore = traceOwnerSessionId(id, null);
     if (ownerBefore === activeSessionId.value) recordActivity(reason);
     await refreshTurnStatusByTraceId(id);
@@ -1214,6 +1238,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   async function refreshTerminalTraceFromRest(traceId: string, reason: string): Promise<void> {
     const id = firstNonEmptyString(traceId);
     if (!id) return;
+    if (traceTerminalBodyIsVisible(id)) {
+      clearActiveTraceRestGapFill(id);
+      recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_terminal_rest_skip", reason, source: "terminal-rest", valuesRedacted: true } });
+      return;
+    }
     const ownerBefore = traceOwnerSessionId(id, null);
     if (ownerBefore === activeSessionId.value) recordActivity(reason);
     await refreshTurnStatusByTraceId(id);
@@ -1305,7 +1334,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
       void refreshRealtimeSessionFromRest(ownerSessionId, `realtime-trace-snapshot:${traceId}`);
     }
     markWorkbenchTraceProjected(traceId);
-    scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListRealtimeRefreshDelayMs);
+    if (!traceProjectionIsTerminalSealed(traceId, serverState.value.messagesBySessionId[ownerSessionId] ?? [])) scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListRealtimeRefreshDelayMs);
   }
 
   async function refreshRealtimeSessionFromRest(sessionId: string, reason: string): Promise<void> {
@@ -1336,7 +1365,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
       void clearActiveTrace(traceId, "trace-terminal");
       restartRealtime("trace-terminal");
     }
-    scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListTerminalRefreshDelayMs);
+    if (!traceProjectionIsTerminalSealed(traceId, serverState.value.messagesBySessionId[ownerSessionId] ?? [])) scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListTerminalRefreshDelayMs);
   }
 
   async function hydrateTerminalMessageDiagnostics(): Promise<void> {
@@ -1649,13 +1678,14 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     const requestId = normalizeWorkbenchSessionRouteId(sessionId);
     if (!requestId) return null;
     const normalizedRequestId = normalizeWorkbenchSessionId(requestId);
-    const eagerMessages = normalizedRequestId ? workbenchColadaQueries.fetchSessionMessages(normalizedRequestId, { limit: 100, force: true, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs }) : null;
+    const messageLimit = sessionMessageProjectionWindowLimit();
+    const eagerMessages = normalizedRequestId ? workbenchColadaQueries.fetchSessionMessages(normalizedRequestId, { limit: messageLimit, force: true, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs }) : null;
     const detail = await workbenchColadaQueries.fetchSession(requestId, { force: true, minIntervalMs: runtimePolicy.workbenchSessionDetailMinRefreshMs });
     if (!detail.ok) return null;
     const detailSession = sessionFromWorkbenchSession(detail.data?.session);
     const id = detailSession?.sessionId ?? normalizedRequestId ?? seed?.sessionId;
     if (!id) return null;
-    const messages = eagerMessages && id === normalizedRequestId ? await eagerMessages : await workbenchColadaQueries.fetchSessionMessages(id, { limit: 100, force: true, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs });
+    const messages = eagerMessages && id === normalizedRequestId ? await eagerMessages : await workbenchColadaQueries.fetchSessionMessages(id, { limit: messageLimit, force: true, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs });
     const base = detailSession ?? seed;
     if (!base) return null;
     const page = messages.ok ? messages.data : null;
@@ -1664,17 +1694,17 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   }
 
   async function sealRestoredActiveTurnMessages(source: ChatMessage[]): Promise<ChatMessage[]> {
-    const targets = source.filter(messageNeedsRestoredTurnSeal).slice(-3);
+    const targets = source.filter(messageNeedsRestoredTurnSeal).slice(-1);
     if (targets.length === 0) return source;
     const patches = new Map<string, Partial<ChatMessage>>();
-    await Promise.all(targets.map(async (message) => {
+    for (const message of targets) {
       const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);
-      if (!traceId) return;
-      const response = await fetchWorkbenchTurnStatus(traceId, false, { force: true });
-      if (!response.ok || !response.data) return;
+      if (!traceId || traceProjectionIsTerminalSealed(traceId, source)) continue;
+      const response = await fetchWorkbenchTurnStatus(traceId, false);
+      if (!response.ok || !response.data) continue;
       const patch = terminalMessagePatchFromTurnResult(message, response.data);
       if (patch) patches.set(traceId, patch);
-    }));
+    }
     if (patches.size === 0) return source;
     return source.map((message) => {
       const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);