diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index af350403..22ad3308 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -186,6 +186,85 @@ test("workbench read model recovers trace events from durable projection without } }); +test("workbench read model projects terminal result atomically across session, messages, and turn", async () => { + const traceStore = createCodeAgentTraceStore(); + const results = createCodeAgentChatResultStore(); + const traceId = "trc_workbench_terminal_atomic"; + const finalText = "terminal projection final response"; + const session = { + id: "ses_workbench_terminal_atomic", + projectId: "prj_hwpod_workbench", + agentId: "hwlab-code-agent", + status: "running", + ownerUserId: ACTOR.id, + conversationId: "cnv_workbench_terminal_atomic", + threadId: "thread-workbench-terminal-atomic", + lastTraceId: traceId, + updatedAt: "2026-06-17T02:10:00.000Z", + session: { + sessionStatus: "running", + lastTraceId: traceId, + messages: [ + { role: "user", text: "terminal atomic ping", traceId, status: "sent", createdAt: "2026-06-17T02:09:58.000Z" }, + { role: "agent", text: "", traceId, status: "running", createdAt: "2026-06-17T02:09:59.000Z" } + ], + valuesRedacted: true, + secretMaterialStored: false + } + }; + traceStore.append(traceId, { seq: 1, type: "request", status: "accepted", label: "request:accepted" }); + traceStore.append(traceId, { seq: 2, type: "result", status: "completed", label: "result:completed", terminal: true }); + results.set(traceId, { + status: "completed", + traceId, + ownerUserId: ACTOR.id, + conversationId: session.conversationId, + sessionId: session.id, + threadId: session.threadId, + finalResponse: { text: finalText, status: "completed", traceId }, + agentRun: { runId: "run_workbench_terminal_atomic", commandId: "cmd_workbench_terminal_atomic", status: "completed", terminalStatus: "completed" } + }); + const accessController = { + store: { + async listAgentSessionsForUser() { return [session]; }, + async getAgentSession(sessionId) { return sessionId === session.id ? session : null; }, + async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; } + }, + async ensureBootstrap() {}, + async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; } + }; + const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + + try { + const { port } = server.address(); + const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`); + assert.equal(sessions.status, 200); + assert.equal(sessions.body.sessions[0].status, "completed"); + assert.equal(sessions.body.sessions[0].running, false); + assert.equal(sessions.body.sessions[0].terminal, true); + + const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=10`); + assert.equal(messages.status, 200); + assert.equal(messages.body.total, 2); + const assistant = messages.body.messages[1]; + assert.equal(assistant.role, "agent"); + assert.equal(assistant.status, "completed"); + assert.equal(assistant.text, finalText); + assert.equal(assistant.parts[0].status, "completed"); + assert.equal(assistant.parts[0].text, finalText); + + const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`); + assert.equal(turn.status, 200); + assert.equal(turn.body.turn.status, "completed"); + assert.equal(turn.body.turn.terminal, true); + assert.equal(turn.body.turn.assistantText, finalText); + assert.equal(turn.body.turn.finalResponse.text, finalText); + } finally { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + test("workbench read model does not expose trace-only memory without visible session or result owner", async () => { const traceStore = createCodeAgentTraceStore(); const traceId = "trc_workbench_trace_only"; diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 11c7fb3d..f7c8d9d3 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -1,5 +1,5 @@ /* - * SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 + * SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 * 职责: Workbench REST read model。只读投影 session/message/turn/trace facts,不执行 AgentRun sync、billing finalize 或 workspace repair。 */ import { createHash } from "node:crypto"; @@ -244,7 +244,7 @@ async function handleWorkbenchSessionDetail(response, options, actor, sessionId) async function handleWorkbenchMessagePage(response, url, options, actor, sessionId) { const session = await visibleSessionByRouteId(options.accessController?.store, sessionId, actor); if (!session) return sendJson(response, 404, workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId })); - const messages = sessionMessages(session); + const messages = sessionMessages(session, options); const limit = boundedLimit(url.searchParams.get("limit")); const offset = cursorOffset(url.searchParams.get("cursor") ?? url.searchParams.get("after")); const page = messages.slice(offset, offset + limit); @@ -347,7 +347,7 @@ function sessionSummary(session, options) { const snapshot = objectValue(session.session); const traceId = safeTraceId(session.lastTraceId ?? snapshot.lastTraceId ?? snapshot.currentTraceId) ?? null; const trace = traceId ? traceSnapshotSync(options, traceId) : null; - const messages = sessionMessages(session); + const messages = sessionMessages(session, options); const status = canonicalWorkbenchStatus(snapshot.sessionStatus ?? session.status, trace?.status); return { sessionId: session.id, @@ -388,23 +388,81 @@ function sessionDetail(session, options) { }; } -function sessionMessages(session) { +function sessionMessages(session, options = {}) { const snapshot = objectValue(session.session); const raw = Array.isArray(snapshot.messages) ? snapshot.messages : Array.isArray(snapshot.chatMessages) ? snapshot.chatMessages : []; - const messages = raw.map((message, index) => messageFact(message, index, session, snapshot)).filter(Boolean); const finalTraceId = safeTraceId(session.lastTraceId ?? snapshot.lastTraceId ?? snapshot.currentTraceId) ?? null; + const terminalProjection = terminalMessageProjection(session, snapshot, options, finalTraceId); + const messages = raw + .map((message, index) => messageFact(message, index, session, snapshot)) + .filter(Boolean) + .map((message) => applyTerminalMessageProjection(message, terminalProjection)); const hasAssistantLikeFinal = messages.some((message) => isAssistantLikeRole(message.role) && (!finalTraceId || message.traceId === finalTraceId)); - if (!hasAssistantLikeFinal && textValue(snapshot.finalResponse)) { - messages.push(messageFact({ role: "assistant", text: snapshot.finalResponse, traceId: session.lastTraceId ?? snapshot.lastTraceId, source: "finalResponse" }, messages.length, session, snapshot)); + const finalText = terminalProjection?.text ?? projectionText(snapshot.finalResponse); + if (!hasAssistantLikeFinal && finalText) { + messages.push(messageFact({ role: "assistant", text: finalText, status: terminalProjection?.status ?? session.status, traceId: finalTraceId, source: terminalProjection?.source ?? "finalResponse" }, messages.length, session, snapshot)); } return messages; } +function terminalMessageProjection(session, snapshot, options, traceId) { + if (!traceId) return null; + const result = options.result ?? options.codeAgentChatResults?.get?.(traceId) ?? null; + const trace = options.trace ?? traceSnapshotSync(options, traceId); + const status = canonicalWorkbenchStatus( + result?.status ?? result?.agentRun?.terminalStatus ?? result?.agentRun?.commandState ?? result?.agentRun?.status ?? session?.status ?? snapshot.sessionStatus, + trace?.status + ); + if (!TERMINAL_STATUSES.has(status) || RUNNING_STATUSES.has(status)) return null; + const text = projectionText(result?.finalResponse, result?.assistantText, result?.reply, result?.text, result?.summary, trace?.finalResponse, trace?.terminalEvidence?.finalResponse, snapshot.finalResponse); + return { traceId, status, text, source: result ? "turn-result" : trace?.status && trace.status !== "missing" ? "trace-projection" : "session-snapshot" }; +} + +function applyTerminalMessageProjection(message, projection) { + if (!projection || !isAssistantLikeRole(message.role)) return message; + if (projection.traceId && message.traceId && message.traceId !== projection.traceId) return message; + const text = projection.text || message.text || ""; + const parts = projectTerminalMessageParts(message.parts, message.messageId, projection.traceId ?? message.traceId, projection, text); + return { + ...message, + traceId: projection.traceId ?? message.traceId, + turnId: message.turnId ?? projection.traceId ?? null, + status: projection.status, + parts, + text, + textPreview: text ? text.slice(0, 240) : message.textPreview ?? null, + updatedAt: message.updatedAt ?? null + }; +} + +function projectTerminalMessageParts(parts, messageId, traceId, projection, text) { + const sourceParts = Array.isArray(parts) ? parts : []; + const textIndex = sourceParts.findIndex((part) => part?.type === "text" || part?.text); + if (textIndex >= 0) { + return sourceParts.map((part, index) => index === textIndex ? { ...part, traceId, text: text || part.text || null, status: projection.status } : part); + } + if (!text) return sourceParts; + return [partFact({ type: "text", text, status: projection.status }, 0, messageId, traceId), ...sourceParts]; +} + +function projectionText(...values) { + for (const value of values) { + if (value && typeof value === "object") { + const nested = textValue(value.text ?? value.content ?? value.message ?? value.summary ?? value.preview ?? value.title); + if (nested) return nested; + continue; + } + const text = textValue(value); + if (text) return text; + } + return null; +} + function messageFact(message, index, session, snapshot) { const role = textValue(message?.role) || (index % 2 === 0 ? "user" : "assistant"); const traceId = safeTraceId(message?.traceId ?? message?.turnTraceId ?? session.lastTraceId ?? snapshot.lastTraceId) ?? null; const messageId = safeMessageId(message?.messageId ?? message?.id) || `msg_${hash(`${session.id}:${index}:${role}:${traceId ?? "none"}`).slice(0, 24)}`; - const text = textValue(message?.text ?? message?.content ?? message?.message ?? message?.finalResponse); + const text = projectionText(message?.text, message?.content, message?.message, message?.finalResponse); const rawParts = Array.isArray(message?.parts) ? message.parts : []; const parts = rawParts.length > 0 ? rawParts.map((part, partIndex) => partFact(part, partIndex, messageId, traceId)).filter(Boolean) @@ -442,9 +500,10 @@ function partFact(part, index, messageId, traceId) { } function turnSnapshot({ turnId, traceId, status, result, session, trace }) { - const messages = session ? sessionMessages(session) : []; + const messages = session ? sessionMessages(session, { result, trace }) : []; const userMessage = messages.find((message) => message.role === "user") ?? null; const assistantMessage = [...messages].reverse().find((message) => isAssistantLikeRole(message.role)) ?? null; + const finalText = projectionText(result?.finalResponse, result?.assistantText, result?.reply, result?.text, trace?.finalResponse, trace?.terminalEvidence?.finalResponse, assistantMessage?.text); return { turnId, traceId, @@ -455,6 +514,8 @@ function turnSnapshot({ turnId, traceId, status, result, session, trace }) { threadId: safeOpaqueId(result?.threadId ?? result?.session?.threadId ?? session?.threadId) ?? (textValue(session?.threadId) || null), userMessageId: userMessage?.messageId ?? null, assistantMessageId: assistantMessage?.messageId ?? null, + assistantText: finalText ?? null, + finalResponse: finalText ? { text: finalText, status, traceId, valuesPrinted: false } : null, agentRun: result?.agentRun ?? null, trace: { traceId, diff --git a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts b/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts index 8c97b8eb..a5b48243 100644 --- a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts +++ b/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts @@ -1,4 +1,4 @@ -// SPEC: PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-010403 API契约 draft-2026-06-18-r1. +// SPEC: PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1. // Responsibility: Same-origin fake Workbench API and static server for Playwright browser regression tests. import { createReadStream, existsSync, statSync } from "node:fs"; @@ -423,6 +423,7 @@ function acceptChatTurn(body: JsonRecord): JsonRecord { state.liveBackfillTraceId = traceId; state.liveBackfillReadyAtMs = Date.now() + 2_000; state.traces[traceId] = liveBackfillEarlyTrace(sessionId, threadId, traceId); + scheduleLiveBackfillProjection(sessionId, threadId, traceId, state.liveBackfillReadyAtMs); } else { state.traces[traceId] = { traceId, status: "running", sessionId, threadId, events: [], eventCount: 0, fullTraceLoaded: false, hasMore: false }; } @@ -460,9 +461,40 @@ function liveBackfillCompletedTrace(sessionId: string, threadId: string | null, return { traceId, status: "completed", sessionId, threadId, events, eventCount: events.length, fullTraceLoaded: true, hasMore: false, finalResponse: { text: finalText, status: "completed" } }; } +function scheduleLiveBackfillProjection(sessionId: string, threadId: string | null, traceId: string, readyAtMs: number): void { + const scenarioId = state.scenarioId; + setTimeout(() => { + if (state.scenarioId !== scenarioId || state.liveBackfillTraceId !== traceId) return; + commitLiveBackfillProjection(sessionId, threadId, traceId); + }, Math.max(0, readyAtMs - Date.now())); +} + +function commitLiveBackfillProjection(sessionId: string, threadId: string | null, traceId: string): void { + const completed = liveBackfillCompletedTrace(sessionId, threadId, traceId); + const finalText = typeof (completed.finalResponse as JsonRecord | undefined)?.text === "string" ? String((completed.finalResponse as JsonRecord).text) : ""; + state.traces[traceId] = completed; + const session = sessionById(sessionId); + if (!session) return; + session.status = "completed"; + session.updatedAt = new Date().toISOString(); + session.messages = (session.messages ?? []).map((message) => { + if (message.role !== "agent" || message.traceId !== traceId) return message; + return { ...message, text: finalText, status: "completed", runnerTrace: completed }; + }); +} + function sessionSummary(session: SessionRecord): SessionRecord { const { messages: _messages, hidden: _hidden, ...rest } = session; - return rest as SessionRecord; + const status = typeof rest.status === "string" ? rest.status : "unknown"; + return { ...rest, status, running: isRunningStatus(status), terminal: isTerminalStatus(status) } as SessionRecord; +} + +function isRunningStatus(status: string): boolean { + return ["active", "accepted", "pending", "running"].includes(status); +} + +function isTerminalStatus(status: string): boolean { + return ["blocked", "canceled", "cancelled", "completed", "failed", "timeout"].includes(status); } function workbenchSessionListPayload(url: URL): JsonRecord { @@ -511,8 +543,7 @@ function turnPayload(traceId: string): JsonRecord { const session = state.sessions.find((item) => item.lastTraceId === traceId) ?? state.sessions[0]; const sessionId = session?.sessionId ?? state.selectedSessionId; const threadId = typeof session?.threadId === "string" ? session.threadId : null; - if (state.liveBackfillReadyAtMs && Date.now() >= state.liveBackfillReadyAtMs) return liveBackfillCompletedTrace(sessionId, threadId, traceId); - return liveBackfillEarlyTrace(sessionId, threadId, traceId); + return state.traces[traceId] ?? liveBackfillEarlyTrace(sessionId, threadId, traceId); } const trace = state.traces[traceId] ?? { traceId, status: "unknown", events: [] }; if (state.scenarioId === "terminal-turn-stale-session-active" && traceId === "trc_completed") return { ...trace, traceId, status: "active", running: true, terminal: false, trace: { status: "completed", eventCount: trace.eventCount ?? 0 } }; diff --git a/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts b/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts index bcef3a6b..c0fcef27 100644 --- a/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-server-state.test.ts @@ -4,6 +4,7 @@ import test from "node:test"; import type { WorkbenchSessionRecord } from "../src/types/index.ts"; import { initialWorkbenchSessionIdFromLocation } from "../src/stores/workbench-projection.ts"; import { createWorkbenchServerState, reduceWorkbenchServerState, selectActiveMessages, selectActiveSession, selectTraceAuthorityById } from "../src/stores/workbench-server-state.ts"; +import { stableSessionList } from "../src/stores/workbench-session.ts"; test("Workbench route parsing accepts only session routes", () => { assert.equal(initialWorkbenchSessionIdFromLocation({ pathname: "/workbench/sessions/ses_route" }), "ses_route"); @@ -26,3 +27,32 @@ test("Workbench server-state is keyed only by session and trace", () => { assert.deepEqual(selectActiveMessages(state, "ses_route").map((message) => message.id), ["msg_route"]); assert.equal(selectTraceAuthorityById(state).trc_route?.traceId, "trc_route"); }); + +test("Workbench session list keeps selected detail but lets server summary override stale running status", () => { + const selected: WorkbenchSessionRecord = { + sessionId: "ses_live", + threadId: "thr_live", + status: "running", + lastTraceId: "trc_live", + messageCount: 2, + messages: [ + { id: "msg_live_user", messageId: "msg_live_user", role: "user", title: "用户", text: "ping", status: "sent", createdAt: "2026-06-18T00:00:00.000Z", sessionId: "ses_live", traceId: "trc_live" }, + { id: "msg_live_agent", messageId: "msg_live_agent", role: "agent", title: "Code Agent", text: "", status: "running", createdAt: "2026-06-18T00:00:01.000Z", sessionId: "ses_live", traceId: "trc_live" } + ] + }; + const serverSummary: WorkbenchSessionRecord = { + sessionId: "ses_live", + threadId: "thr_live", + status: "completed", + lastTraceId: "trc_live", + messageCount: 2, + updatedAt: "2026-06-18T00:00:03.000Z" + }; + + const next = stableSessionList([selected], [serverSummary], "ses_live", selected); + + assert.equal(next[0]?.sessionId, "ses_live"); + assert.equal(next[0]?.status, "completed"); + assert.equal(next[0]?.updatedAt, "2026-06-18T00:00:03.000Z"); + assert.equal(next[0]?.messages?.length, 2); +}); diff --git a/web/hwlab-cloud-web/src/stores/workbench-session.ts b/web/hwlab-cloud-web/src/stores/workbench-session.ts index 73873171..7b81eeaa 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session.ts @@ -1,4 +1,4 @@ -// SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1. +// SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection. // Responsibility: Session-first Workbench UI helpers for composer state, tab projection, and local list stability. import type { ChatMessage, ProviderProfile, WorkbenchSessionRecord } from "@/types"; @@ -170,7 +170,7 @@ export function stableSessionList(current: WorkbenchSessionRecord[], candidate: : current.length > 0 ? current : candidateList; - return selected ? mergeSessionIntoList(stable, selected) : stable.filter((session) => !isArchivedSession(session)); + return selected ? mergeSelectedSessionDetail(stable, selected) : stable.filter((session) => !isArchivedSession(session)); } export function mergeSessionIntoList(current: WorkbenchSessionRecord[], session: WorkbenchSessionRecord | null | undefined): WorkbenchSessionRecord[] { @@ -238,6 +238,14 @@ function mergeSessionRecord(existing: WorkbenchSessionRecord, incoming: Workbenc }; } +function mergeSelectedSessionDetail(stable: WorkbenchSessionRecord[], selected: WorkbenchSessionRecord): WorkbenchSessionRecord[] { + const selectedId = firstNonEmptyString(selected.sessionId); + if (!selectedId) return stable.filter((session) => !isArchivedSession(session)); + const existingIndex = stable.findIndex((session) => session.sessionId === selectedId); + if (existingIndex < 0) return mergeSessionIntoList(stable, selected); + return stable.map((session, index) => index === existingIndex ? mergeSessionRecord(selected, session) : session).filter((session) => !isArchivedSession(session)); +} + function latestAgentMessage(messages: ChatMessage[] | undefined): ChatMessage | null { return [...(messages ?? [])].reverse().find((message) => message.role === "agent") ?? null; } diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index da29bc12..55fcbc77 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -1,4 +1,4 @@ -// SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. +// SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. // Responsibility: Session-first Workbench state orchestration for selection, turn admission, and trace lifecycle rendering. import { computed, ref } from "vue"; @@ -19,6 +19,8 @@ const TRACE_HYDRATION_PAGE_LIMIT = 100; const TRACE_HYDRATION_MAX_PAGES = 60; const TRACE_HYDRATION_MAX_ATTEMPTS = 3; const TRACE_HYDRATION_RETRY_DELAY_MS = 700; +const PROJECTION_CATCHUP_DELAY_MS = 1_000; +const PROJECTION_CATCHUP_MAX_ATTEMPTS = 120; interface HydrateOptions { sessionId?: string | null; @@ -50,9 +52,11 @@ export const useWorkbenchStore = defineStore("workbench", () => { const traceAuthorityById = computed(() => selectTraceAuthorityById(serverState.value)); const traceHydrationInFlight = new Set(); const terminalRealtimeRefreshInFlight = new Set(); + const projectionCatchupAttempts = new Map(); let realtimeStream: WorkbenchEventStream | null = null; let realtimeKey = ""; let realtimeGapTimer: number | null = null; + let projectionCatchupTimer: number | null = null; const activeSession = computed(() => explicitSessionId.value ? sessions.value.find((item) => item.sessionId === explicitSessionId.value) ?? null : null); const selectedSessionId = computed(() => firstNonEmptyString(activeSession.value?.sessionId, explicitSessionId.value)); @@ -246,6 +250,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const response = await api.agent.getAgentTurn(id, 8000, () => activityRef.value); if (response.ok && response.data) { applyTurnStatusSnapshot(id, response.data); + if (response.data.terminal === true || isTerminalMessageStatus(response.data.status)) completeTrace(id, response.data); return; } reduceServerState({ type: "turn.forget", traceId: id }); @@ -332,11 +337,23 @@ export const useWorkbenchStore = defineStore("workbench", () => { return; } markWorkbenchSubmitApiAccepted(traceId); - alignOptimisticTurnMessages(traceId, response.data); - applyTurnStatusSnapshot(traceId, response.data); + const canonicalTraceId = alignOptimisticTurnMessages(traceId, response.data); + if (canonicalTraceId !== traceId) { + reduceServerState({ type: "turn.forget", traceId }); + if (currentRequest.value?.traceId === traceId) { + currentRequest.value = { + ...currentRequest.value, + traceId: canonicalTraceId, + sessionId: firstNonEmptyString(response.data.sessionId, currentRequest.value.sessionId), + threadId: firstNonEmptyString(response.data.threadId, currentRequest.value.threadId), + status: response.data.status ?? currentRequest.value.status + }; + } + } + applyTurnStatusSnapshot(canonicalTraceId, response.data); void refreshSessions(sessionId); if ((response.data as AgentChatResultResponse).terminal === true || isTerminalMessageStatus(response.data.status)) { - completeTrace(traceId, response.data as AgentChatResultResponse); + completeTrace(canonicalTraceId, response.data as AgentChatResultResponse); return; } restartRealtime("submit"); @@ -641,6 +658,44 @@ export const useWorkbenchStore = defineStore("workbench", () => { hydrateTraceEvents(messages.value) ]); restartRealtime(`gap:${reason}`); + scheduleProjectionCatchup(reason); + } + + function scheduleProjectionCatchup(reason: string): void { + const traceId = activeProjectionCatchupTraceId(); + if (!traceId) { + clearProjectionCatchup(); + return; + } + if (projectionCatchupTimer) return; + const attempts = projectionCatchupAttempts.get(traceId) ?? 0; + if (attempts >= PROJECTION_CATCHUP_MAX_ATTEMPTS) return; + projectionCatchupAttempts.set(traceId, attempts + 1); + if (typeof window === "undefined") return; + projectionCatchupTimer = window.setTimeout(() => { + projectionCatchupTimer = null; + void hydrateRealtimeGap(`projection-catchup:${reason}`); + }, PROJECTION_CATCHUP_DELAY_MS); + } + + function activeProjectionCatchupTraceId(): string | null { + const traceId = realtimeTraceId(); + if (!traceId) return null; + const turn = turnStatusAuthority.value[traceId]; + const message = [...messages.value].reverse().find((item) => item.role === "agent" && firstNonEmptyString(item.traceId, item.runnerTrace?.traceId) === traceId) ?? null; + const active = currentRequest.value?.traceId === traceId + || turn?.running === true + || isTraceActiveStatus(turn?.status) + || isTraceActiveStatus(message?.status) + || isTraceActiveStatus(message?.runnerTrace?.status); + return active ? traceId : null; + } + + function clearProjectionCatchup(traceId?: string | null): void { + if (traceId) projectionCatchupAttempts.delete(traceId); + else projectionCatchupAttempts.clear(); + if (typeof window !== "undefined" && projectionCatchupTimer) window.clearTimeout(projectionCatchupTimer); + projectionCatchupTimer = null; } function installRealtimeVisibilityHandler(): void { @@ -676,6 +731,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { } function completeTrace(traceId: string, result: AgentChatResultResponse): void { + clearProjectionCatchup(traceId); const resultTrace = recordValue(result.runnerTrace); const traceAssistantText = assistantTextFromTraceEvents(firstArray(result.events, result.traceEvents, resultTrace?.events)); const text = firstNonEmptyString(result.assistantText, finalResponseText(result.finalResponse), typeof result.reply === "string" ? result.reply : result.reply?.content, agentErrorDisplayText(result.error), result.text, result.summary, traceAssistantText) ?? "Code Agent 已完成,但没有返回可展示的 final response。"; @@ -755,22 +811,27 @@ export const useWorkbenchStore = defineStore("workbench", () => { messages.value = messages.value.map((message) => message.traceId === traceId && message.role === "agent" ? { ...message, ...patch, updatedAt: new Date().toISOString() } : message); } - function alignOptimisticTurnMessages(traceId: string, lifecycle: AgentChatResponse): void { - const turnId = firstNonEmptyString(lifecycle.turnId, traceId); + function alignOptimisticTurnMessages(traceId: string, lifecycle: AgentChatResponse): string { + const canonicalTraceId = firstNonEmptyString(lifecycle.traceId, traceId) ?? traceId; + const turnId = firstNonEmptyString(lifecycle.turnId, canonicalTraceId); const userMessageId = firstNonEmptyString(lifecycle.userMessageId); const assistantMessageId = firstNonEmptyString(lifecycle.assistantMessageId); - if (!turnId && !userMessageId && !assistantMessageId) return; + if (!turnId && !userMessageId && !assistantMessageId && canonicalTraceId === traceId) return canonicalTraceId; messages.value = messages.value.map((message) => { if (message.traceId !== traceId) return message; - if (message.role === "user" && userMessageId) return { ...message, id: userMessageId, messageId: userMessageId, turnId, updatedAt: new Date().toISOString() }; - if (message.role === "agent" && assistantMessageId) return { ...message, id: assistantMessageId, messageId: assistantMessageId, turnId, updatedAt: new Date().toISOString() }; - return turnId ? { ...message, turnId, updatedAt: new Date().toISOString() } : message; + const runnerTrace = message.runnerTrace ? { ...message.runnerTrace, traceId: canonicalTraceId } : message.runnerTrace; + const base = { ...message, traceId: canonicalTraceId, runnerTrace, updatedAt: new Date().toISOString() }; + if (message.role === "user" && userMessageId) return { ...base, id: userMessageId, messageId: userMessageId, turnId }; + if (message.role === "agent" && assistantMessageId) return { ...base, id: assistantMessageId, messageId: assistantMessageId, turnId }; + return turnId ? { ...base, turnId } : base; }); + return canonicalTraceId; } async function clearActiveTrace(traceId: string, reason: string): Promise { void reason; if (currentRequest.value?.traceId === traceId) currentRequest.value = null; + if (!activeProjectionCatchupTraceId()) clearProjectionCatchup(traceId); } function beginSessionSelection(): number { diff --git a/web/hwlab-cloud-web/tests/workbench-e2e/specs/live-backfill-race.spec.ts b/web/hwlab-cloud-web/tests/workbench-e2e/specs/live-backfill-race.spec.ts index 4f8477dd..796f7f04 100644 --- a/web/hwlab-cloud-web/tests/workbench-e2e/specs/live-backfill-race.spec.ts +++ b/web/hwlab-cloud-web/tests/workbench-e2e/specs/live-backfill-race.spec.ts @@ -20,8 +20,22 @@ test.describe("live REST backfill without terminal SSE", () => { return response.ok() ? ((await response.json()).turn?.status as string | undefined) : undefined; }).toBe("completed"); + const listResponse = await page.request.get("/v1/workbench/sessions?includeSessionId=ses_completed"); + const messagesResponse = await page.request.get("/v1/workbench/sessions/ses_completed/messages?limit=100"); + expect(listResponse.ok()).toBe(true); + expect(messagesResponse.ok()).toBe(true); + const listPayload = await listResponse.json() as { sessions?: Array<{ sessionId?: string; status?: string; running?: boolean }> }; + const messagesPayload = await messagesResponse.json() as { messages?: Array<{ role?: string; traceId?: string; status?: string; text?: string }> }; + const sessionSummary = listPayload.sessions?.find((session) => session.sessionId === "ses_completed"); + const assistantMessage = messagesPayload.messages?.find((message) => message.role === "agent" && message.traceId === "trc_live_backfill"); + expect(sessionSummary?.status).toBe("completed"); + expect(sessionSummary?.running).toBe(false); + expect(assistantMessage?.status).toBe("completed"); + expect(assistantMessage?.text).toContain("fake AgentRun completed after REST backfill"); + await saveScreenshot(page, testInfo, "live-backfill-canonical-completed"); await expect(page.locator(sessionTab("ses_completed"))).toHaveAttribute("data-active", "true"); + await expect(page.locator(sessionTab("ses_completed"))).toHaveAttribute("data-status", "completed"); const backfilledFinal = page.locator(`${selectors.messageCard}[data-role="agent"]`).filter({ hasText: "fake AgentRun completed after REST backfill" }); await expect(backfilledFinal).toHaveAttribute("data-status", "completed"); await expect(page.locator(`${selectors.traceTimeline}[data-status="completed"]`).last()).toBeVisible();