From efc777fb6da8f18637fc508f918be3f06d6f3602 Mon Sep 17 00:00:00 2001 From: UniDesk Codex Date: Fri, 3 Jul 2026 03:18:46 +0800 Subject: [PATCH] fix(workbench): budget realtime session detail refresh --- ...rkbench-message-projection-runtime.test.ts | 7 ++- .../src/stores/workbench-server-state.test.ts | 13 +++++ .../src/stores/workbench-server-state.ts | 56 ++++++++++++++++++- ...bench-session-messages-read-budget.test.ts | 10 +++- .../workbench-session-messages-read-budget.ts | 14 +++++ web/hwlab-cloud-web/src/stores/workbench.ts | 30 +++++++--- 6 files changed, 120 insertions(+), 10 deletions(-) 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 4654eed8..b39ac7a8 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 @@ -139,6 +139,7 @@ test("workbench active terminal paths seal final response from turn authority", 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")); + const sessionDetailReadBlock = source.slice(source.indexOf("function fetchSessionDetailPage"), source.indexOf("function sessionMessageProjectionWindowLimit")); 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")); @@ -154,8 +155,12 @@ test("workbench active terminal paths seal final response from turn authority", 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, /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(sessionDetailReadBlock, /workbenchSessionDetailReadKey\(\{ sessionId, force: options\.force \}\)/u); + assert.match(sessionDetailReadBlock, /fetchSession\(sessionId, \{ includeMessages: false,/u); assert.match(loadBlock, /const messageLimit = sessionMessageProjectionWindowLimit\(\);/u); - assert.match(loadBlock, /fetchSession\(requestId, \{ includeMessages: false,/u); + assert.match(loadBlock, /fetchSessionDetailPage\(requestId, \{ reason: "load-session:detail", force: true \}\)/u); assert.match(loadBlock, /sessionFromWorkbenchSession\(detail\.data\?\.session, \{ includeMessages: false \}\)/u); assert.match(loadBlock, /const fallbackMessages = seed\?\.sessionId === id \? seed\.messages \?\? \[\] : \[\];/u); assert.doesNotMatch(loadBlock, /limit:\s*100/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 c8bde864..4f5a00e3 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 @@ -46,6 +46,19 @@ test("server state reducer keeps completed message without final response unseal assert.notEqual(selectSessionStatusAuthority(state)[sessionId]?.status, "completed"); }); +test("server state reducer keeps session detail reference stable when detail has no semantic change", () => { + const sessionId = "ses_state_stable_detail"; + const traceId = "trc_state_stable_detail"; + let state = createWorkbenchServerState(); + const messages = [agentMessage({ id: "msg_state_stable_detail", status: "running", traceId, sessionId })]; + state = reduceWorkbenchServerState(state, { type: "session.detail", session: sessionRecord({ sessionId, status: "running", lastTraceId: traceId, updatedAt: "2026-07-01T00:00:00.000Z", messageCount: 1, messages }) }); + const first = state; + state = reduceWorkbenchServerState(state, { type: "session.detail", session: sessionRecord({ sessionId, status: "running", lastTraceId: traceId, updatedAt: "2026-07-01T00:00:00.000Z", messageCount: 1, messages: [...messages] }) }); + + 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 db93fe85..7a6c2690 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-server-state.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-server-state.ts @@ -116,12 +116,14 @@ function reduceSessionDetail(state: WorkbenchServerState, session: WorkbenchSess const messages = Array.isArray(session.messages) ? mergeMessageList(existingMessages, session.messages) : existingMessages; const merged = mergeSessionRecord(existing, { ...session, messages }); const sessionStatus = sessionStatusAuthorityFromDetail(session); + const nextSessionStatus = sessionStatus ? mergeSessionStatusAuthority(state.sessionStatusById[sessionId], sessionStatus) : null; + if (existing && state.sessionOrder.includes(sessionId) && sessionRecordsEquivalent(existing, merged) && messageListsEquivalent(existingMessages, messages) && sessionStatusAuthorityEquivalent(state.sessionStatusById[sessionId], nextSessionStatus)) return state; return { ...state, sessionOrder: state.sessionOrder.includes(sessionId) ? state.sessionOrder : [sessionId, ...state.sessionOrder], sessionsById: { ...state.sessionsById, [sessionId]: merged }, sessionStatusById: sessionStatus - ? { ...state.sessionStatusById, [sessionId]: mergeSessionStatusAuthority(state.sessionStatusById[sessionId], sessionStatus) } + ? { ...state.sessionStatusById, [sessionId]: nextSessionStatus ?? sessionStatus } : state.sessionStatusById, messagesBySessionId: { ...state.messagesBySessionId, @@ -415,6 +417,58 @@ function mergeSessionRecord(existing: WorkbenchSessionRecord | undefined, incomi }; } +function sessionRecordsEquivalent(left: WorkbenchSessionRecord | null | undefined, right: WorkbenchSessionRecord | null | undefined): boolean { + if (left === right) return true; + if (!left || !right) return false; + const leftMessages = Array.isArray(left.messages) ? left.messages : []; + const rightMessages = Array.isArray(right.messages) ? right.messages : []; + const leftRecord = { ...left, messages: undefined }; + const rightRecord = { ...right, messages: undefined }; + return valuesEquivalent(leftRecord, rightRecord) && messageListsEquivalent(leftMessages, rightMessages); +} + +function sessionStatusAuthorityEquivalent(left: SessionStatusAuthority | null | undefined, right: SessionStatusAuthority | null | undefined): boolean { + if (left === right) return true; + if (!left || !right) return !left && !right; + return left.sessionId === right.sessionId + && left.status === right.status + && (left.updatedAt ?? null) === (right.updatedAt ?? null) + && (left.lastTraceId ?? null) === (right.lastTraceId ?? null) + && valuesEquivalent(left.projection ?? null, right.projection ?? null); +} + +function messageListsEquivalent(left: ChatMessage[], right: ChatMessage[]): boolean { + if (left === right) return true; + if (left.length !== right.length) return false; + for (let index = 0; index < left.length; index += 1) { + if (!valuesEquivalent(left[index], right[index])) return false; + } + return true; +} + +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; + if (Array.isArray(left) || Array.isArray(right)) { + if (!Array.isArray(left) || !Array.isArray(right) || left.length !== right.length) return false; + for (let index = 0; index < left.length; index += 1) { + if (!valuesEquivalent(left[index], right[index])) return false; + } + return true; + } + const leftRecord = left as Record; + const rightRecord = right as Record; + const leftKeys = Object.keys(leftRecord).filter((key) => leftRecord[key] !== undefined).sort(); + const rightKeys = Object.keys(rightRecord).filter((key) => rightRecord[key] !== undefined).sort(); + if (leftKeys.length !== rightKeys.length) return false; + for (let index = 0; index < leftKeys.length; index += 1) { + const key = leftKeys[index]; + if (key !== rightKeys[index]) return false; + if (!valuesEquivalent(leftRecord[key], rightRecord[key])) return false; + } + return true; +} + function projectSessionRecord(session: WorkbenchSessionRecord | null | undefined, authority: SessionStatusAuthority | null | undefined): WorkbenchSessionRecord | null { if (!session) return null; if (!authority?.status || normalizedMessageStatus(session.status) === "archived") return session; diff --git a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts b/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts index e48147a2..8203b7ef 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.test.ts +++ b/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, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget"; +import { projectionMergeCommitSummary, workbenchSessionDetailReadKey, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget"; import { selectActiveTurnStatusRefreshTraceIds } from "./workbench-session"; function message(id: string): ChatMessage { @@ -19,6 +19,14 @@ test("session messages singleflight key separates limit and force class", () => assert.equal(limit20, workbenchSessionMessagesReadKey({ sessionId: "ses_key", limit: 20, force: false })); }); +test("session detail singleflight key separates force freshness class", () => { + const fresh = workbenchSessionDetailReadKey({ sessionId: "ses_key", force: false }); + const forced = workbenchSessionDetailReadKey({ sessionId: "ses_key", force: true }); + + assert.notEqual(fresh, forced); + assert.equal(fresh, workbenchSessionDetailReadKey({ sessionId: "ses_key" })); +}); + 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); diff --git a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts b/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts index 01f50d92..80cf261c 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session-messages-read-budget.ts @@ -11,11 +11,25 @@ export interface SessionMessagesReadKeyInput { force?: boolean | null; } +export interface SessionDetailReadKeyInput { + sessionId: string | null | undefined; + force?: boolean | null; +} + export interface ProjectionMergeCommitSummary { changed: boolean; changedCount: number; } +export function workbenchSessionDetailReadKey(input: SessionDetailReadKeyInput): string { + return composeWorkbenchScopedKey( + "workbench.session-detail.read", + firstNonEmptyString(input.sessionId) ?? "unknown-session", + "messages:false", + input.force === true ? "force" : "fresh" + ); +} + export function workbenchSessionMessagesReadKey(input: SessionMessagesReadKeyInput): string { return composeWorkbenchScopedKey( "workbench.session-messages.read", diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 51f29349..5db9a096 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -11,7 +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 { WorkbenchMessagePageResponse, WorkbenchSessionDetailResponse } 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"; @@ -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, workbenchSessionMessagesReadKey } from "./workbench-session-messages-read-budget"; +import { projectionMergeCommitSummary, workbenchSessionDetailReadKey, 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"; @@ -92,6 +92,11 @@ interface SessionMessagesReadOptions { force?: boolean; } +interface SessionDetailReadOptions { + reason: string; + force?: boolean; +} + export const useWorkbenchStore = defineStore("workbench", () => { const runtimePolicy = workbenchRuntimePolicy(); const workbenchColadaReducer = useWorkbenchColadaReducer(); @@ -100,6 +105,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const turnStatusReadSingleflight = createKeyedSingleflight>(); const traceHydrationSingleflight = createKeyedSingleflight(); const sessionMessagesReadSingleflight = createKeyedSingleflight>(); + const sessionDetailReadSingleflight = createKeyedSingleflight>(); const providerProfile = ref(readString("hwlab.workbench.providerProfile.v1", "codex")); const providerOptions = ref(defaultProviderProfileOptions(providerProfile.value)); const recentDrafts = ref(readRecentDrafts()); @@ -553,6 +559,11 @@ export const useWorkbenchStore = defineStore("workbench", () => { return sessionMessagesReadSingleflight.run(key, () => workbenchColadaQueries.fetchSessionMessages(sessionId, { limit: options.limit, minIntervalMs: runtimePolicy.workbenchSessionMessagesMinRefreshMs, force: options.force }), { reason: options.reason }); } + function fetchSessionDetailPage(sessionId: string, options: SessionDetailReadOptions): Promise> { + const key = workbenchSessionDetailReadKey({ sessionId, force: options.force }); + return sessionDetailReadSingleflight.run(key, () => workbenchColadaQueries.fetchSession(sessionId, { includeMessages: false, minIntervalMs: runtimePolicy.workbenchSessionDetailMinRefreshMs, force: options.force }), { reason: options.reason }); + } + function sessionMessageProjectionWindowLimit(): number { return boundedProjectionMessageLimit(runtimePolicy.workbenchSessionMessagesWindowLimit, runtimePolicy.sessionListPageLimit); } @@ -1413,15 +1424,20 @@ export const useWorkbenchStore = defineStore("workbench", () => { return; } recordActivity(reason); - const existing = sessions.value.find((item) => item.sessionId === id) ?? null; - const selected = await loadWorkbenchSession(id, existing); - if (selected && activeSessionId.value === id) { + 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 }); + if (detailSession) { + const existingMessages = serverState.value.messagesBySessionId[id] ?? []; + rememberSessionDetail({ ...detailSession, messages: existingMessages, messageCount: detailSession.messageCount ?? existingMessages.length }); if (traceTerminalBodyIsVisible(traceId, id)) { recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId: id, traceId, outcome: "ok", diagnostic: { code: "terminal_low_priority_session_detail_apply_skip", reason, source: "realtime-session-detail", valuesRedacted: true } }); return; } - applySelectedSessionDetail(selected, "system"); } + const activeMessages = serverState.value.messagesBySessionId[id] ?? messages.value; + await hydrateTurnStatusAuthority(activeMessages, { traceId, limit: 1, reason: "realtime-session-detail" }); + if (traceId && !traceProjectionIsTerminalSealed(traceId, activeMessages)) hydrateTerminalTraceGaps(selectProjectionMessageWindow(activeMessages, { traceId, limit: traceMessageProjectionWindowLimit() }), `realtime-session-detail:${reason}`); } function completeTrace(traceId: string, result: AgentChatResultResponse, options: { forceRead?: boolean } = {}): void { @@ -1758,7 +1774,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const normalizedRequestId = normalizeWorkbenchSessionId(requestId); const messageLimit = sessionMessageProjectionWindowLimit(); 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 }); + const detail = await fetchSessionDetailPage(requestId, { reason: "load-session:detail", force: true }); if (!detail.ok) return null; const detailSession = sessionFromWorkbenchSession(detail.data?.session, { includeMessages: false }); const id = detailSession?.sessionId ?? normalizedRequestId ?? seed?.sessionId;