fix: seal terminal workbench timing (#2118)

2026-06-25 13:12:42 +08:00
parent 77979025e7
commit d608c272ea
6 changed files with 239 additions and 15 deletions
@@ -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" });
@@ -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;
 }
@@ -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);
@@ -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<st
  });
 }

+function markTerminalCanceledLateTiming(sessions: SessionRecord[], traces: Record<string, JsonRecord>): 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<string, JsonRecord>): 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 };
@@ -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<ChatMessage> = {
+    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 {
@@ -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");
+  });
+});