Merge pull request #2361 from pikasTech/fix/2356-bounded-request-storm

fix(web): bound Workbench projection hydrate storm

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

@@ -24,7 +24,7 @@ function installColadaContext(): ReturnType<typeof createApp> {
 
 test("workbench colada keys keep serializable hierarchical cache scopes", () => {
   assert.deepEqual(workbenchColadaKeys.sessions({ limit: 50, includeSessionId: "ses_colada", cursor: "cur_2" }), ["workbench", "sessions", { cursor: "cur_2", includeSessionId: "ses_colada", limit: 50 }]);
-  assert.deepEqual(workbenchColadaKeys.sessionMessages("ses_colada", { limit: 100 }), ["workbench", "session-messages", "ses_colada", { cursor: null, limit: 100 }]);
+  assert.deepEqual(workbenchColadaKeys.sessionMessages("ses_colada", { limit: 20 }), ["workbench", "session-messages", "ses_colada", { cursor: null, limit: 20 }]);
   assert.deepEqual(workbenchColadaKeys.traceEvents("trc_colada", { afterProjectedSeq: 4, limit: 25 }), ["workbench", "trace-events", "trc_colada", { afterProjectedSeq: 4, limit: 25 }]);
 });
 

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

@@ -0,0 +1,31 @@
+import assert from "node:assert/strict";
+import { test } from "bun:test";
+
+import type { ChatMessage } from "@/types";
+import { mergeBoundedProjectionMessages } from "./workbench-message-projection-budget";
+
+function message(input: Partial<ChatMessage> & Pick<ChatMessage, "id" | "role">): ChatMessage {
+  return {
+    text: "",
+    status: "completed",
+    createdAt: "2026-07-02T00:00:00.000Z",
+    ...input
+  } as ChatMessage;
+}
+
+test("bounded projection merge keeps the same array when no message changes", () => {
+  const existing = [message({ id: "msg_1", role: "agent", traceId: "trc_1", text: "sealed" })];
+  assert.equal(mergeBoundedProjectionMessages(existing, existing), existing);
+});
+
+test("bounded projection merge patches by trace without replacing unrelated messages", () => {
+  const user = message({ id: "msg_user", role: "user", traceId: "trc_1", text: "prompt" });
+  const agent = message({ id: "msg_agent", role: "agent", traceId: "trc_1", text: "old" });
+  const incoming = message({ id: "msg_agent_next", role: "agent", traceId: "trc_1", text: "new" });
+  const existing = [user, agent];
+  const merged = mergeBoundedProjectionMessages(existing, [incoming]);
+
+  assert.notEqual(merged, existing);
+  assert.equal(merged[0], user);
+  assert.equal(merged[1], incoming);
+});

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

@@ -38,19 +38,23 @@ export function mergeBoundedProjectionMessages(existing: ChatMessage[], incoming
   }
   if (incomingByKey.size === 0) return existing;
   const seen = new Set<string>();
+  let changed = false;
   const merged = existing.map((message) => {
     const key = messageProjectionKey(message);
     if (!key) return message;
     seen.add(key);
-    return incomingByKey.get(key) ?? message;
+    const next = incomingByKey.get(key) ?? message;
+    if (next !== message) changed = true;
+    return next;
   });
   for (const message of incoming) {
     const key = messageProjectionKey(message);
     if (!key || seen.has(key)) continue;
     seen.add(key);
     merged.push(message);
+    changed = true;
   }
-  return merged;
+  return changed ? merged : existing;
 }
 
 export function traceProjectionIsTerminalSealed(traceId: string | null | undefined, source: ChatMessage[]): boolean {

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

@@ -0,0 +1,36 @@
+import assert from "node:assert/strict";
+import { test } from "bun:test";
+
+import type { ChatMessage } from "@/types";
+import { projectionMergeCommitSummary, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget";
+import { selectActiveTurnStatusRefreshTraceIds } from "./workbench-session";
+
+function message(id: string): ChatMessage {
+  return { id, role: "agent", text: "", status: "running", createdAt: "2026-07-02T00:00:00.000Z" } as ChatMessage;
+}
+
+test("session messages singleflight key separates limit and force class", () => {
+  const limit8 = workbenchSessionMessagesReadKey({ sessionId: "ses_key", limit: 8, force: false });
+  const limit20 = workbenchSessionMessagesReadKey({ sessionId: "ses_key", limit: 20, force: false });
+  const force20 = workbenchSessionMessagesReadKey({ sessionId: "ses_key", limit: 20, force: true });
+
+  assert.notEqual(limit8, limit20);
+  assert.notEqual(limit20, force20);
+  assert.equal(limit20, workbenchSessionMessagesReadKey({ sessionId: "ses_key", limit: 20, force: false }));
+});
+
+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);
+  const hydrateTraceIds = selectActiveTurnStatusRefreshTraceIds({ messages: existing, currentRequestTraceId: "trc_current", limit: 1 });
+
+  assert.deepEqual(summary, { changed: false, changedCount: 0 });
+  assert.deepEqual(hydrateTraceIds, ["trc_current"]);
+});
+
+test("projection merge summary reports changed message references", () => {
+  const existing = [message("msg_old")];
+  const merged = [message("msg_new")];
+
+  assert.deepEqual(projectionMergeCommitSummary(existing, merged), { changed: true, changedCount: 1 });
+});

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

