From 9b7da7c9eb2204463765ecd09fd09928b57af1c1 Mon Sep 17 00:00:00 2001 From: UniDesk Codex Date: Thu, 2 Jul 2026 12:17:13 +0800 Subject: [PATCH] fix: hydrate passive workbench turn authority --- .../workbench-realtime-runtime.test.ts | 37 ++++++++++++++++ .../src/stores/workbench-session.ts | 39 +++++++++++++++++ web/hwlab-cloud-web/src/stores/workbench.ts | 42 ++----------------- 3 files changed, 80 insertions(+), 38 deletions(-) diff --git a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts index 38efe779..be6223d9 100644 --- a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts @@ -21,6 +21,7 @@ import { reduceWorkbenchRealtimeEvent } from "../src/stores/workbench-event-redu import { planWorkbenchRealtimeApply, planWorkbenchRealtimeRecovery } from "../src/stores/workbench-realtime-plan.ts"; import { cleanupWorkbenchServerStateDroppedSessions, cleanupWorkbenchServerStateSessions, createWorkbenchServerState, reduceWorkbenchServerState } from "../src/stores/workbench-server-state.ts"; import { cleanupDroppedWorkbenchSessionCaches, trimWorkbenchSessionCache } from "../src/stores/workbench-session-cache.ts"; +import { selectActiveTurnStatusRefreshTraceIds } from "../src/stores/workbench-session.ts"; test("Workbench scoped keys encode delimiter characters", () => { assert.deepEqual(splitWorkbenchScopedKey(workbenchRealtimeScopeKey("ses|one", "trc/two")), ["workbench.realtime", "ses|one", "trc/two"]); @@ -285,6 +286,42 @@ test("server-state cleanup removes dropped session trace and turn authority", () assert.deepEqual(Object.keys(cleanedById.traceById), []); }); +test("passive message projection selects active turn status refresh without local request", () => { + const traceIds = selectActiveTurnStatusRefreshTraceIds({ + messages: [ + agentMessage({ id: "m_old", traceId: "trc_old", status: "completed", text: "done" }), + agentMessage({ id: "m_new", traceId: "trc_new", status: "running", text: "thinking" }) + ], + currentRequestTraceId: null, + turnStatusAuthority: {} + }); + + assert.deepEqual(traceIds, ["trc_new"]); +}); + +test("passive message projection skips sealed terminal turn status refresh", () => { + const traceIds = selectActiveTurnStatusRefreshTraceIds({ + messages: [agentMessage({ id: "m_done", traceId: "trc_done", status: "completed", text: "done" })], + turnStatusAuthority: { trc_done: { traceId: "trc_done", status: "completed", running: false, terminal: true, sessionId: "ses_1" } } + }); + + assert.deepEqual(traceIds, []); +}); + +test("turn status refresh keeps local request trace priority", () => { + const traceIds = selectActiveTurnStatusRefreshTraceIds({ + messages: [ + agentMessage({ id: "m_request", traceId: "trc_request", status: "pending", text: "" }), + agentMessage({ id: "m_observed", traceId: "trc_observed", status: "running", text: "thinking" }) + ], + currentRequestTraceId: "trc_request", + turnStatusAuthority: {}, + limit: 2 + }); + + assert.deepEqual(traceIds, ["trc_request", "trc_observed"]); +}); + test("realtime event reducer classifies SSE payloads before store side effects", () => { const trace = reduceWorkbenchRealtimeEvent({ type: "trace.event", traceId: "trc_1", event: { traceId: "trc_1", label: "delta" }, snapshot: { traceId: "trc_1", status: "running" } }, "workbench.trace.event"); assert.equal(trace.activityLabel, "realtime:trace.event"); diff --git a/web/hwlab-cloud-web/src/stores/workbench-session.ts b/web/hwlab-cloud-web/src/stores/workbench-session.ts index 089988ef..1b9f81c0 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session.ts @@ -112,6 +112,21 @@ export function resolveCancelableAgentMessage(input: { messages: ChatMessage[]; return null; } +export function selectActiveTurnStatusRefreshTraceIds(input: { messages: ChatMessage[]; currentRequestTraceId?: string | null; turnStatusAuthority?: TurnStatusAuthorityMap; limit?: number }): string[] { + const traceIds = new Set(); + 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); + for (const message of [...input.messages].reverse()) { + const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId); + if (!traceId || traceIds.has(traceId)) continue; + if (!messageHasActiveTurnStatusEvidence(message, traceId, currentRequestTraceId)) continue; + if (traceNeedsTurnStatusRefresh(traceId, input.messages, input.turnStatusAuthority)) traceIds.add(traceId); + if (traceIds.size >= limit) break; + } + return [...traceIds]; +} + export function shouldShowSessionListLoading(input: { loading: boolean; sessionsReady: boolean }): boolean { return input.loading && !input.sessionsReady; } @@ -282,6 +297,30 @@ function latestAgentMessage(messages: ChatMessage[] | undefined): ChatMessage | return [...(messages ?? [])].reverse().find((message) => message.role === "agent") ?? null; } +function latestMessageForTrace(messages: ChatMessage[], traceId: string): ChatMessage | null { + return [...messages].reverse().find((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === traceId) ?? null; +} + +function messageHasActiveTurnStatusEvidence(message: ChatMessage, traceId: string, currentRequestTraceId: string | null): boolean { + if (currentRequestTraceId === traceId) return true; + return isActiveStatus(message.status) || message.traceAutoLifecycle === "running"; +} + +function traceNeedsTurnStatusRefresh(traceId: string | null | undefined, messages: ChatMessage[], turnStatusAuthority: TurnStatusAuthorityMap | undefined): boolean { + const id = firstNonEmptyString(traceId); + if (!id) return false; + const turn = turnStatusAuthority?.[id]; + const message = latestMessageForTrace(messages, id); + if (turnStatusAuthorityIsSealed(turn, message)) return false; + if (messageIsTerminalForSession(message)) return false; + return true; +} + +function turnStatusAuthorityIsSealed(turn: TurnStatusAuthority | undefined, message: ChatMessage | null): boolean { + if (!(turn?.terminal === true || isTerminalStatus(turn?.status))) return false; + return message ? messageHasTerminalResponse(message) : false; +} + function messageIsTerminalForSession(message: ChatMessage | null | undefined): boolean { if (!message || message.role !== "agent") return false; const status = normalizeSessionStatus(message.status); diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index f7ba813f..e8faaa3d 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -15,7 +15,7 @@ import type { AgentChatResponse, AgentChatResultResponse, AgentRunProvenance, Ap import { firstNonEmptyString, nextProtocolId, normalizeWorkbenchSessionId, normalizeWorkbenchSessionRouteId } from "@/utils"; import { composeWorkbenchScopedKey } from "@/utils/workbench-key"; import { failWorkbenchSessionSwitch, failWorkbenchSubmitJourney, finishWorkbenchSessionSwitchFullLoad, markWorkbenchSubmitApiAccepted, markWorkbenchTraceEventsReceived, markWorkbenchTraceProjected, recordWorkbenchLoadingState, recordWorkbenchRuntimeDiagnostic, startWorkbenchSessionSwitch, startWorkbenchSubmitJourney } from "@/utils/workbench-performance"; -import { RECENT_DRAFTS_STORAGE_KEY, appendSessionPage, defaultProviderProfileOptions, isArchivedSession, mergeSessionIntoList, normalizeChatMessageStatus, normalizeRecentDrafts, normalizeWorkbenchMessageTitle, providerProfileOptionsFromPayload, recordRecentDraft, resolveCancelableAgentMessage, resolveComposerState, shouldShowSessionListLoading, sortSessionTabs, stableSessionList, type DraftEntry, type ProviderProfileOption, type TurnStatusAuthority } from "./workbench-session"; +import { RECENT_DRAFTS_STORAGE_KEY, appendSessionPage, defaultProviderProfileOptions, isArchivedSession, mergeSessionIntoList, normalizeChatMessageStatus, normalizeRecentDrafts, normalizeWorkbenchMessageTitle, providerProfileOptionsFromPayload, recordRecentDraft, resolveCancelableAgentMessage, resolveComposerState, selectActiveTurnStatusRefreshTraceIds, shouldShowSessionListLoading, sortSessionTabs, stableSessionList, type DraftEntry, type ProviderProfileOption, type TurnStatusAuthority } from "./workbench-session"; import { initialWorkbenchSessionIdFromLocation } from "./workbench-projection"; import { cleanupWorkbenchServerStateSessions, createWorkbenchServerState, reduceWorkbenchServerState, selectActiveMessages, selectActiveSession, selectSessionList, selectSessionStatusAuthority, selectTraceAuthorityById, selectTurnStatusAuthority, type WorkbenchServerAction } from "./workbench-server-state"; import { cleanupDroppedWorkbenchSessionCaches, trimWorkbenchSessionCache } from "./workbench-session-cache"; @@ -538,6 +538,7 @@ 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); rememberSessionMessages(id, merged); + await hydrateTurnStatusAuthority(merged); hydrateTerminalTraceGaps(merged, "session-message-page"); } @@ -567,6 +568,7 @@ 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); rememberSessionMessages(id, merged); + await hydrateTurnStatusAuthority(merged); hydrateTerminalTraceGaps(merged, `trace-message-page:${traceId}`); } @@ -602,43 +604,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { } function activeTurnStatusRefreshTraceIds(source: ChatMessage[] = messages.value): string[] { - const traceIds = new Set(); - const requestTraceId = firstNonEmptyString(currentRequest.value?.traceId); - if (requestTraceId && traceNeedsTurnStatusRefresh(requestTraceId, source)) traceIds.add(requestTraceId); - for (const message of [...source].reverse()) { - const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId); - if (!traceId || traceIds.has(traceId)) continue; - if (traceId !== requestTraceId) continue; - if (!messageHasActiveTurnStatusEvidence(message, traceId)) continue; - if (traceNeedsTurnStatusRefresh(traceId, source)) traceIds.add(traceId); - if (traceIds.size >= 3) break; - } - return [...traceIds]; - } - - function messageHasActiveTurnStatusEvidence(message: ChatMessage, traceId: string): boolean { - if (currentRequest.value?.traceId === traceId) return true; - return isTraceActiveStatus(message.status); - } - - function traceNeedsTurnStatusRefresh(traceId: string | null | undefined, source: ChatMessage[] = messages.value): boolean { - const id = firstNonEmptyString(traceId); - if (!id) return false; - const turn = turnStatusAuthority.value[id]; - const message = [...source].reverse().find((item) => firstNonEmptyString(item.traceId, item.runnerTrace?.traceId) === id) ?? null; - if (traceAuthorityIsSealed(turn, message)) return false; - if (messageIsSealedTerminal(message)) return false; - return true; - } - - function traceAuthorityIsSealed(turn: TurnStatusAuthority | undefined, message: ChatMessage | null): boolean { - const status = normalizedStatusText(turn?.status); - if (!(turn?.terminal === true || isTerminalMessageStatus(status))) return false; - return message ? messageHasTerminalResponse(message) : false; - } - - function messageIsSealedTerminal(message: ChatMessage | null | undefined): boolean { - return messageHasTerminalResponse(message); + return selectActiveTurnStatusRefreshTraceIds({ messages: source, currentRequestTraceId: currentRequest.value?.traceId, turnStatusAuthority: turnStatusAuthority.value, limit: 3 }); } async function refreshTurnStatusByTraceId(traceId: string | null | undefined, options: { force?: boolean } = {}): Promise {