Merge pull request #2363 from pikasTech/fix/2356-render-budget-2

fix(workbench): frame-budget realtime turn projection

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"));
+});

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 {
+export function realtimeCoalesceKey(event: WorkbenchRealtimeEvent, eventName: string): string | null {
-  const sessionId = firstScopePart(event.sessionId, event.message?.sessionId, event.snapshot?.sessionId, event.event?.sessionId);
+  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);
+  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);

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<ApiResult<AgentChatResultResponse>>/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);

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";

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.map((item, itemIndex) => itemIndex === index ? mergedMessage : item)
-    : [...existing, message];
+    : [...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;

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<void>();
   const sessionMessagesReadSingleflight = createKeyedSingleflight<ApiResult<WorkbenchMessagePageResponse>>();
   const sessionDetailReadSingleflight = createKeyedSingleflight<ApiResult<WorkbenchSessionDetailResponse>>();
+  const realtimeTurnProjectionQueue = new Map<string, RealtimeTurnProjectionItem>();
+  let realtimeTurnProjectionScheduled = false;
   const providerProfile = ref<ProviderProfile>(readString("hwlab.workbench.providerProfile.v1", "codex"));
   const providerOptions = ref<ProviderProfileOption[]>(defaultProviderProfileOptions(providerProfile.value));
   const recentDrafts = ref<DraftEntry[]>(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<boolean> {
@@ -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);
+    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;
-    if (terminalTurn) void refreshTerminalTraceFromRest(traceId, "realtime-turn-snapshot");
+    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<void> {
@@ -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();