@@ -0,0 +1,40 @@
+// SPEC: pikasTech/HWLAB#2356 Workbench bounded request storm follow-up.
+// Responsibility: Pure budgeting helpers for session messages read de-duplication and projection merge writes.
+
+import type { ChatMessage } from "@/types";
+import { firstNonEmptyString } from "@/utils";
+import { composeWorkbenchScopedKey } from "@/utils/workbench-key";
+
+export interface SessionMessagesReadKeyInput {
+  sessionId: string | null | undefined;
+  limit: number | null | undefined;
+  force?: boolean | null;
+}
+
+export interface ProjectionMergeCommitSummary {
+  changed: boolean;
+  changedCount: number;
+}
+
+export function workbenchSessionMessagesReadKey(input: SessionMessagesReadKeyInput): string {
+  return composeWorkbenchScopedKey(
+    "workbench.session-messages.read",
+    firstNonEmptyString(input.sessionId) ?? "unknown-session",
+    sessionMessagesReadLimitPart(input.limit),
+    input.force === true ? "force" : "fresh"
+  );
+}
+
+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);
+  for (let index = 0; index < sharedLength; index += 1) {
+    if (existing[index] !== merged[index]) changedCount += 1;
+  }
+  return { changed: changedCount > 0, changedCount };
+}
+
+function sessionMessagesReadLimitPart(value: number | null | undefined): number {
+  const number = Number(value);
+  return Number.isFinite(number) && number > 0 ? Math.trunc(number) : 1;
+}

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

@@ -2,7 +2,7 @@ import assert from "node:assert/strict";
 import { test } from "bun:test";
 
 import type { ChatMessage } from "../types";
-import { resolveCancelableAgentMessage, resolveComposerState } from "./workbench-session";
+import { resolveCancelableAgentMessage, resolveComposerState, selectActiveTurnStatusRefreshTraceIds } from "./workbench-session";
 import type { TurnStatusAuthority } from "./workbench-session";
 
 function agentMessage(input: Partial<ChatMessage> & Pick<ChatMessage, "id" | "traceId" | "sessionId">): ChatMessage {
@@ -84,3 +84,18 @@ test("cancel target remains available for failed message until terminal body exi
   assert.equal(resolveCancelableAgentMessage({ messages: [unsealed], targetTraceId: traceId, targetSessionId: sessionId, turnStatusAuthority: { [traceId]: turnStatus({ traceId, sessionId, status: "running", running: true, terminal: false }) } })?.id, unsealed.id);
   assert.equal(resolveCancelableAgentMessage({ messages: [sealed], targetTraceId: traceId, targetSessionId: sessionId, turnStatusAuthority: { [traceId]: turnStatus({ traceId, sessionId, status: "running", running: true, terminal: false }) } }), null);
 });
