fix: settle adapter usage only with terminal accounting evidence

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/adapter-utils/src/acpx-engine/execute.test.ts

@@ -1452,6 +1452,59 @@ describe("shared ACPX engine runtime behavior", () => {
     expect(result.billingType).toBe("unknown");
   });
 
+  it.each(["failed", "cancelled"] as const)("keeps %s in-stream usage provisional without a final receipt", async (status) => {
+    const root = await makeTempRoot();
+    const stateDir = path.join(root, "state");
+    const execute = createAcpxEngineExecutor({
+      createRuntime: () => ({
+        ensureSession: async () => ({
+          backendSessionId: "backend-session",
+          agentSessionId: "agent-session",
+          runtimeSessionName: "runtime-session",
+        }),
+        startTurn: () => ({
+          events: (async function* () {
+            yield {
+              type: "status",
+              text: "usage",
+              tag: "usage_update",
+              cost: { amount: 0.31, currency: "USD" },
+              breakdown: { inputTokens: 40, outputTokens: 700, cachedReadTokens: 60 },
+            };
+            
+          })(),
+          result: Promise.resolve({ status, error: { code: "INTERNAL", message: "fixture failure" } }),
+          cancel: async () => {},
+        }),
+        close: async () => {},
+      }) as never,
+    });
+
+    const onUsage = vi.fn();
+    const result = await execute({
+      runId: "run-usage-event-fallback",
+      agent: {
+        id: "agent-1",
+        companyId: "company-1",
+      },
+      runtime: {},
+      config: { agent: "custom", agentCommand: "node ./fake-acp.js", stateDir },
+      context: {},
+      onLog: async () => {},
+      onMeta: async () => {},
+      onUsage,
+    } as never);
+
+    expect(result.exitCode).not.toBe(0);
+    expect(result.usageComplete).toBe(false);
+    expect(onUsage).toHaveBeenLastCalledWith(expect.objectContaining({ complete: false, costUsd: 0.31 }));
+    expect(result.usage).toEqual({ inputTokens: 40, outputTokens: 700, cachedInputTokens: 60 });
+    expect(result.usageBasis).toBe("per_run");
+    expect(result.costUsd).toBeCloseTo(0.31);
+    expect(result.provider).toBe("acpx");
+    expect(result.billingType).toBe("unknown");
+  });
+
   it.skipIf(process.platform === "win32")("materializes ACPX Claude skills without symlinked descendants", async () => {
     const root = await makeTempRoot();
     const skillRoot = path.join(root, "skills");

packages/adapter-utils/src/acpx-engine/execute.ts

@@ -4995,8 +4995,11 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
           eventBreakdown,
           eventCostUsd,
         });
+        // 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);
         await ctx.onUsage?.({ ...billingFields, usage: turnUsage.usage ?? undefined, costUsd: turnUsage.costUsd,
-          model: prepared.requestedModel, usageBasis: "per_run", complete: !channelLost });
+          model: prepared.requestedModel, usageBasis: "per_run", complete: usageComplete });
         const failedTurn = terminal.status === "failed" || terminal.status === "cancelled" || timedOut;
         // ACPX can defer session/load until runTurn. Forget an unavailable
         // session so the next bounded turn receives the full task conversation.
