From 91b60711e47a4742a1ddf2d151e7c32b89073a5d Mon Sep 17 00:00:00 2001 From: root Date: Wed, 8 Jul 2026 20:27:59 +0200 Subject: [PATCH] fix(workbench): route automatic recovery through sync replay --- ...rkbench-message-projection-runtime.test.ts | 54 ++-- web/hwlab-cloud-web/src/stores/workbench.ts | 234 +++++++----------- 2 files changed, 123 insertions(+), 165 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 c14eb8d0..5c7544e4 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 @@ -136,37 +136,38 @@ test("workbench active terminal paths seal final response from turn authority", const source = fs.readFileSync(path.join(storeDir, "workbench.ts"), "utf8"); const projectBlock = source.slice(source.indexOf("function projectTurnAuthorityToMessages"), source.indexOf("async function submitMessage")); const realtimeTurnBlock = source.slice(source.indexOf("function applyRealtimeTurnSnapshot"), source.indexOf("function scheduleRealtimeTurnProjection")); - const realtimeTurnProjectionBlock = source.slice(source.indexOf("function flushRealtimeTurnProjection"), source.indexOf("async function refreshTerminalTraceFromRest")); - const terminalRestBlock = source.slice(source.indexOf("async function refreshTerminalTraceFromRest"), source.indexOf("function installRealtimeVisibilityHandler")); + const realtimeTurnProjectionBlock = source.slice(source.indexOf("function flushRealtimeTurnProjection"), source.indexOf("async function refreshTerminalTraceFromSyncReplay")); + const terminalSyncBlock = source.slice(source.indexOf("async function refreshTerminalTraceFromSyncReplay"), source.indexOf("function installRealtimeVisibilityHandler")); const completeBlock = source.slice(source.indexOf("function completeTrace"), source.indexOf("async function hydrateTerminalMessageDiagnostics")); - const realtimeSessionDetailBlock = source.slice(source.indexOf("async function refreshRealtimeSessionFromRest"), source.indexOf("function completeTrace")); + const realtimeSessionSyncBlock = source.slice(source.indexOf("async function refreshRealtimeSessionFromSyncReplay"), source.indexOf("function completeTrace")); const sessionDetailReadBlock = source.slice(source.indexOf("function fetchSessionDetailPage"), source.indexOf("function sessionMessageProjectionWindowLimit")); - const traceHydrationBlock = source.slice(source.indexOf("async function hydrateTraceEventsForMessagePages"), source.indexOf("async function fetchTraceHydrationPage")); + const traceHydrationBlock = source.slice(source.indexOf("async function hydrateTraceEventsForExplicitDetailPages"), source.indexOf("async function fetchTraceHydrationPage")); 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")); + const activeSyncReplayBlock = source.slice(source.indexOf("async function refreshActiveTraceFromSyncReplay"), source.indexOf("function stopRealtime")); 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.doesNotMatch(source, new RegExp(["refresh", "RealtimeSessionMessages"].join(""), "u")); assert.match(realtimeTurnBlock, /rememberTurnStatus\(traceId, result\)[\s\S]*scheduleRealtimeTurnProjection\(\{ traceId, result, terminalTurn \}\)/u); - assert.doesNotMatch(realtimeTurnBlock, /applyTurnStatusSnapshot\(|refreshTerminalTraceFromRest\(/u); + assert.doesNotMatch(realtimeTurnBlock, new RegExp(`applyTurnStatusSnapshot\\(|${["refresh", "TerminalTraceFromRest"].join("")}\\(`, "u")); assert.match(realtimeTurnProjectionBlock, /syncTurnStatusToMessage\(next\.traceId, next\.result\)/u); - assert.match(realtimeTurnProjectionBlock, /next\.terminalTurn[\s\S]*refreshTerminalTraceFromRest\(next\.traceId, "realtime-turn-snapshot"\)/u); + assert.match(realtimeTurnProjectionBlock, /next\.terminalTurn[\s\S]*refreshTerminalTraceFromSyncReplay\(next\.traceId, "realtime-turn-snapshot"\)/u); assert.match(realtimeTurnProjectionBlock, /workbench_realtime_turn_projection_budget/u); - assert.match(terminalRestBlock, /traceProjectionIsTerminalSealed\(id, ownerMessagesBeforeProjection\)[\s\S]*return/u); + assert.match(terminalSyncBlock, /refreshWorkbenchSyncReplay\(ownerSessionId, id, null, `terminal-sync-replay:\$\{reason\}`\)/u); + assert.doesNotMatch(terminalSyncBlock, /refreshTurnStatusByTraceId|refreshMessageProjectionForTrace|fetchSessionMessagesPage|hydrateTraceEventsForMessage/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(realtimeSessionDetailBlock, /traceIdFromRealtimeRefreshReason\(reason\)/u); - assert.match(realtimeSessionDetailBlock, /traceTerminalBodyIsVisible\(traceId, id\)[\s\S]*terminal_low_priority_session_detail_skip[\s\S]*return/u); - assert.match(realtimeSessionDetailBlock, /hydrateTurnStatusAuthority\(activeMessagesBeforeDetail, \{ traceId, limit: 1, reason: "realtime-session-detail:turn-authority" \}\)/u); - assert.match(realtimeSessionDetailBlock, /sessionDetailAutoReadDecision\(\{ traceId,[\s\S]*workbenchSessionDetailMinRefreshMs \}\)/u); - assert.match(realtimeSessionDetailBlock, /session_detail_auto_read_skip/u); - assert.match(realtimeSessionDetailBlock, /traceTerminalBodyIsVisible\(traceId, id\)[\s\S]*terminal_low_priority_session_detail_apply_skip[\s\S]*return/u); - assert.match(realtimeSessionDetailBlock, /fetchSessionDetailPage\(id, \{ reason: `realtime-session-detail:\$\{reason\}` \}\)/u); - assert.doesNotMatch(realtimeSessionDetailBlock, /loadWorkbenchSession|applySelectedSessionDetail/u); + assert.match(completeBlock, /options\.forceRead[\s\S]*refreshMessageProjectionForTrace\(ownerSessionId, traceId, \{ force: true \}\)/u); + assert.match(completeBlock, /else scheduleActiveTraceSyncReplay\(traceId, "complete-trace-sync-replay", 0\)/u); + assert.match(realtimeSessionSyncBlock, /traceIdFromRealtimeRefreshReason\(reason\)/u); + assert.match(realtimeSessionSyncBlock, /refreshWorkbenchSyncReplay\(id, traceId, null, `realtime-session-sync:\$\{reason\}`\)/u); + assert.doesNotMatch(realtimeSessionSyncBlock, /hydrateTurnStatusAuthority|sessionDetailAutoReadDecision|fetchSessionDetailPage|fetchSessionMessagesPage|refreshMessageProjectionForTrace/u); + assert.doesNotMatch(realtimeSessionSyncBlock, /loadWorkbenchSession|applySelectedSessionDetail/u); + assert.match(activeSyncReplayBlock, /refreshWorkbenchSyncReplay\(ownerSessionId, id, null, `active-sync-replay:\$\{reason\}`\)/u); + assert.doesNotMatch(activeSyncReplayBlock, /refreshTurnStatusByTraceId|refreshMessageProjectionForTrace|fetchSessionMessagesPage|hydrateTraceEventsForMessage/u); assert.match(traceHydrationBlock, /traceAuthorityById\.value\[traceId\] \?\? message\.runnerTrace/u); assert.match(traceHydrationBlock, /traceEventsHydrationReadDecision\(traceId, afterProjectedSeq, message, options\)/u); assert.match(traceHydrationBlock, /trace_events_auto_read_skip/u); @@ -177,6 +178,23 @@ test("workbench active terminal paths seal final response from turn authority", assert.match(loadBlock, /sessionFromWorkbenchSession\(detail\.data\?\.session, \{ includeMessages: false \}\)/u); assert.match(loadBlock, /const fallbackMessages = seed\?\.sessionId === id \? seed\.messages \?\? \[\] : \[\];/u); assert.doesNotMatch(loadBlock, /limit:\s*100/u); - assert.doesNotMatch(restoreSealBlock, /Promise\.all/u); + assert.match(restoreSealBlock, /workbench_restored_turn_auto_read_disabled/u); + assert.doesNotMatch(restoreSealBlock, /fetchWorkbenchTurnStatus|Promise\.all/u); assert.doesNotMatch(restoreSealBlock, /force:\s*true/u); }); + +test("workbench automatic live repair does not call legacy REST fan-out", () => { + const source = fs.readFileSync(path.join(storeDir, "workbench.ts"), "utf8"); + const automaticBlocks = [ + source.slice(source.indexOf("async function submitMessage"), source.indexOf("async function cancelAgentMessage")), + source.slice(source.indexOf("function reattachTrace"), source.indexOf("function restartRealtime")), + source.slice(source.indexOf("async function refreshActiveTraceFromSyncReplay"), source.indexOf("function stopRealtime")), + source.slice(source.indexOf("function flushRealtimeTurnProjection"), source.indexOf("function installRealtimeVisibilityHandler")), + source.slice(source.indexOf("async function refreshRealtimeSessionFromSyncReplay"), source.indexOf("function completeTrace")), + source.slice(source.indexOf("async function sealRestoredActiveTurnMessages"), source.indexOf("function reattachRestoredActiveTrace")) + ].join("\n"); + + assert.match(automaticBlocks, /refreshWorkbenchSyncReplay/u); + assert.doesNotMatch(automaticBlocks, /fetchWorkbenchTurnStatus|fetchWorkbenchTraceEvents|fetchSessionMessagesPage|fetchSessionDetailPage/u); + assert.doesNotMatch(automaticBlocks, /refreshTurnStatusByTraceId|refreshMessageProjectionForTrace|hydrateTraceEventsForMessage\(/u); +}); diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index d70f3ae9..3a3965b8 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -71,7 +71,7 @@ import { workbenchSyncReplayDiagnostic, workbenchSyncReplayEvents } from "./work import { useWorkbenchColadaMutations } from "./workbench-colada-mutations"; import { useWorkbenchColadaQueries } from "./workbench-colada-queries"; import { useWorkbenchColadaReducer } from "./workbench-colada-reducer"; -import { projectionMergeCommitSummary, sessionDetailAutoReadDecision, traceEventsAutoReadDecision, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey, workbenchTraceEventsReadKey, type TraceEventsReadRangeRecord } from "./workbench-session-messages-read-budget"; +import { projectionMergeCommitSummary, traceEventsAutoReadDecision, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey, workbenchTraceEventsReadKey, type TraceEventsReadRangeRecord } from "./workbench-session-messages-read-budget"; import { terminalSealResultWithoutTraceEvents, traceHydrationProjectedSeq, traceNextProjectedSeq } from "./workbench-trace-hydration"; const WORKBENCH_SESSION_PROJECTION_SIGNAL_CHANNEL = "hwlab.workbench.sessionProjection.v1"; @@ -115,7 +115,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { const sessionMessagesReadSingleflight = createKeyedSingleflight>(); const sessionDetailReadSingleflight = createKeyedSingleflight>(); const traceEventsReadRanges = new Map(); - const sessionDetailAutoReadAtMs = new Map(); const realtimeTurnProjectionQueue = new Map(); let realtimeTurnProjectionScheduled = false; const providerProfile = ref(readString("hwlab.workbench.providerProfile.v1", "codex")); @@ -528,25 +527,6 @@ export const useWorkbenchStore = defineStore("workbench", () => { return isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status); } - async function refreshSessionMessageProjectionPage(sessionId: string | null | undefined, options: { force?: boolean } = {}): Promise { - const id = normalizeWorkbenchSessionId(sessionId); - if (!id) return; - const response = await fetchSessionMessagesPage(id, { limit: sessionMessageProjectionWindowLimit(), reason: "session-message-page", force: options.force }); - if (!response.ok || !response.data) return; - const pageMessages = Array.isArray(response.data.messages) ? response.data.messages.map((message) => normalizeChatMessage(message as ChatMessage)) : []; - const merged = mergeMessageProjectionPage(id, pageMessages, { limit: sessionMessageProjectionWindowLimit() }); - commitMergedSessionMessages(id, pageMessages, merged, { limit: sessionMessageProjectionWindowLimit(), reason: "session-message-page" }); - await hydrateTurnStatusAuthority(merged, { limit: 1, reason: "session-message-page" }); - hydrateTerminalTraceGaps(merged, "session-message-page"); - } - - async function refreshRealtimeSessionMessages(sessionId: string | null | undefined, reason: string, options: { force?: boolean } = {}): Promise { - const id = normalizeWorkbenchSessionId(sessionId); - if (!id || id !== activeSessionId.value) return; - recordActivity(reason); - await refreshSessionMessageProjectionPage(id, { force: options.force }); - } - async function refreshMessageProjectionForTrace(sessionId: string | null | undefined, traceId: string, options: { force?: boolean } = {}): Promise { const id = normalizeWorkbenchSessionId(sessionId); if (!id) return; @@ -636,13 +616,25 @@ export const useWorkbenchStore = defineStore("workbench", () => { return existing.find((item) => item.role === message.role && firstNonEmptyString(item.traceId, item.runnerTrace?.traceId) === traceId) ?? null; } - async function hydrateTurnStatusAuthority(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string } = {}): Promise { - for (const traceId of activeTurnStatusRefreshTraceIds(source, options)) { + async function hydrateTurnStatusAuthority(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string; force?: boolean } = {}): Promise { + const traceIds = activeTurnStatusRefreshTraceIds(source, options); + if (options.force !== true) { + if (traceIds.length > 0) { + recordWorkbenchRuntimeDiagnostic({ + module: "workbench-turn-status", + traceId: options.traceId ?? traceIds[0] ?? null, + outcome: "ok", + diagnostic: { code: "workbench_turn_status_auto_read_disabled", reason: options.reason ?? "auto-turn-status-hydrate", source: "sync-replay-authority", requestedCount: traceIds.length, valuesRedacted: true } + }); + } + return; + } + for (const traceId of traceIds) { await refreshTurnStatusByTraceId(traceId); } } - function activeTurnStatusRefreshTraceIds(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string } = {}): string[] { + function activeTurnStatusRefreshTraceIds(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string; force?: boolean } = {}): string[] { const requestedTraceId = firstNonEmptyString(options.traceId, currentRequest.value?.traceId); const traceIds = selectActiveTurnStatusRefreshTraceIds({ messages: source, currentRequestTraceId: requestedTraceId, turnStatusAuthority: turnStatusAuthority.value, limit: options.limit ?? 1 }); recordWorkbenchRuntimeDiagnostic({ @@ -664,6 +656,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 (options.force !== true) { + recordWorkbenchRuntimeDiagnostic({ module: "workbench-turn-status", traceId: id, outcome: "ok", diagnostic: { code: "workbench_turn_status_auto_read_disabled", source: "sync-replay-authority", valuesRedacted: true } }); + 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; @@ -829,7 +825,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { completeTrace(canonicalTraceId, response.data as AgentChatResultResponse); return true; } - scheduleActiveTraceRestGapFill(canonicalTraceId, "submit-rest-gap"); + scheduleActiveTraceSyncReplay(canonicalTraceId, "submit-sync-replay"); restartRealtime("submit"); return true; } @@ -846,7 +842,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { recordActivity("cancel:accepted"); if (message.status === "running") chatPending.value = false; currentRequest.value = null; - clearActiveTraceRestGapFill(traceId); + clearActiveTraceSyncReplay(traceId); void clearActiveTrace(traceId, "cancel-agent-message"); scheduleSessionListRefresh(sessionId, runtimePolicy.sessionListTerminalRefreshDelayMs); } @@ -872,6 +868,10 @@ export const useWorkbenchStore = defineStore("workbench", () => { async function hydrateTraceEventsForMessage(message: ChatMessage, options: { force?: boolean } = {}): Promise { const traceId = message.traceId ?? message.runnerTrace?.traceId; if (!traceId) return; + if (options.force !== true) { + recordWorkbenchRuntimeDiagnostic({ module: "workbench-trace-events-read", sessionId: message.sessionId ?? message.runnerTrace?.sessionId ?? null, traceId, outcome: "ok", diagnostic: { code: "trace_events_auto_read_disabled", reason: "sync-replay-authority", source: "trace-hydration", valuesRedacted: true } }); + return; + } await hydrateTraceEventsForMessageNow(message, options); } @@ -884,11 +884,11 @@ export const useWorkbenchStore = defineStore("workbench", () => { } await traceHydrationSingleflight.run(traceId, async () => { if (traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId)) return; - await hydrateTraceEventsForMessagePages(message, options); + await hydrateTraceEventsForExplicitDetailPages(message, options); }, { reason: options.force ? "force-trace-hydration" : "trace-hydration" }); } - async function hydrateTraceEventsForMessagePages(message: ChatMessage, options: { force?: boolean } = {}): Promise { + async function hydrateTraceEventsForExplicitDetailPages(message: ChatMessage, options: { force?: boolean } = {}): Promise { const traceId = message.traceId ?? message.runnerTrace?.traceId; if (!traceId) return; let afterProjectedSeq = traceHydrationProjectedSeq(traceAuthorityById.value[traceId] ?? message.runnerTrace); @@ -953,8 +953,16 @@ export const useWorkbenchStore = defineStore("workbench", () => { void reason; } - async function hydrateTraceEvents(source: ChatMessage[] = messages.value): Promise { - for (const message of traceHydrationCandidates(source)) void hydrateTraceEventsForMessage(message); + async function hydrateTraceEvents(source: ChatMessage[] = messages.value, options: { force?: boolean } = {}): Promise { + const candidates = traceHydrationCandidates(source); + if (options.force !== true) { + if (candidates.length > 0) { + const traceId = firstNonEmptyString(candidates[0]?.traceId, candidates[0]?.runnerTrace?.traceId); + recordWorkbenchRuntimeDiagnostic({ module: "workbench-trace-events-read", sessionId: candidates[0]?.sessionId ?? candidates[0]?.runnerTrace?.sessionId ?? null, traceId: traceId ?? null, outcome: "ok", diagnostic: { code: "trace_events_auto_read_disabled", reason: "sync-replay-authority", source: "trace-hydration", candidateCount: candidates.length, valuesRedacted: true } }); + } + return; + } + for (const message of candidates) void hydrateTraceEventsForMessage(message, { force: true }); } function traceHydrationCandidates(source: ChatMessage[]): ChatMessage[] { @@ -1042,7 +1050,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { })); markWorkbenchTraceProjected(traceId); if (traceResultHasTerminalEvidence(result) && !turnResultIsTerminalForMerge(result)) { - void refreshTerminalTraceFromRest(traceId, "trace-hydration-terminal-evidence"); + void refreshTerminalTraceFromSyncReplay(traceId, "trace-hydration-terminal-evidence"); } if (turnResultIsTerminalForMerge(result)) { rememberTurnStatus(traceId, result); @@ -1092,24 +1100,10 @@ export const useWorkbenchStore = defineStore("workbench", () => { const initial: AgentChatResponse = { status: "running", traceId, turnUrl: `/v1/workbench/turns/${encodeURIComponent(traceId)}` }; if (!messages.value.some((message) => message.traceId === traceId)) appendActiveMessages(makeMessage("agent", "", "running", { traceId, sessionId: selectedSessionId.value ?? undefined, threadId: selectedThreadId.value ?? undefined, title: "Code Agent", traceAutoLifecycle: "running" })); currentRequest.value = { traceId, sessionId: selectedSessionId.value ?? null, threadId: selectedThreadId.value ?? null, status: initial.status }; - scheduleActiveTraceRestGapFill(traceId, "reattach-rest-gap"); + scheduleActiveTraceSyncReplay(traceId, "reattach-sync-replay"); restartRealtime("reattach"); } - async function validateAndReattachTrace(traceId: string): Promise { - const result = await fetchWorkbenchTurnStatus(traceId); - if (!result.ok || !result.data) { - await clearActiveTrace(traceId, result.status === 404 ? "reattach-result-not-found" : "reattach-result-unavailable"); - return; - } - applyTurnStatusSnapshot(traceId, result.data); - if (result.data.running === true || isTraceActiveStatus(result.data.status)) { - reattachTrace(traceId); - return; - } - await clearActiveTrace(traceId, "reattach-terminal-result"); - } - function restartRealtime(reason: string): void { const traceId = realtimeTraceId(); const sessionId = selectedSessionId.value ?? null; @@ -1160,58 +1154,45 @@ export const useWorkbenchStore = defineStore("workbench", () => { recordWorkbenchRuntimeDiagnostic({ module: "workbench-sync-replay", sessionId, traceId, outcome: "ok", diagnostic: workbenchSyncReplayDiagnostic(result.data, events, { reason, sinceOutboxSeq }) }); } - function scheduleActiveTraceRestGapFill(traceId: string | null | undefined, reason: string, delayMs = runtimePolicy.workbenchActiveTraceRestGapFillInitialMs): void { + function scheduleActiveTraceSyncReplay(traceId: string | null | undefined, reason: string, delayMs = runtimePolicy.workbenchActiveTraceRestGapFillInitialMs): void { const id = firstNonEmptyString(traceId); if (!id) return; if (typeof window === "undefined") { - void refreshActiveTraceFromRest(id, reason); + void refreshActiveTraceFromSyncReplay(id, reason); return; } - window.setTimeout(() => { void refreshActiveTraceFromRest(id, reason); }, delayMs); + window.setTimeout(() => { void refreshActiveTraceFromSyncReplay(id, reason); }, delayMs); } - function clearActiveTraceRestGapFill(traceId: string | null | undefined): void { + function clearActiveTraceSyncReplay(traceId: string | null | undefined): void { void traceId; } - async function refreshActiveTraceFromRest(traceId: string, reason: string): Promise { + async function refreshActiveTraceFromSyncReplay(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 } }); + clearActiveTraceSyncReplay(id); + recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_sync_replay_skip", reason, source: "active-sync-replay", valuesRedacted: true } }); return; } - const ownerBefore = traceOwnerSessionId(id, null); - const ownerMessagesBeforeRest = ownerBefore ? serverState.value.messagesBySessionId[ownerBefore] ?? [] : messages.value; - if (!shouldHydrateTerminalTraceFromRest({ traceId: id, messages: ownerMessagesBeforeRest, turnStatusAuthority: turnStatusAuthority.value })) { - clearActiveTraceRestGapFill(id); - recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: ownerBefore, traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_rest_gap_skip", reason, source: "active-rest-gap-preflight", valuesRedacted: true } }); - return; - } - if (ownerBefore === activeSessionId.value) recordActivity(reason); - await refreshTurnStatusByTraceId(id); - const ownerSessionId = traceOwnerSessionId(id, turnStatusAuthority.value[id]?.sessionId ?? null) ?? ownerBefore; - const ownerMessagesBeforeProjection = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value; - if (traceProjectionIsTerminalSealed(id, ownerMessagesBeforeProjection)) { - clearActiveTraceRestGapFill(id); - return; - } - if (ownerSessionId) { - scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListRealtimeRefreshDelayMs); - await refreshMessageProjectionForTrace(ownerSessionId, id); - } + const ownerSessionId = traceOwnerSessionId(id, currentRequest.value?.sessionId ?? null); + if (ownerSessionId === activeSessionId.value) recordActivity(reason); + await refreshWorkbenchSyncReplay(ownerSessionId, id, null, `active-sync-replay:${reason}`); const ownerMessages = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value; + if (traceProjectionIsTerminalSealed(id, ownerMessages)) { + clearActiveTraceSyncReplay(id); + return; + } const message = latestMessageForTrace(id, ownerMessages); const turn = turnStatusAuthority.value[id]; const terminal = messageHasTerminalResponse(message); if (terminal) { - clearActiveTraceRestGapFill(id); + clearActiveTraceSyncReplay(id); return; } - if (message) await hydrateTraceEventsForMessage(message); const stillActive = currentRequest.value?.traceId === id || isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status); - if (stillActive) scheduleActiveTraceRestGapFill(id, "active-rest-gap:repeat", runtimePolicy.workbenchActiveTraceRestGapFillRepeatMs); + if (stillActive) scheduleActiveTraceSyncReplay(id, "active-sync-replay:repeat", runtimePolicy.workbenchActiveTraceRestGapFillRepeatMs); } function stopRealtime(): void { @@ -1417,7 +1398,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const startedAt = performanceNowMs(); if (shouldApplyActiveTraceAuthority(next.traceId, traceResultSessionId(next.result))) { syncTurnStatusToMessage(next.traceId, next.result); - if (next.terminalTurn) void refreshTerminalTraceFromRest(next.traceId, "realtime-turn-snapshot"); + if (next.terminalTurn) void refreshTerminalTraceFromSyncReplay(next.traceId, "realtime-turn-snapshot"); } recordWorkbenchRuntimeDiagnostic({ module: "workbench-turn-status", @@ -1437,38 +1418,23 @@ export const useWorkbenchStore = defineStore("workbench", () => { if (realtimeTurnProjectionQueue.size > 0) scheduleRealtimeTurnProjectionFlush(); } - async function refreshTerminalTraceFromRest(traceId: string, reason: string): Promise { + async function refreshTerminalTraceFromSyncReplay(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 } }); + clearActiveTraceSyncReplay(id); + recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_terminal_sync_skip", reason, source: "terminal-sync-replay", valuesRedacted: true } }); return; } - const ownerBefore = traceOwnerSessionId(id, null); - const ownerMessagesBeforeRest = ownerBefore ? serverState.value.messagesBySessionId[ownerBefore] ?? [] : messages.value; - if (!shouldHydrateTerminalTraceFromRest({ traceId: id, messages: ownerMessagesBeforeRest, turnStatusAuthority: turnStatusAuthority.value })) { - clearActiveTraceRestGapFill(id); - recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: ownerBefore, traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_terminal_rest_skip", reason, source: "terminal-rest-preflight", valuesRedacted: true } }); - return; - } - if (ownerBefore === activeSessionId.value) recordActivity(reason); - await refreshTurnStatusByTraceId(id); - const ownerSessionId = traceOwnerSessionId(id, turnStatusAuthority.value[id]?.sessionId ?? null) ?? ownerBefore; - const ownerMessagesBeforeProjection = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value; - if (traceProjectionIsTerminalSealed(id, ownerMessagesBeforeProjection)) { - clearActiveTraceRestGapFill(id); - return; - } - if (ownerSessionId) { - scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListTerminalRefreshDelayMs); - await refreshMessageProjectionForTrace(ownerSessionId, id); - } + const ownerSessionId = traceOwnerSessionId(id, null); + if (ownerSessionId === activeSessionId.value) recordActivity(reason); + await refreshWorkbenchSyncReplay(ownerSessionId, id, null, `terminal-sync-replay:${reason}`); const ownerMessages = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value; - const message = latestMessageForTrace(id, ownerMessages); - if (!message && ownerSessionId) { - await refreshRealtimeSessionMessages(ownerSessionId, `${reason}:message-gap`); + if (traceProjectionIsTerminalSealed(id, ownerMessages)) { + clearActiveTraceSyncReplay(id); + return; } + if (ownerSessionId) scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListTerminalRefreshDelayMs); } function installRealtimeVisibilityHandler(): void { @@ -1512,7 +1478,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { if (!sessionId || sessionId !== activeSessionId.value) return; const traceId = firstNonEmptyString(record.traceId); if (traceId && messages.value.some((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === traceId)) return; - void refreshRealtimeSessionMessages(sessionId, `cross-tab-session-projection:${firstNonEmptyString(record.reason, traceId, "session") ?? "session"}`); + void refreshRealtimeSessionFromSyncReplay(sessionId, `cross-tab-session-projection:${firstNonEmptyString(record.reason, traceId, "session") ?? "session"}`); } function applyTraceSnapshot(traceId: string, snapshot: TraceSnapshot): void { @@ -1538,48 +1504,22 @@ export const useWorkbenchStore = defineStore("workbench", () => { return { ...message, ...messageTimingPatchForMerge(message, trace), ...messageStatusPatchForTerminalMerge(message, traceStatus, terminal), runnerTrace, error: clearCompletedDiagnostics ? null : error ?? message.error ?? null, projection, projectionStatus: projection?.projectionStatus ?? null, projectionHealth: projection?.projectionHealth ?? null, blocker: projection?.blocker ?? null, updatedAt: new Date().toISOString() }; })); if (!matchedMessage && ownerSessionId === activeSessionId.value) { - void refreshRealtimeSessionMessages(ownerSessionId, `realtime-trace-gap:${traceId}`); - void refreshRealtimeSessionFromRest(ownerSessionId, `realtime-trace-snapshot:${traceId}`); + void refreshRealtimeSessionFromSyncReplay(ownerSessionId, `realtime-trace-snapshot:${traceId}`); } markWorkbenchTraceProjected(traceId); if (!traceProjectionIsTerminalSealed(traceId, serverState.value.messagesBySessionId[ownerSessionId] ?? [])) scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListRealtimeRefreshDelayMs); } - async function refreshRealtimeSessionFromRest(sessionId: string, reason: string): Promise { + async function refreshRealtimeSessionFromSyncReplay(sessionId: string, reason: string): Promise { const id = normalizeWorkbenchSessionId(sessionId); if (!id || id !== activeSessionId.value) return; const traceId = traceIdFromRealtimeRefreshReason(reason); if (traceTerminalBodyIsVisible(traceId, id)) { - recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: id, traceId, outcome: "ok", diagnostic: { code: "terminal_low_priority_session_detail_skip", reason, source: "realtime-session-detail", valuesRedacted: true } }); - return; - } - const activeMessagesBeforeDetail = serverState.value.messagesBySessionId[id] ?? messages.value; - await hydrateTurnStatusAuthority(activeMessagesBeforeDetail, { traceId, limit: 1, reason: "realtime-session-detail:turn-authority" }); - if (traceTerminalBodyIsVisible(traceId, id)) { - recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: id, traceId, outcome: "ok", diagnostic: { code: "terminal_low_priority_session_detail_turn_skip", reason, source: "realtime-session-detail", valuesRedacted: true } }); - return; - } - const detailDecision = sessionDetailAutoReadDecision({ traceId, terminalBodyVisible: false, lastReadAtMs: sessionDetailAutoReadAtMs.get(id) ?? null, nowMs: Date.now(), minIntervalMs: runtimePolicy.workbenchSessionDetailMinRefreshMs }); - if (!detailDecision.read) { - recordWorkbenchRuntimeDiagnostic({ module: "workbench-session-detail-read", sessionId: id, traceId, outcome: "ok", diagnostic: { code: "session_detail_auto_read_skip", reason: detailDecision.reason, source: "realtime-session-detail", valuesRedacted: true } }); + recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: id, traceId, outcome: "ok", diagnostic: { code: "terminal_low_priority_session_sync_skip", reason, source: "realtime-session-sync", valuesRedacted: true } }); return; } recordActivity(reason); - sessionDetailAutoReadAtMs.set(id, Date.now()); - const detail = await fetchSessionDetailPage(id, { reason: `realtime-session-detail:${reason}` }); - if (!detail.ok || activeSessionId.value !== id) return; - const detailSession = sessionFromWorkbenchSession(detail.data?.session, { includeMessages: false }); - if (detailSession) { - const existingMessages = serverState.value.messagesBySessionId[id] ?? []; - rememberSessionDetail({ ...detailSession, messages: existingMessages, messageCount: detailSession.messageCount ?? existingMessages.length }); - if (traceTerminalBodyIsVisible(traceId, id)) { - recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: id, traceId, outcome: "ok", diagnostic: { code: "terminal_low_priority_session_detail_apply_skip", reason, source: "realtime-session-detail", valuesRedacted: true } }); - return; - } - } - const activeMessages = serverState.value.messagesBySessionId[id] ?? messages.value; - await hydrateTurnStatusAuthority(activeMessages, { traceId, limit: 1, reason: "realtime-session-detail" }); - if (traceId && !traceProjectionIsTerminalSealed(traceId, activeMessages)) hydrateTerminalTraceGaps(selectProjectionMessageWindow(activeMessages, { traceId, limit: traceMessageProjectionWindowLimit() }), `realtime-session-detail:${reason}`); + await refreshWorkbenchSyncReplay(id, traceId, null, `realtime-session-sync:${reason}`); } function completeTrace(traceId: string, result: AgentChatResultResponse, options: { forceRead?: boolean } = {}): void { @@ -1590,10 +1530,13 @@ export const useWorkbenchStore = defineStore("workbench", () => { projectTurnAuthorityToMessages(traceId, result, "complete-trace"); rememberTurnStatus(traceId, result); markWorkbenchTraceProjected(traceId); - clearActiveTraceRestGapFill(traceId); + clearActiveTraceSyncReplay(traceId); const ownerMessages = serverState.value.messagesBySessionId[ownerSessionId] ?? []; const terminalMessage = latestMessageForTrace(traceId, ownerMessages); - if (!traceProjectionIsTerminalSealed(traceId, ownerMessages)) void refreshMessageProjectionForTrace(ownerSessionId, traceId, { force: options.forceRead }); + if (!traceProjectionIsTerminalSealed(traceId, ownerMessages)) { + if (options.forceRead) void refreshMessageProjectionForTrace(ownerSessionId, traceId, { force: true }); + else scheduleActiveTraceSyncReplay(traceId, "complete-trace-sync-replay", 0); + } if (options.forceRead && terminalMessage && !traceProjectionIsTerminalSealed(traceId, ownerMessages)) void hydrateTraceEventsForMessage(terminalMessage, { force: true }); if (ownerSessionId === activeSessionId.value) { chatPending.value = false; @@ -1853,7 +1796,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { } async function clearActiveTrace(traceId: string, reason: string): Promise { - clearActiveTraceRestGapFill(traceId); + clearActiveTraceSyncReplay(traceId); if (currentRequest.value?.traceId === traceId) currentRequest.value = null; } @@ -1933,27 +1876,24 @@ export const useWorkbenchStore = defineStore("workbench", () => { async function sealRestoredActiveTurnMessages(source: ChatMessage[]): Promise { const targets = source.filter(messageNeedsRestoredTurnSeal).slice(-1); if (targets.length === 0) return source; - const patches = new Map>(); for (const message of targets) { const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId); 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); + recordWorkbenchRuntimeDiagnostic({ + module: "workbench-turn-status", + sessionId: message.sessionId ?? message.runnerTrace?.sessionId ?? null, + traceId, + outcome: "ok", + diagnostic: { code: "workbench_restored_turn_auto_read_disabled", reason: "load-session:restored-turn", source: "sync-replay-authority", valuesRedacted: true } + }); } - if (patches.size === 0) return source; - return source.map((message) => { - const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId); - const patch = traceId ? patches.get(traceId) : null; - return patch ? { ...message, ...patch } : message; - }); + return source; } function reattachRestoredActiveTrace(): void { void hydrateTraceEvents(messages.value); const traceId = activeTraceIdFromMessages(messages.value, turnStatusAuthority.value); - if (traceId) void validateAndReattachTrace(traceId); + if (traceId) reattachTrace(traceId); } function performanceNowMs(): number {