+
+test("turn status refresh selector keeps automatic hydrate bounded to the active trace", () => {
+  const sessionId = "ses_turn_hydrate_budget";
+  const older = agentMessage({ id: "msg_older", traceId: "trc_older", sessionId, status: "running" });
+  const current = agentMessage({ id: "msg_current", traceId: "trc_current", sessionId, status: "running" });
+
+  assert.deepEqual(selectActiveTurnStatusRefreshTraceIds({ messages: [older, current], currentRequestTraceId: current.traceId, limit: 1 }), [current.traceId]);
+});
+
+test("turn status refresh selector skips sealed terminal traces", () => {
+  const sessionId = "ses_turn_hydrate_sealed";
+  const sealed = agentMessage({ id: "msg_sealed", traceId: "trc_sealed", sessionId, status: "completed", text: "final" });
+
+  assert.deepEqual(selectActiveTurnStatusRefreshTraceIds({ messages: [sealed], currentRequestTraceId: sealed.traceId, turnStatusAuthority: { [sealed.traceId]: turnStatus({ traceId: sealed.traceId, sessionId, status: "completed", running: false, terminal: true }) }, limit: 1 }), []);
+});

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

@@ -118,6 +118,7 @@ export function selectActiveTurnStatusRefreshTraceIds(input: { messages: ChatMes
   const limit = Math.max(1, Math.trunc(input.limit ?? 3));
   const currentRequestTraceId = firstNonEmptyString(input.currentRequestTraceId);
   if (currentRequestTraceId && traceNeedsTurnStatusRefresh(currentRequestTraceId, input.messages, input.turnStatusAuthority)) traceIds.add(currentRequestTraceId);
+  if (traceIds.size >= limit) return [...traceIds];
   for (const message of [...input.messages].reverse()) {
     const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);
     if (!traceId || traceIds.has(traceId)) continue;

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

@@ -11,6 +11,7 @@ import { agentErrorFromProjection, normalizeApiErrorRecord, normalizeErrorDiagno
 import { readWorkbenchJson, readWorkbenchNumber, readWorkbenchString, removeWorkbenchStorageKey, writeWorkbenchJson, writeWorkbenchString } from "@/utils/workbench-storage-runtime";
 import { createWorkbenchStreamTransportRuntime, type WorkbenchRealtimeEvent, type WorkbenchStreamTransportRecovery } from "@/utils/workbench-realtime-runtime";
 import { mergeRunnerTrace, snapshotToRunnerTrace, type TraceSnapshot } from "@/composables/useTraceSubscription";
+import type { WorkbenchMessagePageResponse } from "@/api/workbench";
 import type { AgentChatResponse, AgentChatResultResponse, AgentRunProvenance, ApiError, ApiResult, ChatMessage, ErrorDiagnostic, LiveSurface, ProjectionBlocker, ProjectionDiagnostic, ProviderProfile, TraceEvent, WorkbenchSessionRecord, WorkbenchTurnTimingProjection } from "@/types";
 import { firstNonEmptyString, nextProtocolId, normalizeWorkbenchSessionId, normalizeWorkbenchSessionRouteId } from "@/utils";
 import { composeWorkbenchScopedKey } from "@/utils/workbench-key";
@@ -69,6 +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, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget";
 
 const WORKBENCH_SESSION_PROJECTION_SIGNAL_CHANNEL = "hwlab.workbench.sessionProjection.v1";
 const WORKBENCH_SESSION_PROJECTION_SIGNAL_KEY = "hwlab.workbench.sessionProjectionSignal.v1";
@@ -84,6 +86,12 @@ interface SelectSessionOptions {
   source?: SessionSelectionSource;
 }
 
+interface SessionMessagesReadOptions {
+  limit: number;
+  reason: string;
+  force?: boolean;
+}
+
 export const useWorkbenchStore = defineStore("workbench", () => {
   const runtimePolicy = workbenchRuntimePolicy();
   const workbenchColadaReducer = useWorkbenchColadaReducer();
@@ -91,6 +99,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   const workbenchColadaMutations = useWorkbenchColadaMutations();
   const turnStatusReadSingleflight = createKeyedSingleflight<ApiResult<AgentChatResultResponse>>();
   const traceHydrationSingleflight = createKeyedSingleflight<void>();
+  const sessionMessagesReadSingleflight = createKeyedSingleflight<ApiResult<WorkbenchMessagePageResponse>>();
   const providerProfile = ref<ProviderProfile>(readString("hwlab.workbench.providerProfile.v1", "codex"));
   const providerOptions = ref<ProviderProfileOption[]>(defaultProviderProfileOptions(providerProfile.value));
   const recentDrafts = ref<DraftEntry[]>(readRecentDrafts());
@@ -504,12 +513,12 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   async function refreshSessionMessageProjectionPage(sessionId: string | null | undefined, options: { force?: boolean } = {}): Promise<void> {
     const id = normalizeWorkbenchSessionId(sessionId);
     if (!id) return;
-    const response = await workbenchColadaQueries.fetchSessionMessages(id, { limit: sessionMessageProjectionWindowLimit(), minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs, force: options.force });
+    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() });
-    rememberSessionMessages(id, merged);
-    await hydrateTurnStatusAuthority(merged);
+    commitMergedSessionMessages(id, pageMessages, merged, { limit: sessionMessageProjectionWindowLimit(), reason: "session-message-page" });
+    await hydrateTurnStatusAuthority(merged, { limit: 1, reason: "session-message-page" });
     hydrateTerminalTraceGaps(merged, "session-message-page");
   }
 
@@ -525,7 +534,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     if (!id) return;
     const existing = serverState.value.messagesBySessionId[id] ?? [];
     if (traceProjectionIsTerminalSealed(traceId, existing)) return;
-    const response = await workbenchColadaQueries.fetchSessionMessages(id, { limit: traceMessageProjectionWindowLimit(), minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs, force: options.force });
+    const response = await fetchSessionMessagesPage(id, { limit: traceMessageProjectionWindowLimit(), reason: `trace-message-page:${traceId}`, force: options.force });
     if (!response.ok || !response.data) {
       if (shouldSuppressTransientWorkbenchReadFailure(response)) return;
       if (traceHasTerminalResponse(traceId, messages.value)) return;
@@ -534,11 +543,16 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     }
     const pageMessages = Array.isArray(response.data.messages) ? response.data.messages.map((message) => normalizeChatMessage(message as ChatMessage)) : [];
     const merged = mergeMessageProjectionPage(id, pageMessages, { traceId, limit: traceMessageProjectionWindowLimit() });
-    rememberSessionMessages(id, merged);
-    await hydrateTurnStatusAuthority(merged);
+    commitMergedSessionMessages(id, pageMessages, merged, { traceId, limit: traceMessageProjectionWindowLimit(), reason: "trace-message-page" });
+    await hydrateTurnStatusAuthority(merged, { traceId, limit: 1, reason: "trace-message-page" });
     if (!traceProjectionIsTerminalSealed(traceId, merged)) hydrateTerminalTraceGaps(merged, `trace-message-page:${traceId}`);
   }
 
+  function fetchSessionMessagesPage(sessionId: string, options: SessionMessagesReadOptions): Promise<ApiResult<WorkbenchMessagePageResponse>> {
+    const key = workbenchSessionMessagesReadKey({ sessionId, limit: options.limit, force: options.force });
+    return sessionMessagesReadSingleflight.run(key, () => workbenchColadaQueries.fetchSessionMessages(sessionId, { limit: options.limit, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs, force: options.force }), { reason: options.reason });
+  }
+
   function sessionMessageProjectionWindowLimit(): number {
     return boundedProjectionMessageLimit(runtimePolicy.workbenchSessionMessagesWindowLimit, runtimePolicy.sessionListPageLimit);
   }
@@ -554,6 +568,29 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     return mergeBoundedProjectionMessages(existing, incoming);
   }
 
+  function commitMergedSessionMessages(sessionId: string, incoming: ChatMessage[], merged: ChatMessage[], options: { traceId?: string | null; limit: number; reason: string }): void {
+    const existing = serverState.value.messagesBySessionId[sessionId] ?? [];
+    const summary = projectionMergeCommitSummary(existing, merged);
+    recordWorkbenchRuntimeDiagnostic({
+      module: "workbench-message-projection",
+      sessionId,
+      traceId: firstNonEmptyString(options.traceId) ?? null,
+      outcome: "ok",
+      diagnostic: {
+        code: "workbench_message_projection_merge",
+        reason: options.reason,
+        source: "session-messages-page",
+        inputCount: incoming.length,
+        existingCount: existing.length,
+        mergedCount: merged.length,
+        changedCount: summary.changedCount,
+        limit: options.limit,
+        valuesRedacted: true
+      }
+    });
+    if (summary.changed) rememberSessionMessages(sessionId, merged);
+  }
+
   function mergeMessageProjectionMessage(message: ChatMessage, existing: ChatMessage[]): ChatMessage {
     const previous = findExistingProjectionMessage(message, existing);
     const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId, previous?.traceId, previous?.runnerTrace?.traceId);
@@ -576,12 +613,29 @@ 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): Promise<void> {
-    await Promise.all(activeTurnStatusRefreshTraceIds(source).map((traceId) => refreshTurnStatusByTraceId(traceId)));
+  async function hydrateTurnStatusAuthority(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string } = {}): Promise<void> {
+    for (const traceId of activeTurnStatusRefreshTraceIds(source, options)) {
+      await refreshTurnStatusByTraceId(traceId);
+    }
   }
 