@@ -5082,7 +5085,7 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) {
           model: prepared.requestedModel || null,
           ...(turnUsage.usage ? { usage: turnUsage.usage, usageBasis: "per_run" as const } : {}),
           costUsd: turnUsage.costUsd,
-          usageComplete: !channelLost,
+          usageComplete,
           resultJson: {
             status: channelLost ? "failed" : terminal.status,
             ...activityDiagnostics,

packages/adapters/cursor-local/src/server/execute.test.ts

@@ -138,6 +138,36 @@ function createFreshLeaseSandboxRunner(options: {
 }
 
 describe("cursor execute", () => {
+  it.each([0, 7])("settles legacy step usage only after a clean process exit (%s)", async (exitCode) => {
+    setPrepareCursorSandboxCommand.mockReset();
+    setPrepareCursorSandboxCommand.mockImplementation(async (input) => ({
+      command: input.command, env: input.env, remoteSystemHomeDir: null,
+      addedPathEntry: null, preferredCommandPath: null,
+    }));
+    const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-cursor-legacy-"));
+    const command = path.join(root, "agent.sh");
+    await fs.writeFile(command, `#!/bin/sh
+cat >/dev/null
+printf '%s\\n' '{"type":"step_finish","part":{"tokens":{"input":20,"output":5},"cost":0.01}}'
+exit ${exitCode}
+`, { mode: 0o755 });
+    const onUsage = vi.fn();
+    try {
+      const result = await execute({
+        runId: "run-legacy", agent: { id: "agent-1", companyId: "company-1", name: "Cursor", adapterType: "cursor", adapterConfig: {} },
+        runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null },
+        config: { command, cwd: root }, context: createPromptContextFixture(),
+        authToken: "fixture-run-token", onLog: async () => {}, onUsage,
+      });
+      expect(result.usage).toMatchObject({ inputTokens: 20, outputTokens: 5 });
+      expect(result.costUsd).toBe(0.01);
+      expect(result.usageComplete).toBe(exitCode === 0);
+      expect(onUsage).toHaveBeenCalledWith(expect.objectContaining({ complete: false, costUsd: 0.01 }));
+    } finally {
+      await fs.rm(root, { recursive: true, force: true });
+    }
+  });
+
   it.each([
     { detail: "Authentication failed", structured: "", expected: "Authentication failed" },
     { detail: "", structured: "", expected: "Cursor exited with code 7" },

packages/adapters/cursor-local/src/server/execute.ts

@@ -701,12 +701,15 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
     },
     clearSessionOnMissingSession = false,
   ): AdapterExecutionResult => {
+    const usageComplete = attempt.parsed.sawResult || (attempt.parsed.sawLegacyStep
+      && attempt.proc.exitCode === 0 && !attempt.proc.signal && !attempt.proc.timedOut
+      && !attempt.parsed.errorMessage);
     if (attempt.proc.timedOut) {
       return {
         exitCode: attempt.proc.exitCode,
         signal: attempt.proc.signal,
         timedOut: true,
-        usageComplete: attempt.parsed.sawResult,
+        usageComplete,
           usage: attempt.parsed.usage,
           usageBasis: "per_run",
           provider: providerFromModel,
@@ -746,7 +749,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
       exitCode: attempt.proc.exitCode,
       signal: attempt.proc.signal,
       timedOut: false,
-      usageComplete: attempt.parsed.sawResult,
+      usageComplete,
       usageBasis: "per_run",
       errorMessage:
         (attempt.proc.exitCode ?? 0) === 0

packages/adapters/cursor-local/src/server/parse.ts

@@ -68,6 +68,7 @@ export function parseCursorJsonl(stdout: string) {
 export function createCursorJsonlParser() {
   let sessionId: string | null = null;
   let sawResult = false;
+  let sawLegacyStep = false;
   const messages: string[] = [];
   let errorMessage: string | null = null;
   let totalCostUsd = 0;
@@ -156,6 +157,7 @@ export function createCursorJsonlParser() {
       }
 
       if (type === "step_finish") {
+        sawLegacyStep = true;
         const part = parseObject(event.part);
         const tokens = parseObject(part.tokens);
         const cache = parseObject(tokens.cache);
@@ -169,6 +171,7 @@ export function createCursorJsonlParser() {
 
     return {
       sawResult,
+      sawLegacyStep,
       sessionId,
       summary: messages.join("\n\n").trim(),
       usage: { ...usage },

packages/adapters/kimi-local/src/server/execute.test.ts

@@ -326,6 +326,9 @@ describe("kimi_local execute", () => {
     const result = await execute(makeContext(root, { config: { cwd: root, timeoutSec: 5 } }));
 
     expect(result.timedOut).toBe(true);
+    expect(result.usageComplete).toBe(true);
+    expect(result.costStatus).toBe("unpriced");
+    expect(result.costUsd).toBeNull();
     expect(result.errorMessage).toContain("5s");
   });
 

packages/adapters/kimi-local/src/server/execute.ts

@@ -648,7 +648,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
         exitCode: attempt.proc.exitCode,
         signal: attempt.proc.signal,
         timedOut: true,
-        usageComplete: false,
+        // This stream protocol has no usage receipt to recover after exit.
+        // Close collection with an explicitly unknown price, never a free run.
+        usageComplete: true,
+        costStatus: "unpriced",
         usageBasis: "per_run",
         provider: "moonshot",
         biller: "moonshot",
@@ -711,6 +714,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
       exitCode: attempt.proc.exitCode,
       signal: attempt.proc.signal,
       timedOut: false,
+      usageComplete: true,
+      costStatus: "unpriced",
       usageBasis: "per_run",
       errorMessage: failed ? fallbackErrorMessage : null,
       // Forward the transport-level error code from the run-disposition seam