diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index d5736269477..485c8b17dc7 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -2033,7 +2033,9 @@ run's other turns do, and its provenance says the observed usage is a lower 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 +sent the final report was stopped. Local and remote Runners both apply it; the +remote producer journals the restore as a receipt so a replacement controller +rebuilds the same turn state. 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 9ffa5f5ec52..42ecfedc646 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -8698,12 +8698,25 @@ describe("native terminal-turn accounting", () => { expect(firstCapture).toHaveBeenCalledTimes(2); expect(secondCapture).toHaveBeenCalledTimes(1); await expect(retainedHook({ ...firstReport, runId: "unrelated-run" })).rejects.toThrow("native_usage_run_binding_mismatch"); + // A session restore reaches the producer that owns its run, so an open + // turn can close as a lower bound; another run's restore is ignored. + const currentResume = { ...secondReport, eventType: "session.resumed" } as PrpEvent; + const drainingResume = { ...firstReport, eventType: "session.resumed" } as PrpEvent; + await retainedHook(currentResume); + expect(secondCapture).toHaveBeenLastCalledWith(currentResume); + await retainedHook(drainingResume); + expect(firstCapture).toHaveBeenLastCalledWith(drainingResume); + const [firstCalls, secondCalls] = [firstCapture.mock.calls.length, secondCapture.mock.calls.length]; + await expect(retainedHook({ ...currentResume, runId: "unrelated-run" })).resolves.toBeUndefined(); + expect([firstCapture.mock.calls.length, secondCapture.mock.calls.length]).toEqual([firstCalls, secondCalls]); const recoveryHook = state.createTransport.mock.calls.at(-1)![0].onRecoveryAccountingEvent!; const source = { runnerInstanceId: "warm-producer", environmentLeaseId: "warm-usage-workspace", runId: first.binding.runId, normalizedSessionId: first.session.normalizedSessionId!, turnId: "source-turn", itemId: "source-item" }; await recoveryHook(firstReport, source); - expect(firstCapture).toHaveBeenCalledTimes(3); - expect(secondCapture).toHaveBeenCalledTimes(1); + expect(firstCapture).toHaveBeenCalledTimes(firstCalls + 1); + expect(secondCapture).toHaveBeenCalledTimes(secondCalls); + // Recovery routing stays usage-only: a source restore needs no proven source. + await expect(recoveryHook({ ...drainingResume, runId: "foreign-source" }, source)).resolves.toBeUndefined(); await expect(recoveryHook({ ...firstReport, runId: "foreign-source" }, source)).rejects.toThrow("native_recovery_usage_source_unproven"); } finally { if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index a5917119028..e15be30eaad 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -26,7 +26,7 @@ import { prepareRemoteOpenCodeArtifacts, type OpenCodeLaunchArtifacts } from "./ import { rehydratePinnedSkillAssets } from "./runtime-context.js"; import { createNativeUsageDelivery } from "./native-usage-delivery.js"; import { createNativeRecoveryUsageDelivery } from "./native-recovery-usage-delivery.js"; -import { createNativeTurnAccounting, NATIVE_LOST_TURN_USAGE_EVIDENCE, nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage, usesNativeUsageDelivery, isNativeUsageAccountingEvent } from "./native-usage-accounting.js"; +import { createNativeTurnAccounting, NATIVE_LOST_TURN_USAGE_EVIDENCE, nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage, usesNativeUsageDelivery, isNativeUsageAccountingEvent, isNativeUsageDeliveryEvent } from "./native-usage-accounting.js"; import { crashPoint } from "@paperclipai/shared/crash-points"; export { nativeUsageCostUsd, nativeUsageBiller, normalizeNativeUsage } from "./native-usage-accounting.js"; import { measureRunnerOperation, recordRunnerTimingReceipt } from "./runner-operation-timing.js"; @@ -1968,7 +1968,9 @@ class SessionAuthorityEpoch { } async acceptAccountingEvent(event: PrpEvent) { - if (!isNativeUsageAccountingEvent(event)) return; + if (!isNativeUsageDeliveryEvent(event)) return; + // A restore carries no usage; it never fails delivery for another run. + if (!isNativeUsageAccountingEvent(event) && event.runId !== this.runId && event.runId !== this.#drainingRun?.runId) return; this.#assertCurrent(); if (event.runId !== this.runId && event.runId !== this.#drainingRun?.runId) throw new Error("native_usage_run_binding_mismatch"); await (event.runId === this.runId ? this.onAccountingEvent : this.#drainingRun?.capture)?.(event); diff --git a/server/src/services/native-runtime/native-usage-accounting.ts b/server/src/services/native-runtime/native-usage-accounting.ts index 4d12032bbc3..93acb96733c 100644 --- a/server/src/services/native-runtime/native-usage-accounting.ts +++ b/server/src/services/native-runtime/native-usage-accounting.ts @@ -143,6 +143,11 @@ export function createNativeTurnAccounting(provider: NativeExecutionInput["provi return true; }, snapshot, + /** A restore matters only while a turn is open. */ + hasOpenTurn() { + for (const turn of turns.values()) if (!turn.terminal) return true; + return false; + }, finish(turnId: string | null) { const turn = turnId ? turns.get(turnId) : undefined; if (turn) turn.terminal = true; @@ -215,3 +220,11 @@ export function usesNativeUsageDelivery(input: { export function isNativeUsageAccountingEvent(event: Pick) { return ["usage.reported", "turn.started", "turn.completed", "turn.failed", "turn.cancelled", "turn.interrupted"].includes(event.eventType); } + +/** Producer delivery for the run itself also needs the session restore: a + * turn open across it can close as a lower bound (see createNativeTurnAccounting). + * Recovery routing keeps isNativeUsageAccountingEvent: a source run's restore + * carries no usage and must not require a proven recovery source. */ +export function isNativeUsageDeliveryEvent(event: Pick) { + return isNativeUsageAccountingEvent(event) || event.eventType === "session.resumed"; +} diff --git a/server/src/services/native-runtime/native-usage-delivery.integration.test.ts b/server/src/services/native-runtime/native-usage-delivery.integration.test.ts index b35b6c532cb..0cf51f78357 100644 --- a/server/src/services/native-runtime/native-usage-delivery.integration.test.ts +++ b/server/src/services/native-runtime/native-usage-delivery.integration.test.ts @@ -559,6 +559,54 @@ describe("producer-owned native usage", () => { expect((await read()).usageJson).toMatchObject({ accountingReceiptReady: true }); }); + describe("a turn restored mid-turn on a remote Runner", () => { + const restoredSeed = async () => ({ ...await seed(), provider: { kind: "opencode" as const, model: "test-model" } }); + const partial = (input: Awaited>, sequence: number) => { + const value = event(input, sequence, "usage.reported"); + (value.payload as Record).runDeltaAvailable = false; + return value; + }; + const read = async (runId: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]; + + // The local Runner rule from #20517: K1 stopped the provider before the + // turn's final report; the restored Runner reported the turn completed. + it("closes the turn as a lower bound, also after the controller restarts between restore and terminal", async () => { + const input = await restoredSeed(); + const history = [event(input, 1, "turn.started"), partial(input, 2), event(input, 3, "session.resumed", undefined as never)]; + const done = event(input, 4, "turn.completed"); + let delivery = await createNativeUsageDelivery(db, input); + for (const value of history) await delivery.capture(value); + // A replacement controller rebuilds the receiver from the saved journal. + delivery = await createNativeUsageDelivery(db, input); + await delivery.capture(done); + expect((await read(input.runId)).usageJson).toMatchObject({ accountingReceiptReady: true, + pricingProvenance: { evidence: expect.stringContaining("lower bound") } }); + // Replaying every event into another receiver is idempotent. + delivery = await createNativeUsageDelivery(db, input); + for (const value of [...history, done]) await delivery.capture(value); + expect((await read(input.runId)).usageJson).toMatchObject({ accountingReceiptReady: true }); + }); + + it("keeps the turn's own final report when it arrives after the restore", async () => { + const input = await restoredSeed(); + const delivery = await createNativeUsageDelivery(db, input); + for (const value of [event(input, 1, "turn.started"), partial(input, 2), event(input, 3, "session.resumed", undefined as never), + event(input, 4, "usage.reported"), event(input, 5, "turn.completed")]) await delivery.capture(value); + const usage = (await read(input.runId)).usageJson; + expect(usage).toMatchObject({ accountingReceiptReady: true, outputTokens: 10 }); + expect(JSON.stringify(usage?.pricingProvenance ?? {})).not.toContain("lower bound"); + }); + + it("keeps a turn started after the restore pending, and journals nothing for a restore with no open turn", async () => { + const input = await restoredSeed(); + const delivery = await createNativeUsageDelivery(db, input); + for (const value of [event(input, 1, "session.resumed", undefined as never), event(input, 2, "turn.started"), partial(input, 3), + event(input, 4, "turn.completed")]) await delivery.capture(value); + expect((await read(input.runId)).usageJson).toMatchObject({ accountingReceiptReady: false }); + expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, input.runId))).toHaveLength(3); + }); + }); + it("replays partial and final usage once after a lost commit response and restores turn aggregation", async () => { const input = { ...await seed(), provider: { kind: "opencode" as const, model: "test-model" } }; let loseResponse = false; diff --git a/server/src/services/native-runtime/native-usage-delivery.ts b/server/src/services/native-runtime/native-usage-delivery.ts index c263db4d9f0..44ceb7aa6eb 100644 --- a/server/src/services/native-runtime/native-usage-delivery.ts +++ b/server/src/services/native-runtime/native-usage-delivery.ts @@ -155,7 +155,15 @@ function createNativeUsageReceiver(db: Db, input: NativeUsageDeliveryInput, sour let pending: UsageReceiptEnvelope | undefined; let chain = Promise.resolve(); const capture = async (event: PrpEvent) => { - if (event.eventType !== "usage.reported" && event.eventType !== "turn.started" && !terminal.has(event.eventType)) return; + // A restore is journaled only for turn-reporting providers with an open + // turn, so a rebuilt receiver can close that turn as a lower bound. + // Receivers that predate it replay the receipt as an ignored event. + if (event.eventType === "session.resumed") { + // Never fail delivery for a lifecycle event: an unacknowledged event + // would replay without end. Another source's restore is not ours. + if (!turnAccounting || event.runId !== input.runId || event.sourceInstanceId !== input.runnerInstanceId + || (!turnAccounting.hasOpenTurn() && !accepted.has(event.sourceEventId))) return; + } else if (event.eventType !== "usage.reported" && event.eventType !== "turn.started" && !terminal.has(event.eventType)) return; if (event.runId !== input.runId || event.sourceInstanceId !== input.runnerInstanceId) throw new Error("native_usage_source_mismatch"); const payload = event.payload as Record; if (event.eventType === "usage.reported" && await captureLateUsageForOwner(db, input, event, payload)) return;