From e15756d4a804b827564db1696ebf8be9a4e415bd Mon Sep 17 00:00:00 2001 From: root Date: Thu, 2 Jul 2026 14:46:06 +0000 Subject: [PATCH] fix(workbench): prioritize terminal turn sealing --- ...rkbench-message-projection-runtime.test.ts | 10 +++ .../src/stores/workbench-server-state.test.ts | 16 ++++ .../src/stores/workbench-server-state.ts | 20 +++++ web/hwlab-cloud-web/src/stores/workbench.ts | 86 +++++++++++++------ 4 files changed, 104 insertions(+), 28 deletions(-) diff --git a/web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts b/web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts index 76a6944b..dd4c785b 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-message-projection-runtime.test.ts +++ b/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>/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); }); 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 c3e4753f..c8bde864 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 @@ -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"; 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 3daf973b..db93fe85 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-server-state.ts +++ b/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; diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 1e800fd4..2df1a32b 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/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>(); + const traceHydrationSingleflight = createKeyedSingleflight(); const providerProfile = ref(readString("hwlab.workbench.providerProfile.v1", "codex")); const providerOptions = ref(defaultProviderProfileOptions(providerProfile.value)); const recentDrafts = ref(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> { 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> { @@ -576,6 +587,10 @@ export const useWorkbenchStore = defineStore("workbench", () => { async function refreshTurnStatusByTraceId(traceId: string | null | undefined, options: { force?: boolean } = {}): Promise { 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 { + 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 { 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 { 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 { 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 { @@ -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 { @@ -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 { - const targets = source.filter(messageNeedsRestoredTurnSeal).slice(-3); + const targets = source.filter(messageNeedsRestoredTurnSeal).slice(-1); if (targets.length === 0) return source; const patches = new Map>(); - 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);