diff --git a/web/hwlab-cloud-web/src/api/workbench-events.test.ts b/web/hwlab-cloud-web/src/api/workbench-events.test.ts new file mode 100644 index 00000000..33273739 --- /dev/null +++ b/web/hwlab-cloud-web/src/api/workbench-events.test.ts @@ -0,0 +1,18 @@ +import assert from "node:assert/strict"; +import { test } from "bun:test"; + +import { realtimeCoalesceKey, type WorkbenchRealtimeEvent } from "./workbench-events"; + +test("turn snapshot coalescing keys by trace instead of per sequence", () => { + const first: WorkbenchRealtimeEvent = { type: "turn.snapshot", turn: { sessionId: "ses_queue", traceId: "trc_queue", status: "running" }, cursor: { traceSeq: 10 } }; + const next: WorkbenchRealtimeEvent = { type: "turn.snapshot", turn: { sessionId: "ses_queue", traceId: "trc_queue", status: "running" }, cursor: { traceSeq: 11 } }; + + assert.equal(realtimeCoalesceKey(first, "workbench.turn.snapshot"), realtimeCoalesceKey(next, "workbench.turn.snapshot")); +}); + +test("trace events keep sequence-specific coalescing keys", () => { + const first: WorkbenchRealtimeEvent = { type: "trace.event", traceId: "trc_queue", event: { traceId: "trc_queue", projectedSeq: 10 } }; + const next: WorkbenchRealtimeEvent = { type: "trace.event", traceId: "trc_queue", event: { traceId: "trc_queue", projectedSeq: 11 } }; + + assert.notEqual(realtimeCoalesceKey(first, "workbench.trace.event"), realtimeCoalesceKey(next, "workbench.trace.event")); +}); diff --git a/web/hwlab-cloud-web/src/api/workbench-events.ts b/web/hwlab-cloud-web/src/api/workbench-events.ts index bf0d2063..f7075854 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.ts @@ -176,11 +176,12 @@ function scheduleRealtimeFlushYield(flush: () => void, yieldMs: number | null | return () => clearTimeout(id); } -function realtimeCoalesceKey(event: WorkbenchRealtimeEvent, eventName: string): string | null { - const sessionId = firstScopePart(event.sessionId, event.message?.sessionId, event.snapshot?.sessionId, event.event?.sessionId); - const traceId = firstScopePart(event.traceId, event.message?.traceId, event.snapshot?.traceId, event.event?.traceId); +export function realtimeCoalesceKey(event: WorkbenchRealtimeEvent, eventName: string): string | null { + const sessionId = firstScopePart(event.sessionId, event.message?.sessionId, event.snapshot?.sessionId, event.event?.sessionId, event.turn?.sessionId); + const traceId = firstScopePart(event.traceId, event.message?.traceId, event.snapshot?.traceId, event.event?.traceId, event.turn?.traceId); const outboxSeq = numericCursor(event.cursor?.outboxSeq ?? event.outboxSeq); const traceSeq = numericCursor(event.cursor?.traceSeq ?? event.traceSeq ?? event.event?.seq ?? event.event?.projectedSeq); + if (eventName === "workbench.turn.snapshot" && traceId) return composeWorkbenchScopedKey("workbench.sse.turn-snapshot", sessionId, traceId); if (eventName === "workbench.trace.event" || eventName === "message") { const eventSeq = outboxSeq ?? traceSeq; return eventSeq === null ? null : composeWorkbenchScopedKey("workbench.sse.event", sessionId, traceId, eventName, eventSeq); 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 b39ac7a8..464d9798 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 @@ -135,7 +135,8 @@ test("terminal response helpers treat failed body as sealed authority", () => { 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("async function refreshTerminalTraceFromRest")); + 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 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")); @@ -149,6 +150,11 @@ test("workbench active terminal paths seal final response from turn authority", assert.match(source, /createKeyedSingleflight>/u); assert.match(source, /traceHydrationSingleflight\.run/u); assert.match(realtimeTurnBlock, /!terminalTurn[\s\S]*refreshRealtimeSessionMessages/u); + assert.match(realtimeTurnBlock, /rememberTurnStatus\(traceId, result\)[\s\S]*scheduleRealtimeTurnProjection\(\{ traceId, result, terminalTurn \}\)/u); + assert.doesNotMatch(realtimeTurnBlock, /applyTurnStatusSnapshot\(|refreshTerminalTraceFromRest\(/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, /workbench_realtime_turn_projection_budget/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); 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 4f5a00e3..e297c98a 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 @@ -59,6 +59,22 @@ test("server state reducer keeps session detail reference stable when detail has assert.equal(selectActiveMessages(state, sessionId), selectActiveMessages(first, sessionId)); }); +test("server state reducer skips equivalent message upsert and loadedAt-only turn status", () => { + const sessionId = "ses_state_realtime_noop"; + const traceId = "trc_state_realtime_noop"; + let state = createWorkbenchServerState(); + const message = agentMessage({ id: "msg_state_realtime_noop", status: "running", traceId, sessionId, updatedAt: "2026-07-01T00:00:01.000Z" }); + state = reduceWorkbenchServerState(state, { type: "session.detail", session: sessionRecord({ sessionId, status: "running", lastTraceId: traceId, messageCount: 1, messages: [message] }) }); + state = reduceWorkbenchServerState(state, { type: "turn.status", turn: { traceId, status: "running", running: true, terminal: false, sessionId, updatedAt: "2026-07-01T00:00:01.000Z", loadedAt: "2026-07-01T00:00:01.000Z" } }); + state = reduceWorkbenchServerState(state, { type: "message.upsert", sessionId, message: { ...message } }); + const first = state; + state = reduceWorkbenchServerState(state, { type: "message.upsert", sessionId, message: { ...message } }); + state = reduceWorkbenchServerState(state, { type: "turn.status", turn: { traceId, status: "running", running: true, terminal: false, sessionId, updatedAt: "2026-07-01T00:00:01.000Z", loadedAt: "2026-07-01T00:00:02.000Z" } }); + + assert.equal(state, first); + assert.equal(selectActiveMessages(state, sessionId), selectActiveMessages(first, sessionId)); +}); + test("server state reducer keeps failed message without terminal body unsealed", () => { const sessionId = "ses_state_failed_without_body"; const traceId = "trc_state_failed_without_body"; 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 7a6c2690..99852af7 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-server-state.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-server-state.ts @@ -170,6 +170,7 @@ function reduceSessionMessages(state: WorkbenchServerState, sessionId: string | function reduceTurnStatus(state: WorkbenchServerState, turn: TurnStatusAuthority): WorkbenchServerState { const nextTurn = mergeTurnStatusAuthority(state.turnStatusByTraceId[turn.traceId], turn); + if (turnStatusEquivalent(state.turnStatusByTraceId[turn.traceId], nextTurn)) return state; const nextTurns = { ...state.turnStatusByTraceId, [turn.traceId]: nextTurn }; const sessionStatus = sessionStatusAuthorityFromTurn(nextTurn); return { @@ -199,9 +200,11 @@ function reduceMessageUpsert(state: WorkbenchServerState, sessionId: string | nu const existingSession = state.sessionsById[id]; const existing = state.messagesBySessionId[id] ?? existingSession?.messages ?? []; const index = existing.findIndex((item) => messageMatchesSnapshot(item, message)); + const mergedMessage = index >= 0 ? mergeMessageSnapshot(existing[index], message) : message; + if (index >= 0 && messageEquivalent(existing[index], mergedMessage)) return state; const mergedMessages = index >= 0 - ? existing.map((item, itemIndex) => itemIndex === index ? mergeMessageSnapshot(item, message) : item) - : [...existing, message]; + ? existing.map((item, itemIndex) => itemIndex === index ? mergedMessage : item) + : [...existing, mergedMessage]; const session = existingSession ? { ...existingSession, messages: mergedMessages, messageCount: mergedMessages.length } : null; return { ...state, @@ -446,6 +449,18 @@ function messageListsEquivalent(left: ChatMessage[], right: ChatMessage[]): bool return true; } +function messageEquivalent(left: ChatMessage | undefined, right: ChatMessage | undefined): boolean { + return valuesEquivalent(left ?? null, right ?? null); +} + +function turnStatusEquivalent(left: TurnStatusAuthority | undefined, right: TurnStatusAuthority | undefined): boolean { + if (left === right) return true; + if (!left || !right) return false; + const leftComparable = { ...left, loadedAt: undefined }; + const rightComparable = { ...right, loadedAt: undefined }; + return valuesEquivalent(leftComparable, rightComparable); +} + function valuesEquivalent(left: unknown, right: unknown): boolean { if (Object.is(left, right)) return true; if (!left || !right || typeof left !== "object" || typeof right !== "object") return false; diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 5db9a096..02ef25d4 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -97,6 +97,12 @@ interface SessionDetailReadOptions { force?: boolean; } +interface RealtimeTurnProjectionItem { + traceId: string; + result: AgentChatResultResponse; + terminalTurn: boolean; +} + export const useWorkbenchStore = defineStore("workbench", () => { const runtimePolicy = workbenchRuntimePolicy(); const workbenchColadaReducer = useWorkbenchColadaReducer(); @@ -106,6 +112,8 @@ export const useWorkbenchStore = defineStore("workbench", () => { const traceHydrationSingleflight = createKeyedSingleflight(); const sessionMessagesReadSingleflight = createKeyedSingleflight>(); const sessionDetailReadSingleflight = createKeyedSingleflight>(); + const realtimeTurnProjectionQueue = new Map(); + let realtimeTurnProjectionScheduled = false; const providerProfile = ref(readString("hwlab.workbench.providerProfile.v1", "codex")); const providerOptions = ref(defaultProviderProfileOptions(providerProfile.value)); const recentDrafts = ref(readRecentDrafts()); @@ -730,7 +738,8 @@ export const useWorkbenchStore = defineStore("workbench", () => { const agentRun = agentRunFromResult(result, runnerTrace) ?? agentRunFromMessage(message); const projection = clearCompletedDiagnostics ? nonBlockingProjection(resultProjection) : resultProjection ?? (terminal ? null : runnerTrace.projection ?? message.projection ?? null); const terminalPatch = terminalMessagePatchFromTurnResult(message, result) ?? {}; - return { ...message, ...messageTimingPatchForMerge(message, result), ...messageStatusPatchForTerminalMerge(message, resultStatus, terminal), runnerTrace, error, projection, projectionStatus: projection?.projectionStatus ?? null, projectionHealth: projection?.projectionHealth ?? null, blocker: projection?.blocker ?? null, agentRun: agentRun ?? undefined, ...terminalPatch, updatedAt: new Date().toISOString() }; + const updatedAt = firstNonEmptyString(result.updatedAt, runnerTrace?.updatedAt, message.updatedAt) ?? new Date().toISOString(); + return { ...message, ...messageTimingPatchForMerge(message, result), ...messageStatusPatchForTerminalMerge(message, resultStatus, terminal), runnerTrace, error, projection, projectionStatus: projection?.projectionStatus ?? null, projectionHealth: projection?.projectionHealth ?? null, blocker: projection?.blocker ?? null, agentRun: agentRun ?? undefined, ...terminalPatch, updatedAt }; } async function submitMessage(text: string): Promise { @@ -1309,8 +1318,53 @@ export const useWorkbenchStore = defineStore("workbench", () => { const status = firstNonEmptyString(turn.status) ?? undefined; const terminalTurn = turn.terminal === true || isTerminalMessageStatus(status); if (activeId && !terminalTurn && !messages.value.some((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === traceId)) void refreshRealtimeSessionMessages(activeId, `realtime-turn-gap:${traceId}`); - applyTurnStatusSnapshot(traceId, { ...turn, traceId, status, running: turn.running === true, terminal: turn.terminal === true, sessionId: firstNonEmptyString(turn.sessionId) ?? undefined, threadId: firstNonEmptyString(turn.threadId) ?? undefined, agentRun: turn.agentRun as AgentRunProvenance | undefined } as AgentChatResultResponse); - if (terminalTurn) void refreshTerminalTraceFromRest(traceId, "realtime-turn-snapshot"); + const result = { ...turn, traceId, status, running: turn.running === true, terminal: turn.terminal === true, sessionId: firstNonEmptyString(turn.sessionId) ?? undefined, threadId: firstNonEmptyString(turn.threadId) ?? undefined, agentRun: turn.agentRun as AgentRunProvenance | undefined } as AgentChatResultResponse; + rememberTurnStatus(traceId, result); + scheduleRealtimeTurnProjection({ traceId, result, terminalTurn }); + } + + function scheduleRealtimeTurnProjection(item: RealtimeTurnProjectionItem): void { + realtimeTurnProjectionQueue.set(item.traceId, item); + scheduleRealtimeTurnProjectionFlush(); + } + + function scheduleRealtimeTurnProjectionFlush(): void { + if (realtimeTurnProjectionScheduled) return; + realtimeTurnProjectionScheduled = true; + const flush = () => flushRealtimeTurnProjection(); + if (typeof window !== "undefined" && typeof window.requestAnimationFrame === "function") { + window.requestAnimationFrame(flush); + return; + } + setTimeout(flush, Math.max(0, runtimePolicy.workbenchRealtimeFlushYieldMs)); + } + + function flushRealtimeTurnProjection(): void { + realtimeTurnProjectionScheduled = false; + const next = realtimeTurnProjectionQueue.values().next().value as RealtimeTurnProjectionItem | undefined; + if (!next) return; + realtimeTurnProjectionQueue.delete(next.traceId); + 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"); + } + recordWorkbenchRuntimeDiagnostic({ + module: "workbench-turn-status", + sessionId: traceResultSessionId(next.result), + traceId: next.traceId, + outcome: "ok", + diagnostic: { + code: "workbench_realtime_turn_projection_budget", + reason: "realtime-turn-snapshot", + source: "turn-snapshot-frame-queue", + remainingCount: realtimeTurnProjectionQueue.size, + terminal: next.terminalTurn, + flushDurationMs: Math.round(performanceNowMs() - startedAt), + valuesRedacted: true + } + }); + if (realtimeTurnProjectionQueue.size > 0) scheduleRealtimeTurnProjectionFlush(); } async function refreshTerminalTraceFromRest(traceId: string, reason: string): Promise { @@ -1814,6 +1868,10 @@ export const useWorkbenchStore = defineStore("workbench", () => { if (traceId) void validateAndReattachTrace(traceId); } + function performanceNowMs(): number { + return typeof performance !== "undefined" && typeof performance.now === "function" ? performance.now() : Date.now(); + } + installRealtimeVisibilityHandler(); installWorkbenchProjectionSignalHandler();