From d608c272eaaf9f91d1dc9db1c5a3c470b78f8630 Mon Sep 17 00:00:00 2001 From: Lyon <88232613+pikasTech@users.noreply.github.com> Date: Thu, 25 Jun 2026 13:12:42 +0800 Subject: [PATCH] fix: seal terminal workbench timing (#2118) --- .../cloud/workbench-projection-writer.test.ts | 64 +++++++++++++ internal/cloud/workbench-projection-writer.ts | 13 +-- internal/cloud/workbench-turn-projection.ts | 9 +- .../scripts/workbench-e2e-server.ts | 90 ++++++++++++++++++- .../src/stores/workbench-server-state.ts | 56 +++++++++++- .../specs/message-timing.spec.ts | 22 +++++ 6 files changed, 239 insertions(+), 15 deletions(-) diff --git a/internal/cloud/workbench-projection-writer.test.ts b/internal/cloud/workbench-projection-writer.test.ts index a1598d3c..fa05122d 100644 --- a/internal/cloud/workbench-projection-writer.test.ts +++ b/internal/cloud/workbench-projection-writer.test.ts @@ -90,6 +90,70 @@ test("workbench projection writer commits terminal owner evidence as sealed dura assert.equal(facts.checkpoints[0].timing.finishedAt, "2026-06-20T11:00:00.000Z"); }); +test("workbench projection writer does not synthesize terminal duration from updatedAt", async () => { + const factWrites = []; + const runtimeStore = { + async writeWorkbenchFacts(params, requestMeta) { + factWrites.push({ params, requestMeta }); + return { written: true, facts: params.facts }; + } + }; + const accessController = { + async recordAgentSessionOwner(input) { + return { + id: input.sessionId, + projectId: input.projectId, + ownerUserId: input.ownerUserId, + conversationId: input.conversationId, + threadId: input.threadId, + lastTraceId: input.traceId, + status: input.status, + session: input.session, + updatedAt: "2026-06-20T11:05:00.000Z" + }; + } + }; + + await writeWorkbenchProjectionSession({ + accessController, + runtimeStore, + traceId: "trc_writer_terminal_no_finished_at", + ownerUserId: "usr_writer", + ownerRole: "user", + sessionId: "ses_writer_terminal_no_finished_at", + projectId: "prj_writer", + conversationId: "cnv_writer_terminal_no_finished_at", + threadId: "thread-writer-terminal-no-finished-at", + status: "canceled", + payload: { + traceId: "trc_writer_terminal_no_finished_at", + status: "canceled", + startedAt: "2026-06-20T11:00:00.000Z", + updatedAt: "2026-06-20T11:05:00.000Z", + agentRun: { runId: "run_writer_terminal_no_finished_at", commandId: "cmd_writer_terminal_no_finished_at", terminalStatus: "canceled" } + }, + session: { + sessionStatus: "canceled", + messages: [ + { messageId: "msg_writer_terminal_no_finished_user", role: "user", text: "cancel me", status: "sent", turnId: "trc_writer_terminal_no_finished_at", traceId: "trc_writer_terminal_no_finished_at" }, + { messageId: "msg_writer_terminal_no_finished_agent", role: "agent", text: "hwlab-user-cancel", status: "canceled", turnId: "trc_writer_terminal_no_finished_at", traceId: "trc_writer_terminal_no_finished_at" } + ] + } + }); + + assert.equal(factWrites.length, 1); + const facts = factWrites[0].params.facts; + const agentMessage = facts.messages.find((message) => message.messageId === "msg_writer_terminal_no_finished_agent"); + assert.equal(facts.turns[0].status, "canceled"); + assert.equal(facts.turns[0].finishedAt, null); + assert.equal(facts.turns[0].durationMs, null); + assert.equal(agentMessage.finishedAt, null); + assert.equal(agentMessage.durationMs, null); + assert.equal(facts.checkpoints[0].timing.finishedAt, null); + assert.equal(facts.checkpoints[0].timing.durationMs, null); + assert.equal(facts.checkpoints[0].diagnostic.blocker.code, "workbench_terminal_timing_authority_missing"); +}); + test("workbench projection writer keeps event writes durable when checkpoint reads are blocked", async () => { const factWrites = []; const baseStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:30:00.000Z" }); diff --git a/internal/cloud/workbench-projection-writer.ts b/internal/cloud/workbench-projection-writer.ts index d5489126..40879f4e 100644 --- a/internal/cloud/workbench-projection-writer.ts +++ b/internal/cloud/workbench-projection-writer.ts @@ -477,8 +477,7 @@ function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, own const projectedStatus = normalizeWorkbenchStatus(projection.status ?? normalizedStatus); const terminal = projection.terminal === true; const terminalStatus = terminal ? (TERMINAL_STATUSES.has(projectedStatus) ? projectedStatus : normalizedStatus) : projectedStatus; - const projectedAt = new Date().toISOString(); - const timing = terminalTimingAtLeastProjectedAt(projectionTimingForStatus(projection.timing, terminal), terminal, projectedAt); + const timing = terminalTimingAtLeastProjectedAt(projectionTimingForStatus(projection.timing, terminal), terminal); const timingAuthorityIssue = terminalTimingAuthorityIssue(timing, { terminal, traceId: safeId, source: "facts", status: terminalStatus, label: payload?.lastEventLabel, sourceSeq: projection.lastProjectedSeq }); const baseDiagnostic = projectionDiagnostics({ traceId: safeId, result: payload, trace: payload?.runnerTrace ?? null, projection }); const diagnostic = timingAuthorityIssue @@ -772,14 +771,15 @@ function projectionTimingForStatus(value, terminal) { return { ...timing, finishedAt: null, durationMs: null, valuesRedacted: timing.valuesRedacted !== false }; } -function terminalTimingAtLeastProjectedAt(value, terminal, projectedAt) { +function terminalTimingAtLeastProjectedAt(value, terminal) { const timing = normalizeTimingProjection(value) ?? emptyTimingProjection(); if (!terminal) return timing; - const finishedAt = timing.finishedAt ?? timing.lastEventAt ?? optionalTimestampValue(projectedAt); + const finishedAt = timing.finishedAt ?? null; + const durationMs = durationValue(timing.durationMs) ?? elapsedBetween(timing.startedAt, finishedAt); return { ...timing, finishedAt, - durationMs: elapsedBetween(timing.startedAt, finishedAt) ?? timing.durationMs, + durationMs, valuesRedacted: timing.valuesRedacted !== false }; } @@ -801,7 +801,7 @@ function eventTimingProjection({ startedAt = null, lastEventAt = null, finishedA startedAt: normalizedStartedAt, lastEventAt: normalizedLastEventAt, finishedAt: normalizedFinishedAt, - durationMs: terminal ? elapsedBetween(normalizedStartedAt, normalizedFinishedAt ?? normalizedLastEventAt) ?? normalizedDurationMs : null, + durationMs: terminal ? normalizedDurationMs ?? elapsedBetween(normalizedStartedAt, normalizedFinishedAt ?? normalizedLastEventAt) : null, valuesRedacted: true }; } @@ -822,6 +822,7 @@ function emptyTimingProjection() { } function durationValue(value) { + if (value === null || value === undefined || value === "") return null; const number = Number(value); return Number.isFinite(number) && number >= 0 ? Math.trunc(number) : null; } diff --git a/internal/cloud/workbench-turn-projection.ts b/internal/cloud/workbench-turn-projection.ts index bd308c92..b9a74172 100644 --- a/internal/cloud/workbench-turn-projection.ts +++ b/internal/cloud/workbench-turn-projection.ts @@ -88,11 +88,9 @@ export function createWorkbenchTurnTimingProjection({ result = null, session = n trace?.terminalEvidence?.updatedAt, terminalEvent?.updatedAt, terminalEvent?.createdAt, - terminalEvent?.occurredAt, - result?.updatedAt, - trace?.updatedAt, - lastEventAt + terminalEvent?.occurredAt ) : null; + const elapsedDurationMs = isTerminal ? elapsedBetween(startedAt, finishedAt) : elapsedBetween(startedAt, lastEventAt); const durationMs = durationValue( result?.durationMs, directTiming?.durationMs, @@ -100,7 +98,7 @@ export function createWorkbenchTurnTimingProjection({ result = null, session = n traceSummary?.durationMs, trace?.elapsedMs, result?.elapsedMs, - elapsedBetween(startedAt, finishedAt ?? lastEventAt) + elapsedDurationMs ); return { startedAt, @@ -410,6 +408,7 @@ function latestTimestamp(...values) { function durationValue(...values) { let max = null; for (const value of values) { + if (value === null || value === undefined || value === "") continue; const number = Number(value); if (!Number.isFinite(number) || number < 0) continue; const duration = Math.trunc(number); diff --git a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts b/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts index f3162950..9b4758e4 100644 --- a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts +++ b/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts @@ -670,6 +670,7 @@ function createScenarioState(scenarioId: string): ScenarioState { if (id === "progress-only-final-response") markRunningProgressOnly(sessions, traces); if (id === "terminal-completed-no-final-response") markRunningNoFinalResponse(sessions, traces); if (id === "tool-completed-projection-running") markToolCompletedProjectionRunning(sessions, traces); + if (id === "terminal-canceled-late-timing") markTerminalCanceledLateTiming(sessions, traces); if (id === "projection-degraded-diagnostics") { sessions.unshift(projectionDegradedSession()); traces.trc_projection_degraded = projectionDegradedTrace(); @@ -721,7 +722,7 @@ function createScenarioState(scenarioId: string): ScenarioState { ? "ses_markdown_final" : id === "terminal-assistant-event-final-response" ? "ses_terminal_assistant_final" - : id === "progress-only-final-response" || id === "terminal-completed-no-final-response" || id === "tool-completed-projection-running" + : id === "progress-only-final-response" || id === "terminal-completed-no-final-response" || id === "tool-completed-projection-running" || id === "terminal-canceled-late-timing" ? "ses_running" : id === "projector-resume-from-checkpoint" ? "ses_317f78e1-ed91-4b09-a72a-3668d5469f59" @@ -752,7 +753,7 @@ function createScenarioState(scenarioId: string): ScenarioState { sessionDelayMs: id === "loading" ? 2_500 : 0, sessionDetailDelayMs: id === "legacy-cnv-deeplink-canonical" ? 1_500 : id === "session-switch-delayed-detail-frame" ? 900 : id === "submit-authority-race" ? 1_000 : 0, chatDelayMs: id === "submit-authority-race" ? 1_000 : 0, - terminalScript: id === "event-replay" || id === "running-to-terminal" || id === "stale-submit-restore" || id === "progress-only-final-response" || id === "terminal-completed-no-final-response", + terminalScript: id === "event-replay" || id === "running-to-terminal" || id === "stale-submit-restore" || id === "progress-only-final-response" || id === "terminal-completed-no-final-response" || id === "terminal-canceled-late-timing", terminalFailureScript: id === "stale-submit-restore", staleTraceId, liveBackfillTraceId: id === "projector-resume-from-checkpoint" ? "trc_f2456023233a485b" : null, @@ -889,6 +890,75 @@ function markRunningNoFinalResponse(sessions: SessionRecord[], traces: Record): void { + const trace = terminalCanceledLateTimingRunningTrace(); + traces.trc_running = trace; + const session = sessions.find((item) => item.sessionId === "ses_running"); + if (!session) return; + session.status = "running"; + session.lastTraceId = "trc_running"; + session.firstUserMessagePreview = "terminal canceled timing should stay sealed"; + session.messages = (session.messages ?? []).map((message) => { + if (message.role === "user") return { ...message, text: "terminal canceled timing should stay sealed" }; + if (message.role !== "agent" || message.traceId !== "trc_running") return message; + return { ...message, text: "", status: "running", timing: trace.timing, startedAt: trace.startedAt, lastEventAt: trace.lastEventAt, finishedAt: null, durationMs: null, runnerTrace: trace }; + }); +} + +function terminalCanceledLateTimingRunningTrace(): JsonRecord { + const startedAt = "2026-06-17T09:39:00.000Z"; + const lastEventAt = "2026-06-17T09:39:04.000Z"; + const timing = { startedAt, lastEventAt, finishedAt: null, durationMs: null, valuesRedacted: true }; + return { + traceId: "trc_running", + status: "running", + sessionId: "ses_running", + threadId: "thr_running", + turnId: "turn_running", + timing, + startedAt, + lastEventAt, + finishedAt: null, + durationMs: null, + events: [{ seq: 1, createdAt: lastEventAt, label: "agentrun:backend:turn/running", type: "backend_status", status: "running" }], + eventCount: 1, + fullTraceLoaded: false, + hasMore: false + }; +} + +function applyTerminalCanceledLateTiming(durationMs: number, finishedAt: string): JsonRecord | null { + const startedAt = "2026-06-17T09:39:00.000Z"; + const timing = { startedAt, lastEventAt: finishedAt, finishedAt, durationMs, valuesRedacted: true }; + const event = { seq: 2, createdAt: finishedAt, label: "agentrun:terminal:canceled", type: "backend_status", status: "canceled", terminal: true }; + const trace = { + traceId: "trc_running", + status: "canceled", + sessionId: "ses_running", + threadId: "thr_running", + turnId: "turn_running", + timing, + startedAt, + lastEventAt: finishedAt, + finishedAt, + durationMs, + events: [event], + eventCount: 2, + fullTraceLoaded: true, + hasMore: false, + finalResponse: { text: "hwlab-user-cancel", status: "canceled" } + }; + state.traces.trc_running = trace; + const session = sessionById("ses_running"); + if (!session) return null; + session.status = "canceled"; + session.updatedAt = finishedAt; + session.messages = (session.messages ?? []).map((message) => message.role === "agent" && message.traceId === "trc_running" + ? { ...message, text: "hwlab-user-cancel", status: "canceled", timing, startedAt, lastEventAt: finishedAt, finishedAt, durationMs, runnerTrace: trace } + : message); + return trace; +} + function markToolCompletedProjectionRunning(sessions: SessionRecord[], traces: Record): void { const trace = toolCompletedProjectionRunningTrace(); traces.trc_running = trace; @@ -1778,6 +1848,22 @@ function sse(request: IncomingMessage, response: ServerResponse, url: URL): void const scenarioId = state.scenarioId; setTimeout(() => { if (state.scenarioId !== scenarioId) return; + if (state.scenarioId === "terminal-canceled-late-timing") { + const trace = applyTerminalCanceledLateTiming(12_000, "2026-06-17T09:39:12.000Z"); + const event = Array.isArray(trace?.events) ? trace.events.at(-1) as JsonRecord : null; + if (event) writeSse(response, "workbench.trace.event", { type: "trace.event", sessionId: "ses_running", threadId: "thr_running", traceId: "trc_running", event, snapshot: trace }); + const message = runningAgentMessageSnapshot(); + if (message) writeSse(response, "workbench.message.snapshot", { type: "message.snapshot", sessionId: "ses_running", threadId: "thr_running", traceId: "trc_running", message }); + writeSse(response, "workbench.turn.snapshot", { type: "turn.snapshot", sessionId: "ses_running", threadId: "thr_running", traceId: "trc_running", turn: turnPayload("trc_running") }); + setTimeout(() => { + if (state.scenarioId !== scenarioId) return; + applyTerminalCanceledLateTiming(14_000, "2026-06-17T09:39:14.000Z"); + const late = runningAgentMessageSnapshot(); + if (late) writeSse(response, "workbench.message.snapshot", { type: "message.snapshot", sessionId: "ses_running", threadId: "thr_running", traceId: "trc_running", message: late }); + writeSse(response, "workbench.turn.snapshot", { type: "turn.snapshot", sessionId: "ses_running", threadId: "thr_running", traceId: "trc_running", turn: turnPayload("trc_running") }); + }, 650); + return; + } if (state.scenarioId === "terminal-completed-no-final-response") { const trace = completedNoFinalTrace(); const event = (trace.events as JsonRecord[]).at(-1) ?? { status: "completed", terminal: true }; 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 626dc9fc..5f9b9b49 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-server-state.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-server-state.ts @@ -92,7 +92,8 @@ function reduceSessionDetail(state: WorkbenchServerState, session: WorkbenchSess const sessionId = session?.sessionId; if (!sessionId || !session) return state; const existing = state.sessionsById[sessionId]; - const messages = Array.isArray(session.messages) ? session.messages : state.messagesBySessionId[sessionId] ?? existing?.messages ?? []; + const existingMessages = state.messagesBySessionId[sessionId] ?? existing?.messages ?? []; + const messages = Array.isArray(session.messages) ? mergeMessageList(existingMessages, session.messages) : existingMessages; const merged = mergeSessionRecord(existing, { ...session, messages }); const sessionStatus = sessionStatusAuthorityFromDetail(session); return { @@ -163,13 +164,64 @@ function messageMatchesSnapshot(existing: ChatMessage, incoming: ChatMessage): b } function mergeMessageSnapshot(existing: ChatMessage, incoming: ChatMessage): ChatMessage { - return { + const merged = { ...existing, ...incoming, runnerTrace: incoming.runnerTrace ?? existing.runnerTrace ?? null, traceAutoLifecycle: existing.traceAutoLifecycle, updatedAt: incoming.updatedAt ?? existing.updatedAt }; + return sealExistingTerminalMessageTiming(existing, merged); +} + +function mergeMessageList(existing: ChatMessage[], incoming: ChatMessage[]): ChatMessage[] { + return incoming.map((message) => { + const previous = existing.find((item) => messageMatchesSnapshot(item, message)); + return previous ? sealExistingTerminalMessageTiming(previous, message) : message; + }); +} + +function sealExistingTerminalMessageTiming(existing: ChatMessage, incoming: ChatMessage): ChatMessage { + if (!isTerminalMessageStatus(existing.status)) return incoming; + const timing = terminalTimingProjection(existing); + if (!timing || timing.durationMs == null) return incoming; + const patch: Partial = { + status: existing.status, + timing, + startedAt: timing.startedAt ?? null, + lastEventAt: timing.lastEventAt ?? null, + finishedAt: timing.finishedAt ?? null, + durationMs: timing.durationMs ?? null, + traceAutoLifecycle: existing.traceAutoLifecycle ?? "terminal" + }; + if (typeof existing.text === "string" && existing.text.trim()) patch.text = existing.text; + return { ...incoming, ...patch }; +} + +function terminalTimingProjection(message: ChatMessage): ChatMessage["timing"] | null { + const timing = message.timing && typeof message.timing === "object" ? message.timing : null; + const startedAt = timestampOrNull(timing?.startedAt ?? message.startedAt); + const lastEventAt = timestampOrNull(timing?.lastEventAt ?? message.lastEventAt); + const finishedAt = timestampOrNull(timing?.finishedAt ?? message.finishedAt); + const durationMs = nonNegativeDuration(timing?.durationMs ?? message.durationMs); + if (!startedAt && !lastEventAt && !finishedAt && durationMs == null) return null; + return { ...(timing ?? {}), startedAt, lastEventAt, finishedAt, durationMs, valuesRedacted: timing?.valuesRedacted !== false }; +} + +function isTerminalMessageStatus(value: unknown): boolean { + return ["completed", "failed", "blocked", "timeout", "canceled", "cancelled", "stale", "thread-resume-failed"].includes(String(value ?? "").trim().toLowerCase().replace(/_/gu, "-")); +} + +function timestampOrNull(value: unknown): string | null { + if (typeof value !== "string") return null; + const text = value.trim(); + return text && Number.isFinite(Date.parse(text)) ? text : null; +} + +function nonNegativeDuration(value: unknown): number | null { + if (value === null || value === undefined || value === "") return null; + const number = Number(value); + return Number.isFinite(number) && number >= 0 ? Math.trunc(number) : null; } function mergeSessionRecord(existing: WorkbenchSessionRecord | undefined, incoming: WorkbenchSessionRecord): WorkbenchSessionRecord { diff --git a/web/hwlab-cloud-web/tests/workbench-e2e/specs/message-timing.spec.ts b/web/hwlab-cloud-web/tests/workbench-e2e/specs/message-timing.spec.ts index 0495d0e0..062a710b 100644 --- a/web/hwlab-cloud-web/tests/workbench-e2e/specs/message-timing.spec.ts +++ b/web/hwlab-cloud-web/tests/workbench-e2e/specs/message-timing.spec.ts @@ -38,8 +38,30 @@ test("Code Agent message headers render canonical timing metadata", async ({ pag const canceled = page.locator(`${selectors.messageCard}[data-role="agent"][data-status="canceled"]`).first(); await expect(canceled.locator(selectors.messageDurationMeta)).toContainText("耗时 0 秒"); await expect(canceled.locator(selectors.messageActivityMeta)).toHaveCount(0); + await page.clock.fastForward(60000); + await expect(canceled.locator(selectors.messageDurationMeta)).toContainText("耗时 0 秒"); await expect(page.locator(`${selectors.messageCard}[data-role="user"] ${selectors.messageDurationMeta}`)).toHaveCount(0); await expect(page.locator(`${selectors.messageCard}[data-role="user"] ${selectors.messageActivityMeta}`)).toHaveCount(0); await saveScreenshot(page, testInfo, "message-timing-headers"); }); + +test.describe("terminal timing seal", () => { + test.use({ scenarioId: "terminal-canceled-late-timing" }); + + test("canceled message duration stays sealed after late timing snapshots", async ({ page }, testInfo) => { + const baseTime = new Date("2026-06-17T09:39:04.000Z"); + await page.clock.install({ time: baseTime }); + await page.clock.pauseAt(baseTime); + await gotoWorkbench(page, "/workbench/sessions/ses_running"); + + const canceled = page.locator(`${selectors.messageCard}[data-role="agent"][data-status="canceled"]`).first(); + await expect(canceled.locator(selectors.messageDurationMeta)).toContainText("耗时 12 秒"); + await expect(canceled.locator(selectors.messageActivityMeta)).toHaveCount(0); + + await page.waitForTimeout(1200); + await expect(canceled.locator(selectors.messageDurationMeta)).toContainText("耗时 12 秒"); + await expect(canceled.locator(selectors.messageDurationMeta)).not.toContainText("14 秒"); + await saveScreenshot(page, testInfo, "message-timing-canceled-sealed"); + }); +});