Merge pull request #2366 from pikasTech/fix/2356-response-json-storm

fix: gate Workbench heavy REST hydration

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

@@ -141,6 +141,7 @@ test("workbench active terminal paths seal final response from turn authority",
   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 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 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"));
 
@@ -160,9 +161,15 @@ test("workbench active terminal paths seal final response from turn authority",
   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(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);
   assert.match(sessionDetailReadBlock, /workbenchSessionDetailReadKey\(\{ sessionId, force: options\.force \}\)/u);
   assert.match(sessionDetailReadBlock, /fetchSession\(sessionId, \{ includeMessages: false,/u);
   assert.match(loadBlock, /const messageLimit = sessionMessageProjectionWindowLimit\(\);/u);

web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts

@@ -2,7 +2,7 @@ import assert from "node:assert/strict";
 import { test } from "bun:test";
 
 import type { ChatMessage } from "@/types";
-import { projectionMergeCommitSummary, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget";
+import { projectionMergeCommitSummary, sessionDetailAutoReadDecision, traceEventsAutoReadDecision, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey, workbenchTraceEventsReadKey } from "./workbench-session-messages-read-budget";
 import { selectActiveTurnStatusRefreshTraceIds } from "./workbench-session";
 
 function message(id: string): ChatMessage {
@@ -27,6 +27,21 @@ test("session detail singleflight key separates force freshness class", () => {
   assert.equal(fresh, workbenchSessionDetailReadKey({ sessionId: "ses_key" }));
 });
 
+test("trace events auto read skips initial and completed equivalent ranges", () => {
+  const key = workbenchTraceEventsReadKey({ traceId: "trc_range", afterProjectedSeq: 12, limit: 50 });
+
+  assert.equal(traceEventsAutoReadDecision({ traceId: "trc_range", afterProjectedSeq: 0, limit: 50 }).read, false);
+  assert.equal(traceEventsAutoReadDecision({ traceId: "trc_range", afterProjectedSeq: 12, limit: 50, cachedRange: { traceId: "trc_range", afterProjectedSeq: 12, limit: 50, nextProjectedSeq: 12, hasMore: false } }).read, false);
+  assert.equal(traceEventsAutoReadDecision({ traceId: "trc_range", afterProjectedSeq: 12, limit: 50, cachedRange: { traceId: "trc_range", afterProjectedSeq: 12, limit: 50, nextProjectedSeq: 40, hasMore: true } }).read, true);
+  assert.equal(key, workbenchTraceEventsReadKey({ traceId: "trc_range", afterProjectedSeq: 12, limit: 50 }));
+});
+
+test("session detail auto read defers trace scoped realtime refresh to turn authority", () => {
+  assert.deepEqual(sessionDetailAutoReadDecision({ traceId: "trc_turn", nowMs: 10_000, minIntervalMs: 5_000 }), { read: false, reason: "turn-authority-first" });
+  assert.deepEqual(sessionDetailAutoReadDecision({ nowMs: 10_000, lastReadAtMs: 9_000, minIntervalMs: 5_000 }), { read: false, reason: "min-interval" });
+  assert.deepEqual(sessionDetailAutoReadDecision({ nowMs: 10_000, lastReadAtMs: 1_000, minIntervalMs: 5_000 }), { read: true, reason: "metadata-only" });
+});
+
 test("projection merge summary reports no write without blocking follow-up hydrate", () => {
   const existing = [{ ...message("msg_same"), traceId: "trc_current", sessionId: "ses_same" }];
   const summary = projectionMergeCommitSummary(existing, existing);

web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts

@@ -1,5 +1,5 @@
 // SPEC: pikasTech/HWLAB#2356 Workbench bounded request storm follow-up.
-// Responsibility: Pure budgeting helpers for session messages read de-duplication and projection merge writes.
+// Responsibility: Pure budgeting helpers for Workbench read hydration de-duplication and projection merge writes.
 
 import type { ChatMessage } from "@/types";
 import { firstNonEmptyString } from "@/utils";
@@ -21,6 +21,37 @@ export interface ProjectionMergeCommitSummary {
   changedCount: number;
 }
 
+export interface TraceEventsReadRangeInput {
+  traceId: string | null | undefined;
+  afterProjectedSeq: number | null | undefined;
+  limit: number | null | undefined;
+}
+
+export interface TraceEventsReadRangeRecord extends TraceEventsReadRangeInput {
+  nextProjectedSeq: number | null;
+  hasMore: boolean;
+}
+
+export interface TraceEventsAutoReadDecisionInput extends TraceEventsReadRangeInput {
+  force?: boolean | null;
+  terminalBodyVisible?: boolean | null;
+  cachedRange?: TraceEventsReadRangeRecord | null;
+}
+
+export interface SessionDetailAutoReadDecisionInput {
+  force?: boolean | null;
+  traceId?: string | null;
+  terminalBodyVisible?: boolean | null;
+  lastReadAtMs?: number | null;
+  nowMs: number;
+  minIntervalMs: number;
+}
+
+export interface ReadHydrationDecision {
+  read: boolean;
+  reason: string;
+}
+
 export function workbenchSessionDetailReadKey(input: SessionDetailReadKeyInput): string {
   return composeWorkbenchScopedKey(
     "workbench.session-detail.read",
@@ -39,6 +70,35 @@ export function workbenchSessionMessagesReadKey(input: SessionMessagesReadKeyInp
   );
 }
 
+export function workbenchTraceEventsReadKey(input: TraceEventsReadRangeInput): string {
+  return composeWorkbenchScopedKey(
+    "workbench.trace-events.read",
+    firstNonEmptyString(input.traceId) ?? "unknown-trace",
+    `after:${readSeqPart(input.afterProjectedSeq)}`,
+    `limit:${sessionMessagesReadLimitPart(input.limit)}`
+  );
+}
+
+export function traceEventsAutoReadDecision(input: TraceEventsAutoReadDecisionInput): ReadHydrationDecision {
+  if (input.force === true) return { read: true, reason: "force" };
+  if (input.terminalBodyVisible === true) return { read: false, reason: "terminal-visible" };
+  const afterProjectedSeq = readSeqPart(input.afterProjectedSeq);
+  if (afterProjectedSeq <= 0) return { read: false, reason: "initial-range-deferred" };
+  const cached = input.cachedRange;
+  if (cached && readSeqPart(cached.afterProjectedSeq) === afterProjectedSeq && sessionMessagesReadLimitPart(cached.limit) === sessionMessagesReadLimitPart(input.limit) && cached.hasMore !== true) return { read: false, reason: "equivalent-range-complete" };
+  return { read: true, reason: "delta-range" };
+}
+
+export function sessionDetailAutoReadDecision(input: SessionDetailAutoReadDecisionInput): ReadHydrationDecision {
+  if (input.force === true) return { read: true, reason: "force" };
+  if (input.terminalBodyVisible === true) return { read: false, reason: "terminal-visible" };
+  if (firstNonEmptyString(input.traceId)) return { read: false, reason: "turn-authority-first" };
+  const lastReadAtMs = Number(input.lastReadAtMs);
+  const minIntervalMs = Math.max(0, Math.trunc(Number(input.minIntervalMs)));
+  if (Number.isFinite(lastReadAtMs) && input.nowMs - lastReadAtMs < minIntervalMs) return { read: false, reason: "min-interval" };
+  return { read: true, reason: "metadata-only" };
+}
+
 export function projectionMergeCommitSummary(existing: ChatMessage[], merged: ChatMessage[]): ProjectionMergeCommitSummary {
   let changedCount = Math.max(0, merged.length - existing.length);
   const sharedLength = Math.min(existing.length, merged.length);
@@ -52,3 +112,8 @@ function sessionMessagesReadLimitPart(value: number | null | undefined): number
   const number = Number(value);
   return Number.isFinite(number) && number > 0 ? Math.trunc(number) : 1;
 }
+
+function readSeqPart(value: number | null | undefined): number {
+  const number = Number(value);
+  return Number.isFinite(number) && number > 0 ? Math.trunc(number) : 0;
+}

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

@@ -70,7 +70,7 @@ import { planWorkbenchRealtimeApply, planWorkbenchRealtimeRecovery, type Workben
 import { useWorkbenchColadaMutations } from "./workbench-colada-mutations";
 import { useWorkbenchColadaQueries } from "./workbench-colada-queries";
 import { useWorkbenchColadaReducer } from "./workbench-colada-reducer";
-import { projectionMergeCommitSummary, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget";
+import { projectionMergeCommitSummary, sessionDetailAutoReadDecision, 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";
@@ -113,6 +113,8 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   const traceHydrationSingleflight = createKeyedSingleflight<void>();
   const sessionMessagesReadSingleflight = createKeyedSingleflight<ApiResult<WorkbenchMessagePageResponse>>();
   const sessionDetailReadSingleflight = createKeyedSingleflight<ApiResult<WorkbenchSessionDetailResponse>>();
+  const traceEventsReadRanges = new Map<string, TraceEventsReadRangeRecord>();
+  const sessionDetailAutoReadAtMs = new Map<string, number>();
   const realtimeTurnProjectionQueue = new Map<string, RealtimeTurnProjectionItem>();
   let realtimeTurnProjectionScheduled = false;
   const providerProfile = ref<ProviderProfile>(readString("hwlab.workbench.providerProfile.v1", "codex"));
@@ -888,9 +890,14 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   async function hydrateTraceEventsForMessagePages(message: ChatMessage, options: { force?: boolean } = {}): Promise<void> {
     const traceId = message.traceId ?? message.runnerTrace?.traceId;
     if (!traceId) return;
-    let afterProjectedSeq = traceHydrationProjectedSeq(message.runnerTrace);
+    let afterProjectedSeq = traceHydrationProjectedSeq(traceAuthorityById.value[traceId] ?? message.runnerTrace);
     for (let page = 0; page < runtimePolicy.traceHydrationMaxPages; page += 1) {
       if (traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId)) return;
+      const decision = traceEventsHydrationReadDecision(traceId, afterProjectedSeq, message, options);
+      if (!decision.read) {
+        recordTraceEventsHydrationSkip(traceId, message, decision.reason, afterProjectedSeq);
+        return;
+      }
       const result = await fetchTraceHydrationPage(traceId, afterProjectedSeq, { force: options.force });
       if (!result.ok || !result.data) {
         if (shouldSuppressTransientWorkbenchReadFailure(result)) return;
@@ -900,11 +907,40 @@ export const useWorkbenchStore = defineStore("workbench", () => {
       }
       applyTraceHydrationResult(traceId, result.data);
       const nextProjectedSeq = traceNextProjectedSeq(result.data, afterProjectedSeq);
+      rememberTraceEventsHydrationRead(traceId, afterProjectedSeq, nextProjectedSeq, result.data);
       if (result.data.hasMore !== true || nextProjectedSeq <= afterProjectedSeq) return;
       afterProjectedSeq = nextProjectedSeq;
     }
   }
 
+  function traceEventsHydrationReadDecision(traceId: string, afterProjectedSeq: number, message: ChatMessage, options: { force?: boolean }): { read: boolean; reason: string } {
+    const limit = runtimePolicy.traceHydrationPageLimit;
+    const key = workbenchTraceEventsReadKey({ traceId, afterProjectedSeq, limit });
+    return traceEventsAutoReadDecision({
+      traceId,
+      afterProjectedSeq,
+      limit,
+      force: options.force,
+      terminalBodyVisible: traceTerminalBodyIsVisible(traceId, message.sessionId ?? message.runnerTrace?.sessionId),
+      cachedRange: traceEventsReadRanges.get(key) ?? null
+    });
+  }
+
+  function rememberTraceEventsHydrationRead(traceId: string, afterProjectedSeq: number, nextProjectedSeq: number, result: AgentChatResultResponse): void {
+    const limit = runtimePolicy.traceHydrationPageLimit;
+    traceEventsReadRanges.set(workbenchTraceEventsReadKey({ traceId, afterProjectedSeq, limit }), { traceId, afterProjectedSeq, limit, nextProjectedSeq, hasMore: result.hasMore === true });
+  }
+
+  function recordTraceEventsHydrationSkip(traceId: string, message: ChatMessage, reason: string, afterProjectedSeq: number): void {
+    recordWorkbenchRuntimeDiagnostic({
+      module: "workbench-trace-events-read",
+      sessionId: message.sessionId ?? message.runnerTrace?.sessionId ?? null,
+      traceId,
+      outcome: "ok",
+      diagnostic: { code: "trace_events_auto_read_skip", reason, source: "trace-hydration", afterProjectedSeq, limit: runtimePolicy.traceHydrationPageLimit, valuesRedacted: true }
+    });
+  }
+
   async function fetchTraceHydrationPage(traceId: string, afterProjectedSeq: number, options: { force?: boolean } = {}): Promise<ApiResult<AgentChatResultResponse>> {
     return fetchWorkbenchTraceEvents(traceId, afterProjectedSeq, shouldUseActivityTimeoutForTrace(traceId), { force: options.force });
   }
@@ -1495,7 +1531,19 @@ export const useWorkbenchStore = defineStore("workbench", () => {
       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 } });
+      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 });