From e1899dab91f6c870c197f0cf703122ba4b8e9fe2 Mon Sep 17 00:00:00 2001 From: lyon Date: Wed, 17 Jun 2026 21:18:55 +0800 Subject: [PATCH] fix: freeze terminal code agent traces --- internal/cloud/code-agent-trace-store.test.ts | 68 ++++++++++++++ internal/cloud/code-agent-trace-store.ts | 91 ++++++++++++++++--- 2 files changed, 148 insertions(+), 11 deletions(-) diff --git a/internal/cloud/code-agent-trace-store.test.ts b/internal/cloud/code-agent-trace-store.test.ts index c1a7c3ca..fc7e944e 100644 --- a/internal/cloud/code-agent-trace-store.test.ts +++ b/internal/cloud/code-agent-trace-store.test.ts @@ -101,3 +101,71 @@ test("code agent trace store preserves upstream event timestamps", () => { assert.equal(snapshot.events[0].backend, "agentrun-v01/codex"); assert.equal(snapshot.elapsedMs, 0); }); + +test("code agent trace store anchors trace time to replayed upstream events", () => { + const traceStore = createCodeAgentTraceStore({ maxEvents: 20 }); + const traceId = "trc_trace-store-replayed-upstream-time"; + traceStore.ensure(traceId, { now: "2026-06-17T13:03:50.000Z" }); + traceStore.append(traceId, { + type: "backend", + status: "running", + label: "agentrun:backend:run-created", + createdAt: "2026-06-17T12:56:13.650Z" + }, { now: "2026-06-17T13:03:51.000Z" }); + traceStore.append(traceId, { + type: "assistant", + status: "completed", + label: "agentrun:assistant:message", + message: "OK", + terminal: true, + createdAt: "2026-06-17T12:56:39.484Z" + }, { now: "2026-06-17T13:03:52.000Z" }); + + const snapshot = traceStore.snapshot(traceId); + assert.equal(snapshot.createdAt, "2026-06-17T12:56:13.650Z"); + assert.equal(snapshot.startedAt, "2026-06-17T12:56:13.650Z"); + assert.equal(snapshot.updatedAt, "2026-06-17T12:56:39.484Z"); + assert.equal(snapshot.finishedAt, "2026-06-17T12:56:39.484Z"); + assert.equal(snapshot.elapsedMs, 25834); +}); + +test("code agent trace store freezes terminal traces against stale post-terminal events", () => { + const traceStore = createCodeAgentTraceStore({ maxEvents: 20 }); + const traceId = "trc_trace-store-terminal-freeze"; + traceStore.append(traceId, { + type: "backend", + status: "running", + label: "agentrun:backend:turn/started", + createdAt: "2026-06-17T12:56:27.325Z" + }); + traceStore.append(traceId, { + type: "result", + status: "completed", + label: "agentrun:result:completed", + message: "AgentRun result is ready for HWLAB short-connection polling.", + terminal: true, + createdAt: "2026-06-17T12:56:41.293Z" + }); + traceStore.append(traceId, { + type: "result", + status: "completed", + label: "agentrun:result:completed", + message: "AgentRun result is ready for HWLAB short-connection polling.", + terminal: true, + createdAt: "2026-06-17T12:56:41.293Z" + }); + traceStore.append(traceId, { + type: "backend", + status: "running", + label: "agentrun:backend:runner-claim-waiting-for-stale-lease", + createdAt: "2026-06-17T13:04:49.500Z" + }); + + const snapshot = traceStore.snapshot(traceId); + assert.equal(snapshot.eventCount, 2); + assert.deepEqual(snapshot.events.map((item) => item.label), ["agentrun:backend:turn/started", "agentrun:result:completed"]); + assert.equal(snapshot.status, "completed"); + assert.equal(snapshot.updatedAt, "2026-06-17T12:56:41.293Z"); + assert.equal(snapshot.finishedAt, "2026-06-17T12:56:41.293Z"); + assert.equal(snapshot.elapsedMs, 13968); +}); diff --git a/internal/cloud/code-agent-trace-store.ts b/internal/cloud/code-agent-trace-store.ts index 4a96322e..02ddd87c 100644 --- a/internal/cloud/code-agent-trace-store.ts +++ b/internal/cloud/code-agent-trace-store.ts @@ -29,6 +29,7 @@ export function createCodeAgentTraceStore(options = {}) { nextSeq: 1, events: [], sourceEventKeys: new Set(), + eventSignatures: new Set(), assistantStreams: new Map(), listeners: new Set(), meta: { @@ -68,27 +69,41 @@ export function createCodeAgentTraceStore(options = {}) { now: meta.now, fallbackRunnerKind: trace.meta.runnerKind }); + const normalizedSignature = traceEventSignature(normalized); + if (normalizedSignature && trace.eventSignatures.has(normalizedSignature)) { + return trace.events.find((candidate) => traceEventSignature(candidate) === normalizedSignature) ?? null; + } + if (shouldSuppressPostTerminalEvent(trace, normalized)) return null; const normalizedSourceKey = traceSourceEventKey(normalized); if (normalizedSourceKey && trace.sourceEventKeys.has(normalizedSourceKey)) { return trace.events.find((candidate) => traceSourceEventKey(candidate) === normalizedSourceKey) ?? null; } + const hadEvents = trace.events.length > 0; trace.nextSeq += 1; trace.events.push(normalized); if (normalizedSourceKey) trace.sourceEventKeys.add(normalizedSourceKey); + if (normalizedSignature) trace.eventSignatures.add(normalizedSignature); if (trace.events.length > maxEvents) { trace.events.splice(0, trace.events.length - maxEvents); + rebuildTraceEventIndexes(trace); } - trace.updatedAt = normalized.createdAt; - if (normalized.terminal === true) { - trace.status = normalized.status === "completed" - ? "completed" - : normalized.status === "canceled" - ? "canceled" - : normalized.type; - trace.finishedAt = normalized.createdAt; - } else if (normalized.status === "failed" || normalized.type === "error" || normalized.type === "timeout") { - trace.status = normalized.type; - trace.finishedAt = normalized.createdAt; + trace.createdAt = minTimestamp(trace.createdAt, normalized.createdAt); + trace.startedAt = minTimestamp(trace.startedAt, normalized.createdAt); + trace.updatedAt = hadEvents ? maxTimestamp(trace.updatedAt, normalized.createdAt) : normalized.createdAt; + if (!trace.finishedAt) { + if (normalized.terminal === true) { + trace.status = normalized.status === "completed" + ? "completed" + : normalized.status === "canceled" + ? "canceled" + : normalized.type; + trace.finishedAt = normalized.createdAt; + trace.updatedAt = maxTimestamp(trace.updatedAt, trace.finishedAt); + } else if (normalized.status === "failed" || normalized.type === "error" || normalized.type === "timeout") { + trace.status = normalized.type; + trace.finishedAt = normalized.createdAt; + trace.updatedAt = maxTimestamp(trace.updatedAt, trace.finishedAt); + } } notify(trace, normalized); return normalized; @@ -630,6 +645,60 @@ function traceSourceEventKey(event) { return `${source}:${sourceSeq}:${label}`; } +function rebuildTraceEventIndexes(trace) { + trace.sourceEventKeys = new Set(); + trace.eventSignatures = new Set(); + for (const event of trace.events) { + const sourceKey = traceSourceEventKey(event); + if (sourceKey) trace.sourceEventKeys.add(sourceKey); + const signature = traceEventSignature(event); + if (signature) trace.eventSignatures.add(signature); + } +} + +function traceEventSignature(event) { + if (!event || typeof event !== "object") return null; + const type = textOrNull(event.type); + const status = textOrNull(event.status); + const label = textOrNull(event.label); + const createdAt = textOrNull(event.createdAt); + if (!type || !status || !label || !createdAt) return null; + return [ + type, + status, + label, + createdAt, + textOrNull(event.runId), + textOrNull(event.commandId), + textOrNull(event.itemId), + textOrNull(event.message) + ].map((item) => item ?? "").join("\u0001"); +} + +function shouldSuppressPostTerminalEvent(trace, event) { + if (!trace?.finishedAt) return false; + const finishedMs = Date.parse(trace.finishedAt); + const eventMs = Date.parse(event?.createdAt ?? ""); + if (!Number.isFinite(finishedMs) || !Number.isFinite(eventMs)) return true; + return eventMs > finishedMs; +} + +function minTimestamp(left, right) { + const leftMs = Date.parse(left ?? ""); + const rightMs = Date.parse(right ?? ""); + if (!Number.isFinite(leftMs)) return right; + if (!Number.isFinite(rightMs)) return left; + return rightMs < leftMs ? right : left; +} + +function maxTimestamp(left, right) { + const leftMs = Date.parse(left ?? ""); + const rightMs = Date.parse(right ?? ""); + if (!Number.isFinite(leftMs)) return right; + if (!Number.isFinite(rightMs)) return left; + return rightMs > leftMs ? right : left; +} + function textOrNull(value) { const text = String(value ?? "").trim(); return text ? text : null;