diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index a91f069d681..d5736269477 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -2030,7 +2030,10 @@ that every usage recorder binds is not evidence that provider work began. A nati turn that the restored Runner reports as lost on restore can never report its usage. It does not hold the run's receipt open: the receipt closes when the run's other turns do, and its provenance says the observed usage is a lower -bound. Raw provider errors and credentials are excluded from +bound. The same applies to a turn that was open when the provider session was +restored after process loss and that the restored Runner then reports +terminal without a complete usage report: the provider process that would have +sent the final report was stopped. Raw provider errors and credentials are excluded from cost details. ACP usage retains the runtime-selected model, including defaults; conflicting before/after model IDs do not authorize a single-model estimate. Remote OpenCode billing uses only credentials forwarded to the execution target. diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 16de3f1308b..9ffa5f5ec52 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -8487,6 +8487,34 @@ describe("native terminal-turn accounting", () => { } }); + // w-20261110T100747-s20386 run 71f71d10: K1 stopped the ACPX Claude + // provider before the turn's final report; the restored Runner resumed the + // session and reported the same turn completed without it. + it.each([false, true])("certifies a turn completed across a session restore as a lower bound (history replay: %s)", async replay => { + const { rehydrateRunnerdUsageNotification } = await import("@paperclipai/paperclip-runner/live"); + const onUsage = vi.fn(async (_receipt: import("@paperclipai/adapter-utils").AdapterUsageCheckpoint) => {}); + const event = (eventType: string, turnId: string | undefined, sourceSeq: number, payload: object = {}) => ({ + eventType, turnId, sourceSeq, sourceInstanceId: "provider", emittedAt: new Date().toISOString(), payload, + }) as unknown as PrpEvent; + const provisional = { kind: "usage", usage: rehydrateRunnerdUsageNotification({ provider: "acpx", runDeltaAvailable: false, + cumulative: { inputTokens: 0, outputTokens: 0, providerCostUsd: 0 }, + runDelta: { inputTokens: 0, outputTokens: 0, cacheReadTokens: 0, cacheWriteTokens: 0, providerCostUsd: 0 } }, "session", "turn").tokenUsage }; + const before = [event("turn.started", "turn", 7), event("item.completed", "turn", 10, provisional), event("item.completed", "turn", 113, provisional)]; + const restored = [event("session.resumed", undefined, 119, { context: {} }), event("turn.completed", "turn", 124, { status: "completed" })]; + state.execute.mockReset().mockImplementationOnce(async () => { + for (const entry of [...(replay ? [] : before), ...restored]) await accountingEvents.committed!(entry); + return { result: { summary: "done" }, terminal: { runTerminalState: "succeeded" }, turnId: "turn", + normalizedSessionId: "session", providerSessionId: null, driverKind: "acpx_runtime", driverVersion: "1", + nativeEventCount: 5, highestContiguousSourceSeq: 124, usage: null }; + }); + const result = await executePaperclipNativeSession({ db: leaseDb(execution, {}, {}, [], {}, "running", replay ? before : []), + execution: { ...execution, provider: { kind: "acpx", agent: "claude", model: "claude-sonnet-5" } } as NativeExecutionInput, + runnerInstanceId: "runner", runnerEnvironment: { ANTHROPIC_API_KEY: "fixture" }, onUsage }); + expect(result).toMatchObject({ usageComplete: true }); + expect(onUsage).toHaveBeenLastCalledWith(expect.objectContaining({ complete: true, + pricingProvenance: expect.objectContaining({ evidence: expect.any(String) }) })); + }); + it.each(["claude", "codex"].flatMap(agent => ["missing", "partial", "complete_zero", "later_partial", "recovered"].map(scenario => ({ agent, scenario })), ))("honors runner completeness for ACPX $agent ($scenario)", async ({ agent, scenario }) => { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index be4537caf08..a5917119028 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -10339,7 +10339,7 @@ async function executePaperclipNativeSessionAttempt( const history = await input.db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( eq(heartbeatRunEvents.companyId, input.execution.binding.companyId), eq(heartbeatRunEvents.runId, input.execution.binding.runId), or(sql`${heartbeatRunEvents.payload}->'prpEvent'->'payload'->>'kind' = 'usage'`, - inArray(heartbeatRunEvents.eventType, ["turn.started", "turn.completed", "turn.failed", "turn.cancelled", "turn.interrupted"])), + inArray(heartbeatRunEvents.eventType, ["turn.started", "turn.completed", "turn.failed", "turn.cancelled", "turn.interrupted", "session.resumed"])), )).orderBy(heartbeatRunEvents.seq); for (const row of history) { const event = record(row.payload).prpEvent as PrpEvent; diff --git a/server/src/services/native-runtime/native-usage-accounting.test.ts b/server/src/services/native-runtime/native-usage-accounting.test.ts index c5cc1be1089..88f6e4d3b1e 100644 --- a/server/src/services/native-runtime/native-usage-accounting.test.ts +++ b/server/src/services/native-runtime/native-usage-accounting.test.ts @@ -55,6 +55,55 @@ describe("native turn accounting", () => { expect(normalizeNativeUsage(snapshot.usage)).toEqual({ inputTokens: 2800, outputTokens: 1446, cachedInputTokens: 124322, cacheWriteTokens: 2794 }); }); + // w-20261110T100747-s20386 run 71f71d10 (ACPX Claude): K1 killed the + // provider after the agent's finish tool call, before the turn's final usage + // report. The restored Runner resumed the session and reported the same + // turn completed without that report, so its receipt stayed incomplete until + // the 10-minute closure timeout although the result was accepted. + it("closes a turn that completed across a session restore without its final report, as a lower bound", () => { + const acpx = { kind: "acpx" as const, agent: "claude", model: "claude-sonnet-5" } as never; + const provisional = { kind: "usage", usage: { runDeltaComplete: false, + runDelta: { requests: 1, inputTokens: 0, outputTokens: 0, cacheReadTokens: 0, cacheWriteTokens: 0, providerCostUsd: 0 } } }; + const before = [event(7, "turn.started", "turn-span"), event(10, "item.completed", "turn-span", provisional), + event(113, "item.completed", "turn-span", provisional)]; + const restored = [event(119, "session.resumed", undefined as never, { context: {} }), + event(124, "turn.completed", "turn-span", { status: "completed" })]; + const accounting = createNativeTurnAccounting(acpx); + [...before, ...restored].forEach(value => accounting.observe(value)); + const snapshot = accounting.snapshot(); + expect(snapshot.complete).toBe(true); + // The executor certifies completion only when the snapshot names the terminal turn. + expect(snapshot.turnId).toBe("turn-span"); + expect(snapshot.usage).toMatchObject({ accountingUsageLost: true, accountingCostIncomplete: true }); + // A replay of the same history after another restart reaches the same state. + const replayed = createNativeTurnAccounting(acpx); + [...before, ...restored, ...before, ...restored].forEach(value => replayed.observe(value)); + expect(replayed.snapshot()).toEqual(snapshot); + }); + + it("keeps a restored turn's own final report when it arrives after the restore", () => { + const acpx = { kind: "acpx" as const, agent: "claude", model: "claude-sonnet-5" } as never; + const accounting = createNativeTurnAccounting(acpx); + const provisional = { kind: "usage", usage: { runDeltaComplete: false, runDelta: { inputTokens: 0, outputTokens: 0, providerCostUsd: 0 } } }; + const final = { kind: "usage", usage: { runDeltaComplete: true, runDelta: { inputTokens: 40, outputTokens: 7, providerCostUsd: 0.01 } } }; + [event(1, "turn.started", "turn-span"), event(2, "item.completed", "turn-span", provisional), + event(3, "session.resumed", undefined as never), event(4, "item.completed", "turn-span", final), + event(5, "turn.completed", "turn-span")].forEach(value => accounting.observe(value)); + const snapshot = accounting.snapshot(); + expect(snapshot.complete).toBe(true); + expect(snapshot.usage).not.toHaveProperty("accountingUsageLost"); + expect(normalizeNativeUsage(snapshot.usage)).toMatchObject({ inputTokens: 40, outputTokens: 7 }); + }); + + it("keeps an incomplete turn pending when the session resumed before it started", () => { + const accounting = createNativeTurnAccounting(provider); + accounting.observe(event(1, "session.resumed", undefined as never)); + accounting.observe(event(2, "turn.started")); + accounting.observe(event(3, "item.completed", "turn-1", { kind: "usage", usage: { runDeltaComplete: false, runDelta: { inputTokens: 0, outputTokens: 0 } } })); + accounting.observe(event(4, "turn.completed")); + expect(accounting.snapshot().complete).toBe(false); + }); + it("keeps a turn that failed for any other reason without usage pending", () => { const accounting = createNativeTurnAccounting(provider); accounting.observe(event(1, "turn.started")); diff --git a/server/src/services/native-runtime/native-usage-accounting.ts b/server/src/services/native-runtime/native-usage-accounting.ts index a89fb00c2df..4d12032bbc3 100644 --- a/server/src/services/native-runtime/native-usage-accounting.ts +++ b/server/src/services/native-runtime/native-usage-accounting.ts @@ -64,7 +64,9 @@ export const NATIVE_TURN_LOST_ON_RESTORE = "provider_turn_lost_on_restore"; const LOST_ON_RESTORE = NATIVE_TURN_LOST_ON_RESTORE; export function createNativeTurnAccounting(provider: NativeExecutionInput["provider"]) { - const turns = new Map | null; terminal: boolean; lost?: boolean }>(); + const turns = new Map | null; terminal: boolean; lost?: boolean; + /** Open when the provider session was restored after process loss. */ + restored?: boolean }>(); const sequences = new Map(); let currentTurn: string | undefined; const terminalTypes = ["turn.completed", "turn.failed", "turn.cancelled", "turn.interrupted"]; @@ -96,20 +98,30 @@ export function createNativeTurnAccounting(provider: NativeExecutionInput["provi const current = currentTurn ? turns.get(currentTurn) : undefined; return { usage: { runDelta, accountingCostIncomplete: costIncomplete, ...(usageLost ? { accountingUsageLost: true } : {}), ...(retainCost ? { accountingCostUsdExact: `${cost / 1_000_000_000n}.${String(cost % 1_000_000_000n).padStart(9, "0")}` } : {}) }, - turnId: current && normalizeNativeUsage(current.usage) ? currentTurn : undefined, complete }; + // A turn closed as lost still names the terminal turn the executor must match. + turnId: current && (normalizeNativeUsage(current.usage) || (current.lost && current.terminal)) ? currentTurn : undefined, complete }; }; return { observe(event: PrpEvent) { if (!event) return false; const payload = record(event.payload); const terminal = terminalTypes.includes(event.eventType); - if (payload.kind !== "usage" && event.eventType !== "turn.started" && !terminal) return false; + const resumed = event.eventType === "session.resumed"; + if (payload.kind !== "usage" && event.eventType !== "turn.started" && !terminal && !resumed) return false; // Historical replay can revisit an earlier report after restoration. // Re-persist its aggregate, but never rewind or count that report twice. if (event.sourceInstanceId && Number.isSafeInteger(event.sourceSeq)) { if (event.sourceSeq <= (sequences.get(event.sourceInstanceId) ?? -1)) return true; sequences.set(event.sourceInstanceId, event.sourceSeq); } + // A restore after process loss ends the provider process that would + // have sent an open turn's final report. Mark those turns; a terminal + // event for one without a complete report closes it as a lower bound, + // like a turn the Runner fails as lost on restore. + if (resumed) { + for (const turn of turns.values()) if (!turn.terminal) turn.restored = true; + return false; + } const key = event.turnId ?? "unattributed"; const turn = turns.get(key) ?? { usage: null, terminal: false }; if (event.eventType === "turn.started") { @@ -117,6 +129,7 @@ export function createNativeTurnAccounting(provider: NativeExecutionInput["provi turn.usage = null; turn.terminal = false; turn.lost = false; + turn.restored = false; } else if (payload.kind === "usage" && (!currentTurn || !turns.has(key))) currentTurn = event.turnId; if (payload.kind === "usage") { const usage = record(payload.usage); @@ -125,6 +138,7 @@ export function createNativeTurnAccounting(provider: NativeExecutionInput["provi } if (terminal) turn.terminal = true; if (event.eventType === "turn.failed" && record(payload.error).code === LOST_ON_RESTORE) turn.lost = true; + if (terminal && turn.restored) turn.lost = true; turns.set(key, turn); return true; },