-  function activeTurnStatusRefreshTraceIds(source: ChatMessage[] = messages.value): string[] {
-    return selectActiveTurnStatusRefreshTraceIds({ messages: source, currentRequestTraceId: currentRequest.value?.traceId, turnStatusAuthority: turnStatusAuthority.value, limit: 3 });
+  function activeTurnStatusRefreshTraceIds(source: ChatMessage[] = messages.value, options: { traceId?: string | null; limit?: number; reason?: string } = {}): string[] {
+    const requestedTraceId = firstNonEmptyString(options.traceId, currentRequest.value?.traceId);
+    const traceIds = selectActiveTurnStatusRefreshTraceIds({ messages: source, currentRequestTraceId: requestedTraceId, turnStatusAuthority: turnStatusAuthority.value, limit: options.limit ?? 1 });
+    recordWorkbenchRuntimeDiagnostic({
+      module: "workbench-turn-status",
+      traceId: requestedTraceId ?? traceIds[0] ?? null,
+      outcome: "ok",
+      diagnostic: {
+        code: "workbench_turn_status_hydrate_budget",
+        reason: options.reason ?? "auto-turn-status-hydrate",
+        source: "session-message-projection",
+        requestedCount: traceIds.length,
+        limit: options.limit ?? 1,
+        valuesRedacted: true
+      }
+    });
+    return traceIds;
   }
 
   async function refreshTurnStatusByTraceId(traceId: string | null | undefined, options: { force?: boolean } = {}): Promise<void> {
@@ -1703,13 +1757,13 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     if (!requestId) return null;
     const normalizedRequestId = normalizeWorkbenchSessionId(requestId);
     const messageLimit = sessionMessageProjectionWindowLimit();
-    const eagerMessages = normalizedRequestId ? workbenchColadaQueries.fetchSessionMessages(normalizedRequestId, { limit: messageLimit, force: true, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs }) : null;
+    const eagerMessages = normalizedRequestId ? fetchSessionMessagesPage(normalizedRequestId, { limit: messageLimit, reason: "load-session:eager", force: true }) : null;
     const detail = await workbenchColadaQueries.fetchSession(requestId, { includeMessages: false, force: true, minIntervalMs: runtimePolicy.workbenchSessionDetailMinRefreshMs });
     if (!detail.ok) return null;
     const detailSession = sessionFromWorkbenchSession(detail.data?.session, { includeMessages: false });
     const id = detailSession?.sessionId ?? normalizedRequestId ?? seed?.sessionId;
     if (!id) return null;
-    const messages = eagerMessages && id === normalizedRequestId ? await eagerMessages : await workbenchColadaQueries.fetchSessionMessages(id, { limit: messageLimit, force: true, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs });
+    const messages = eagerMessages && id === normalizedRequestId ? await eagerMessages : await fetchSessionMessagesPage(id, { limit: messageLimit, reason: "load-session", force: true });
     const fallbackMessages = seed?.sessionId === id ? seed.messages ?? [] : [];
     const base = detailSession ? { ...detailSession, messages: fallbackMessages } : seed;
     if (!base) return null;