From 722db815da39e47c5176b0d2ae51b48f00127df3 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 4 Oct 2026 00:22:36 -0700 Subject: [PATCH 1/2] test: drain heartbeat work before deleting recovery fixtures Co-Authored-By: Paperclip --- server/src/__tests__/heartbeat-process-recovery.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index d88dac7d329..cb02a301008 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -548,7 +548,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { } await new Promise((resolve) => setTimeout(resolve, 50)); await waitForHeartbeatIdle(db, 5_000); - await new Promise((resolve) => setTimeout(resolve, 100)); + // Terminal rows do not mean their background finalizers have returned. + // Drain tracked wakeups too: one can still create a run after the row poll. + await heartbeatService(db).drainActiveRunExecutions(); await db.delete(activityLog); await db.delete(agentRuntimeState); await db.delete(companySkills); From 79b75163221a9cf3402d13e509fd6d3c16a63176 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 4 Oct 2026 00:22:37 -0700 Subject: [PATCH 2/2] fix: retain ACPX usage when receipt persistence rejects Co-Authored-By: Paperclip --- .../src/acpx-engine/execute.test.ts | 39 +++++++++++++++++++ .../adapter-utils/src/acpx-engine/execute.ts | 6 +++ 2 files changed, 45 insertions(+) diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index fe984e662aa..c14028abf67 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -1450,6 +1450,45 @@ describe("shared ACPX engine runtime behavior", () => { expect(statusLine?.text).toContain('"cost"'); }); + it.each(["checkpoint", "terminal"])("returns received usage after the %s receipt sink rejects and still closes the runtime", async (failedSave) => { + const root = await makeTempRoot(); + const close = vi.fn(async () => {}); + const cancel = vi.fn(async () => {}); + let reads = 0; + const execute = createAcpxEngineExecutor({ + resolveBillingIdentity: () => ({ provider: "anthropic", biller: "anthropic", billingType: "api" }), + createRuntime: () => ({ + ...buildRuntime(), + getStatus: async () => ++reads === 1 + ? { usage: { cost: { amount: 0.4, currency: "USD" } } } + : { usage: { cumulative: { inputTokens: 120, outputTokens: 4500, cachedReadTokens: 900, cachedWriteTokens: 30 }, cost: { amount: 1.15, currency: "USD" } } }, + startTurn: () => ({ + events: (async function* () { + yield { type: "status", text: "usage", tag: "usage_update", + breakdown: { inputTokens: 100, outputTokens: 4000, cachedReadTokens: 800, cachedWriteTokens: 20 }, cost: { amount: 1.1, currency: "USD" } }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), cancel, + }), close, + }) as never, + }); + const onUsage = vi.fn(async (receipt: { complete: boolean }) => { + if (failedSave === "checkpoint" || receipt.complete) throw new Error("Receipt storage unavailable"); + }); + const result = await execute({ runId: `failed-${failedSave}-save`, agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir: path.join(root, "state") }, + context: {}, onMeta: async () => {}, onLog: async () => {}, onUsage } as never); + expect(result).toMatchObject({ exitCode: 1, usageBasis: "per_run", usageComplete: failedSave === "terminal", + provider: "anthropic", biller: "anthropic", billingType: "api" }); + expect(result.errorMessage).toContain("Receipt storage unavailable"); + expect(result.usage).toEqual(failedSave === "terminal" + ? { inputTokens: 150, outputTokens: 4500, cachedInputTokens: 900 } + : { inputTokens: 120, outputTokens: 4000, cachedInputTokens: 800 }); + expect(result.costUsd).toBeCloseTo(failedSave === "terminal" ? 0.75 : 0.7); + expect(onUsage).toHaveBeenCalledTimes(failedSave === "terminal" ? 2 : 1); + expect(cancel).toHaveBeenCalledTimes(1); + expect(close).toHaveBeenCalledTimes(1); + }); + it("falls back to usage_update events when the runtime lacks getStatus", async () => { const root = await makeTempRoot(); const stateDir = path.join(root, "state"); diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index e5b4e84a424..a18932893b1 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -4704,6 +4704,9 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { }; let eventBreakdown: AcpRuntimeUsageBreakdown | null = null; let eventCostUsd: number | null = null; + // Receipt persistence is fallible; retain the provider evidence before + // calling the sink so the failure result can still return it to the host. + let observedUsage: Pick = { usageComplete: false }; let lastRuntimeEventAt: number | null = null; let observedRuntimeEvents = 0; // The turn-local state the sequence steps share. `promptBuild` sets the @@ -4905,6 +4908,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { eventBreakdown = event.breakdown ?? eventBreakdown; eventCostUsd = usdCostAmount(event.cost) ?? eventCostUsd; const checkpoint = summarizeAcpxTurnUsage({ preStatus: preTurnStatus, postStatus: null, eventBreakdown, eventCostUsd }); + observedUsage = { usage: checkpoint.usage ?? undefined, costUsd: checkpoint.costUsd, usageBasis: "per_run", usageComplete: false }; await ctx.onUsage?.({ ...billingFields, usage: checkpoint.usage ?? undefined, costUsd: checkpoint.costUsd, model: prepared.requestedModel, usageBasis: "per_run", complete: false }); } @@ -4996,6 +5000,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { // An interrupted stream is provisional unless the runtime supplies its // post-terminal receipt. An in-stream usage_update alone cannot close it. const usageComplete = !channelLost && (turnSucceeded || postTurnStatus?.usage != null); + observedUsage = { usage: turnUsage.usage ?? undefined, costUsd: turnUsage.costUsd, usageBasis: "per_run", usageComplete }; await ctx.onUsage?.({ ...billingFields, usage: turnUsage.usage ?? undefined, costUsd: turnUsage.costUsd, model: prepared.requestedModel, usageBasis: "per_run", complete: usageComplete }); const failedTurn = terminal.status === "failed" || terminal.status === "cancelled" || timedOut; @@ -5217,6 +5222,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { errorMessage: message, errorCode: timedOut ? "acpx_timeout" : (emitted?.classified.errorCode ?? null), errorMeta: emitted?.classified.errorMeta, + ...observedUsage, ...billingFields, ...referencedProjectStagingFailuresField, model: prepared.requestedModel || null,