mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(accounting): close a local-Runner turn completed across a session restore as a lower bound (#20517)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native runs record usage receipts, and accounting settles a run's charge once its final receipt is complete. > - ACPX Claude reports provisional zero usage until a turn ends, then one final report. > - A restart can stop the provider after the agent's finish call and before that final report. The restored Runner then resumes the session and reports the same turn completed with no complete report. > - The receipt stayed incomplete, so the accepted run waited for the 10-minute closure timeout and then settled as unpriced. Its spend was missing from company totals in the meantime. > - This pull request closes such a turn as a lower bound, the same way a turn the Runner fails as lost on restore is already closed. > - The benefit is that an accepted run settles right away with its observed usage and an honest lower-bound label. ## Linked Issues or Issue Description Refs #20484 (restart-chaos L1 contention follow-up). **What happened?** After #20484 fixed the database contention, restart-chaos L1 (`--seed 20386 --db-pressure 4`, codex and claude, K1) reached 4/4 tasks done. Each task still failed the T-oracle `company_spend_settled` with `pending_runs=1`. A replay on master with `--keep` (`w-20261110T100747-s20386`) reproduced it, and I read the kept database: - The pending run (`71f71d10`, ACPX Claude) succeeded and had its result accepted. Its usage had `accountingReceiptReady: false`, zero tokens, `costStatus: unpriced`, and no cost event. - Its only turn started before K1, and every usage report in it was provisional (`runDeltaComplete: false`). K1 struck right after the agent's `paperclip_finish` call (source seq 118), before the turn's final report. - On restart, recovery verified the local process stopped and resumed the session (`session.resumed`, seq 119). The restored Runner then reported the same turn `turn.completed` (seq 124) with no usage report. - The other Claude run in that wave lost its turn as `provider_turn_lost_on_restore`, ran a new turn live with full usage, and settled at once. **Expected behavior** An accepted run whose open turn's final report was lost with the provider process settles promptly. It records its observed usage as a lower bound, like a turn lost on restore. A complete report that does arrive after the restore is still used. **Steps to reproduce** `server/src/services/native-runtime/native-usage-accounting.test.ts > closes a turn that completed across a session restore without its final report, as a lower bound` uses the event sequence from the kept database. It fails on master because the snapshot stays incomplete. ## What Changed - `createNativeTurnAccounting` observes `session.resumed` and marks every open turn as restored. A terminal event for a restored turn without a complete report marks it lost. The snapshot then closes it as a lower bound (`accountingUsageLost`), through the existing lost-on-restore path. A new `turn.started` clears the mark. - A lost terminal turn keeps its identity in the snapshot, so the executor's terminal-turn match certifies the final checkpoint as complete. This came from review. - The native session executor includes `session.resumed` when it rebuilds turn state from the run's event log after a controller restart. - `doc/SPEC-implementation.md` documents the case next to the existing lost-on-restore rule. ## Verification - Red on master, green here: `native-usage-accounting.test.ts > closes a turn that completed across a session restore without its final report, as a lower bound`. It also checks that replaying the history twice gives the same snapshot. - Red on the first revision, green after the review fix: `native-session-executor.test.ts > certifies a turn completed across a session restore as a lower bound (history replay: false|true)`. It drives the restored sequence through executor completion, both live and from the event history. - New guards that pass before and after: - `keeps a restored turn's own final report when it arrives after the restore`: a complete report after resume is used, not marked lost. - `keeps an incomplete turn pending when the session resumed before it started`. - The existing `keeps an incomplete provider report pending instead of trusting placeholder zeroes` still passes. - Related suites pass: `native-usage-accounting`, `native-usage-delivery.integration`, `closed-native-receipt-accounting` (107 tests), and the usage, accounting and receipt cases in `native-session-executor.test.ts` (151 tests). Server typecheck passes. - Live, restart-chaos L1, seed 20386, `--db-pressure 4`, codex and claude, K1, head `a6e521f06c`, clean worktree: `w-20261110T104332-s20386` passed 4/4. R1–R6 and the T-oracle passed for every task. `company_spend_settled` reported `pending_runs=0` (summary 85.30823 = cost events 85.30823), with zero serialization failures under about 397k pressure updates. On master the same seed failed every task on `company_spend_settled pending_runs=1` (`w-20261110T100747-s20386`, among others). ## Risks - A restored turn without a complete report now settles at its observed usage, flagged as a lower bound, instead of after 10 minutes as "closure timed out". In both cases the final report never arrived. The money recorded is the same observed usage; only the timing and the provenance label differ. - Scope: local Runner only. The executor fallback-recorder path sees `session.resumed`. The producer-owned delivery path for remote runners (`native-usage-delivery.ts`, `isNativeUsageAccountingEvent`) still drops it, so a remote ACPX/OpenCode turn restored mid-turn still waits for the closure timeout. That is a follow-up. - No schema or persisted-format change. ## Model Used - Claude Opus 5.5 (Anthropic), model ID `claude-opus-5-5`, used through Claude Code with shell, file-edit and code-execution tools. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
8b3348c6ba
commit
37808f2a4e
5 files changed
+99
-5
No files matched your search
@@ -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.
|
||||
|
||||
@@ -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 }) => {
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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"));
|
||||
|
||||
@@ -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<string, { usage: Record<string, unknown> | null; terminal: boolean; lost?: boolean }>();
|
||||
const turns = new Map<string, { usage: Record<string, unknown> | null; terminal: boolean; lost?: boolean;
|
||||
/** Open when the provider session was restored after process loss. */
|
||||
restored?: boolean }>();
|
||||
const sequences = new Map<string, number>();
|
||||
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;
|
||||
},
|
||||
|
||||
Reference in new issue
Block a user