From dbe31ce56615ba4bc6fd61ce11717b70f4e19736 Mon Sep 17 00:00:00 2001 From: UniDesk Codex Date: Fri, 3 Jul 2026 02:25:43 +0800 Subject: [PATCH] fix: bound workbench projection hydrate storm --- .../src/stores/workbench-colada.test.ts | 2 +- ...orkbench-message-projection-budget.test.ts | 31 +++++++ .../workbench-message-projection-budget.ts | 8 +- .../src/stores/workbench-session.test.ts | 17 +++- .../src/stores/workbench-session.ts | 1 + web/hwlab-cloud-web/src/stores/workbench.ts | 88 ++++++++++++++++--- 6 files changed, 131 insertions(+), 16 deletions(-) create mode 100644 web/hwlab-cloud-web/src/stores/workbench-message-projection-budget.test.ts diff --git a/web/hwlab-cloud-web/src/stores/workbench-colada.test.ts b/web/hwlab-cloud-web/src/stores/workbench-colada.test.ts index b18971e0..8cf5f384 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-colada.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-colada.test.ts @@ -24,7 +24,7 @@ function installColadaContext(): ReturnType { 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 }]); }); diff --git a/web/hwlab-cloud-web/src/stores/workbench-message-projection-budget.test.ts b/web/hwlab-cloud-web/src/stores/workbench-message-projection-budget.test.ts new file mode 100644 index 00000000..e1de9008 --- /dev/null +++ b/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 & Pick): 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); +}); diff --git a/web/hwlab-cloud-web/src/stores/workbench-message-projection-budget.ts b/web/hwlab-cloud-web/src/stores/workbench-message-projection-budget.ts index eeb23e43..1b4c5e48 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-message-projection-budget.ts +++ b/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(); + 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 { diff --git a/web/hwlab-cloud-web/src/stores/workbench-session.test.ts b/web/hwlab-cloud-web/src/stores/workbench-session.test.ts index 53d577e9..222afc01 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session.test.ts +++ b/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 & Pick): 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 }), []); +}); diff --git a/web/hwlab-cloud-web/src/stores/workbench-session.ts b/web/hwlab-cloud-web/src/stores/workbench-session.ts index 57dd6fe4..0fb93c84 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session.ts +++ b/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; diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 2f9fe0e3..b3ba60a1 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/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"; @@ -84,6 +85,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 +98,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const workbenchColadaMutations = useWorkbenchColadaMutations(); const turnStatusReadSingleflight = createKeyedSingleflight>(); const traceHydrationSingleflight = createKeyedSingleflight(); + const sessionMessagesReadSingleflight = createKeyedSingleflight>(); const providerProfile = ref(readString("hwlab.workbench.providerProfile.v1", "codex")); const providerOptions = ref(defaultProviderProfileOptions(providerProfile.value)); const recentDrafts = ref(readRecentDrafts()); @@ -504,12 +512,12 @@ export const useWorkbenchStore = defineStore("workbench", () => { async function refreshSessionMessageProjectionPage(sessionId: string | null | undefined, options: { force?: boolean } = {}): Promise { 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); + if (!commitMergedSessionMessages(id, pageMessages, merged, { limit: sessionMessageProjectionWindowLimit(), reason: "session-message-page" })) return; + await hydrateTurnStatusAuthority(merged, { limit: 1, reason: "session-message-page" }); hydrateTerminalTraceGaps(merged, "session-message-page"); } @@ -525,7 +533,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 +542,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); + if (!commitMergedSessionMessages(id, pageMessages, merged, { traceId, limit: traceMessageProjectionWindowLimit(), reason: "trace-message-page" })) return; + 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> { + const key = composeWorkbenchScopedKey("workbench.session-messages.read", sessionId); + 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 +567,40 @@ 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 }): boolean { + const existing = serverState.value.messagesBySessionId[sessionId] ?? []; + const changedCount = projectionChangedMessageCount(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, + limit: options.limit, + valuesRedacted: true + } + }); + if (changedCount === 0) return false; + rememberSessionMessages(sessionId, merged); + return true; + } + + function projectionChangedMessageCount(existing: ChatMessage[], merged: ChatMessage[]): number { + let changed = 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]) changed += 1; + } + return changed; + } + 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 +623,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 { - 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 { + 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 { @@ -1703,13 +1767,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;