fix: retain failed native receipts and known attempted-run charges

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip committed 2026-10-03 21:51:37 -07:00
1 parent a9c6f25b88
commit b2815ee3cd
6 files changed
+107 -20

No files matched your search

@@ -191,6 +191,17 @@ const state = vi.hoisted(() => ({
release: null as null | (() => void),
}));
const accountingEvents = vi.hoisted(() => ({ committed: undefined as undefined | ((event: PrpEvent) => Promise<void>) }));
vi.mock("./paperclip-control-plane-port.js", async importOriginal => {
const original = await importOriginal<typeof import("./paperclip-control-plane-port.js")>();
return { ...original, PaperclipControlPlanePort: class extends original.PaperclipControlPlanePort {
constructor(...args: ConstructorParameters<typeof original.PaperclipControlPlanePort>) {
super(...args);
accountingEvents.committed = args[2]?.onCommittedEvent;
}
} };
});
const grokCopyBack = vi.hoisted(() => vi.fn(async (_input: { readSandboxAuth: () => Promise<Buffer>; hostHomeDir: string }) => undefined));
vi.mock("@paperclipai/adapter-grok-local/server", async importOriginal => ({
...await importOriginal<typeof import("@paperclipai/adapter-grok-local/server")>(),
@@ -4873,7 +4884,7 @@ function leaseDb(
where: () => query,
orderBy: () => query,
for: () => query,
limit: () => Promise.resolve(rows),
limit: () => query,
};
return query;
},
@@ -4881,7 +4892,9 @@ function leaseDb(
const insert = (table: unknown) => ({
values: (values: Record<string, unknown>) => {
updates.push({ table, values });
return { returning: async () => [values] };
const query = { returning: async () => [values], onConflictDoUpdate: () => query, onConflictDoNothing: () => query,
then: Promise.resolve([values]).then.bind(Promise.resolve([values])) };
return query;
},
});
const tx = {
@@ -5134,6 +5147,27 @@ describe("native startup restart detachment", () => {
});
});
describe("native failed-turn accounting", () => {
beforeEach(() => vi.spyOn(issueServiceModule, "issueService").mockReturnValue({ update: vi.fn(async () => ({ status: "blocked", statusVersion: 1 })) } as unknown as ReturnType<typeof issueServiceModule.issueService>));
afterEach(() => vi.restoreAllMocks());
it.each(["turn.failed", "turn.cancelled", "turn.interrupted", "missing_terminal", "session_total", "empty_usage", "different_turn"])("preserves run usage for %s without certifying incomplete evidence", async (scenario) => {
const onUsage = vi.fn(async (_receipt: import("@paperclipai/adapter-utils").AdapterUsageCheckpoint) => {});
const failure = new Error("provider_transport_failed: no semantic result");
state.execute.mockReset().mockImplementationOnce(async () => {
const usage = { inputTokens: 120, cacheReadTokens: 100, outputTokens: 10, providerCostUsd: 0.25 };
await accountingEvents.committed!({ eventType: "item.completed", turnId: "turn", emittedAt: new Date().toISOString(), payload: { kind: "usage", usage: scenario === "session_total" ? { total: usage } : { runDelta: scenario === "empty_usage" ? { requests: 1 } : usage } } } as unknown as PrpEvent);
if (scenario !== "missing_terminal") await accountingEvents.committed!({ eventType: scenario.startsWith("turn.") ? scenario : "turn.failed", turnId: scenario === "different_turn" ? "another-turn" : "turn", emittedAt: new Date().toISOString(), payload: { status: "failed" } } as unknown as PrpEvent);
throw failure;
});
await expect(executePaperclipNativeSession({ db: leaseDb(), execution: { ...execution, provider: { kind: "codex", model: "gpt-6-astra" } } as NativeExecutionInput, runnerInstanceId: "runner", onUsage })).rejects.toBe(failure);
if (["session_total", "empty_usage"].includes(scenario)) expect(onUsage).not.toHaveBeenCalled();
else {
expect(onUsage).toHaveBeenLastCalledWith(expect.objectContaining({ complete: scenario.startsWith("turn."), costUsd: 0.25, usageBasis: "per_run", usage: { inputTokens: 20, cachedInputTokens: 100, outputTokens: 10 } }));
expect(onUsage.mock.calls[0]?.[0]).toMatchObject({ complete: false });
}
});
});
describe("native resumed preparation timing", () => {
it("keeps answered-question ingress at the run root rather than charging it to preparation", async () => {
const answeredAtMs = Date.now();
@@ -7838,6 +7838,22 @@ async function executePaperclipNativeSessionWithinScope(
binding: { ...input.execution.binding, normalizedSessionId: nativeSessionKey(input.execution), runnerSourceInstanceId: effectiveRunnerInstanceId },
resolve: resolveNativeRuntimeRequest,
});
let observedAccountingUsage: Record<string, unknown> | null = null;
let observedAccountingTurn: string | undefined;
const persistAccountingUsage = async (usage: Record<string, unknown> | null, complete: boolean) => {
const accountingUsage = normalizeNativeUsage(usage, { inputIncludesCacheReads: input.execution.provider.kind === "codex" });
const accountingCost = nativeUsageCostUsd(usage, input.execution.provider);
const billing = resolveNativeBilling(input.execution.provider, input.runnerEnvironment, input.billingIdentity);
await input.onUsage?.({ usage: accountingUsage, ...billing, complete, usageBasis: "per_run",
model: input.execution.provider.model ?? "unknown", costUsd: accountingCost ?? null });
if (!input.onUsage) await input.db.update(heartbeatRuns).set({
costAccountingPending: true,
usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || ${JSON.stringify({
...accountingUsage, ...billing, accountingReceiptReady: complete, model: input.execution.provider.model ?? "unknown",
costUsd: accountingCost ?? null, usageSource: "per_run",
})}::jsonb`,
}).where(and(eq(heartbeatRuns.id, input.execution.binding.runId), eq(heartbeatRuns.companyId, input.execution.binding.companyId), isNull(heartbeatRuns.costAccountedAt)));
};
let completedConversationReply: PrpEvent | null = null;
const controlPlane = new PaperclipControlPlanePort(
input.db,
@@ -7854,6 +7870,25 @@ async function executePaperclipNativeSessionWithinScope(
},
{
onCommittedEvent: async (event) => {
const payload = record(event.payload);
const usage = record(payload.usage);
const delta = record(usage.runDelta);
if (event.eventType === "turn.started" && observedAccountingUsage) {
observedAccountingTurn = undefined;
await persistAccountingUsage(observedAccountingUsage, false);
}
// Only an explicit run delta is safe here: session totals can include
// earlier work. Save provisional evidence before ancillary finalizers.
if (payload.kind === "usage" && numericUsageField(delta, ["inputTokens"]) !== undefined
&& numericUsageField(delta, ["outputTokens"]) !== undefined) {
observedAccountingUsage = { runDelta: usage.runDelta };
observedAccountingTurn = event.turnId;
await persistAccountingUsage(observedAccountingUsage, false);
}
if (["turn.failed", "turn.cancelled", "turn.interrupted"].includes(event.eventType)
&& observedAccountingUsage && event.turnId === observedAccountingTurn) {
await persistAccountingUsage(observedAccountingUsage, true);
}
await toolTrace.observe(event);
if (event.eventType === "item.completed" &&
record(event.payload).kind === "agentMessage" &&
@@ -8602,18 +8637,8 @@ async function executePaperclipNativeSessionWithinScope(
);
// Persist provider accounting before any workspace/issue finalization. A
// detached controller or failed finalizer must not lose a completed turn.
const accountingUsage = normalizeNativeUsage(native.usage, { inputIncludesCacheReads: input.execution.provider.kind === "codex" });
const accountingCost = nativeUsageCostUsd(native.usage, input.execution.provider);
const billing = resolveNativeBilling(input.execution.provider, input.runnerEnvironment, input.billingIdentity);
await input.onUsage?.({ usage: accountingUsage, ...billing, complete: true, usageBasis: "per_run",
model: input.execution.provider.model ?? "unknown", costUsd: accountingCost ?? null });
if (!input.onUsage) await input.db.update(heartbeatRuns).set({
costAccountingPending: true,
usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || ${JSON.stringify({
...accountingUsage, ...billing, accountingReceiptReady: true, model: input.execution.provider.model ?? "unknown",
costUsd: accountingCost ?? null, usageSource: "per_run",
})}::jsonb`,
}).where(and(eq(heartbeatRuns.id, input.execution.binding.runId), eq(heartbeatRuns.companyId, input.execution.binding.companyId), isNull(heartbeatRuns.costAccountedAt)));
native = { ...native, usage: native.usage ?? observedAccountingUsage };
await persistAccountingUsage(native.usage, true);
try {
await completeManagedNativeCredentialTurn(managedCredentialSession);
} catch {