Merge credential and adapter review follow-ups into accounting stack

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

server/src/__tests__/adapter-cost-failure-paths.test.ts

@@ -86,6 +86,39 @@ describe("CLI adapter accounting on timeout", () => {
     onUsage.mockRejectedValue(new Error("Receipt persistence failed"));
     await expect(execute(context)).rejects.toThrow("Receipt persistence failed");
   });
+  it.each(cases.filter(f => ["pi", "opencode"].includes(f.name)))("$name preserves full-stream accounting after stdout truncation", async (fixture) => {
+    const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-tail-")); directories.push(dir);
+    const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'anthropic  test\\n'\n", { mode: 0o755 });
+    const execute = await fixture.execute();
+    for (const checkpoint of [true, false]) for (const timedOut of [false, true]) for (const missingPrice of [false, true]) {
+      const early = structuredClone(fixture.event) as any;
+      if (missingPrice) {
+        if (fixture.name === "pi") delete early.message.usage.cost;
+        else delete early.part.cost;
+      }
+      const end = fixture.name === "pi" && !timedOut ? '\n{"type":"agent_end","messages":[]}' : "";
+      const first = JSON.stringify(early) + "\n";
+      const noise = JSON.stringify({ type: "text", part: { text: "x".repeat(5 * 1024 * 1024) } }) + "\n";
+      const last = JSON.stringify(fixture.event) + end;
+      processResult.mockImplementation(async (_run, _target, _command, _args, options) => {
+        for (const chunk of [first, noise, last]) await options.onLog("stdout", chunk);
+        return { exitCode: timedOut ? null : 0, signal: timedOut ? "SIGTERM" : null, timedOut, stdout: (first + noise + last).slice(-4 * 1024 * 1024), stderr: "", pid: 123, startedAt: new Date().toISOString() };
+      });
+      const onUsage = checkpoint ? vi.fn() : undefined;
+      const result = await execute({
+        runId: "tail-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} },
+        runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null },
+        config: { engine: "cli", command, cwd: dir, model: "anthropic/test", paperclipRuntimeSkills: [], env: { ANTHROPIC_API_KEY: "fixture", OPENAI_BASE_URL: "https://unrelated-proxy.example/v1", OPENROUTER_API_KEY: "unrelated-key", OPENCODE_ALLOW_ALL_MODELS: "1" } },
+        context: {}, onLog: async () => {}, onUsage,
+      });
+      expect(result.usage).toMatchObject({ inputTokens: 40, outputTokens: 20, cachedInputTokens: 200 });
+      expect(result.costUsd).toBe(missingPrice ? null : 0.008);
+      expect(result.biller).toBe("anthropic");
+      expect(result.usageComplete).toBe(!timedOut);
+      if (onUsage) expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: result.usage, costUsd: result.costUsd });
+    }
+  });
+
   it.each(cases)("$name retains observed accounting when the process times out", async (fixture) => {
     const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-adapter-")); directories.push(dir);
     const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'openai  test\\n'\n", { mode: 0o755 });

server/src/__tests__/ai-connections.test.ts

@@ -84,6 +84,40 @@ describe("managed AI connections", () => {
     } finally { fetchSpy.mockRestore(); }
   });
 
+  it("serializes quota exchange with a concurrent secret edit and runtime credential read", async () => {
+    const owner = "quota-race-owner";
+    await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" });
+    const original = JSON.stringify({ tokens: { access_token: "old-access", refresh_token: "single-use", id_token: "identity", account_id: "race-account" } });
+    const account = await service.save(companyId, owner, { provider: "openai", method: "subscription", ownership: "personal", name: "Racing refresh", loginSessionId: "fixture", agentIds: [], allAgents: true }, original);
+    const [row] = await service.quotaAccounts(companyId, owner);
+    const secretId = row.grant.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!.secretId;
+    const exchangeStarted = deferredSignal();
+    const releaseExchange = deferredSignal();
+    const request = vi.spyOn(globalThis, "fetch").mockImplementation(async () => {
+      exchangeStarted.resolve(); await releaseExchange.promise;
+      return Response.json({ access_token: "new-access", refresh_token: "new-refresh" });
+    });
+    let runtime: Awaited<ReturnType<typeof prepareManagedAiRuntime>> | undefined;
+    const refreshing = service.refreshQuotaCredential(row, original, new AbortController().signal);
+    try {
+      await exchangeStarted.promise;
+      const edit = secretService(db).update(secretId, { description: "Changed during refresh" });
+      const preparing = prepareManagedAiRuntime(db, { companyId, agentId, responsibleUserId: owner, adapterType: "codex_local", binding: { provider: "openai", method: "subscription", mode: "responsible_user" }, config: {} });
+      // Wait for the runtime's old-token read to reach its bookkeeping update.
+      // After the refresh commits, resolution must detect the changed version.
+      await vi.waitFor(async () => {
+        const waiting = await db.execute(sql`select 1 from pg_stat_activity where datname = current_database() and wait_event_type = 'Lock' and query like '%last_resolved_at%'`);
+        expect(waiting.length).toBeGreaterThan(0);
+      });
+      releaseExchange.resolve();
+      const results = await Promise.all([refreshing, edit, preparing]);
+      runtime = results[2];
+      expect(JSON.parse(await readFile(path.join(String(runtime.config.env.CODEX_HOME), "auth.json"), "utf8"))).toMatchObject({ tokens: { access_token: "new-access", refresh_token: "new-refresh" } });
+      expect(JSON.parse(await service.credential(row))).toMatchObject({ tokens: { access_token: "new-access", refresh_token: "new-refresh" } });
+      expect((await secretService(db).getById(secretId))?.description).toBe("Changed during refresh");
+    } finally { releaseExchange.resolve(); await runtime?.cleanup(); request.mockRestore(); }
+  }, 10_000);
+
   it("serializes quota credential refresh, persists rotation, and defers while the provider is active", async () => {
     const owner = "quota-refresh-owner";
     await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" });
@@ -93,6 +127,14 @@ describe("managed AI connections", () => {
     const request = vi.spyOn(globalThis, "fetch").mockImplementation(async () => new Response(JSON.stringify({ access_token: "fresh-access", refresh_token: "rotated-once" })));
     const signal = new AbortController().signal;
     try {
+      // A pre-existing transaction may own a secret row before asking for the
+      // company file lock. Refresh must defer before exchanging its token.
+      await db.transaction(async tx => {
+        const secretId = row.grant.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!.secretId;
+        await tx.select().from(companySecrets).where(eq(companySecrets.id, secretId)).for("update");
+        await expect(service.refreshQuotaCredential(row, original, signal)).rejects.toThrow();
+        expect(request).not.toHaveBeenCalled();
+      });
       const activeId = randomUUID();
       await db.insert(heartbeatRuns).values({ id: activeId, companyId, agentId, invocationSource: "on_demand", status: "running", contextSnapshot: { aiConnection: { grantId: account.grantId } } });
       await expect(service.refreshQuotaCredential(row, original, signal)).rejects.toThrow("provider_unavailable");
@@ -1069,3 +1111,9 @@ describe("AI connection recovery delivery", () => {
     }, 30000,
   );
 });
+
+function deferredSignal() {
+  let resolve!: () => void;
+  const promise = new Promise<void>(done => { resolve = done; });
+  return { promise, resolve };
+}

server/src/services/ai-connection-runtime.ts

@@ -11,7 +11,7 @@ import {
 } from "@paperclipai/shared";
 import { aiConnectionService } from "./ai-connections.js";
 import { secretService } from "./secrets.js";
-import { decideCodexAuthMerge } from "@paperclipai/adapter-codex-local/server";
+import { decideCodexAuthMerge, withAccountHomeSecretMutationLock } from "@paperclipai/adapter-codex-local/server";
 import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
 import { runAdapterExecutionTargetProcess } from "@paperclipai/adapter-utils/execution-target";
 import { decideGrokAuthMerge } from "@paperclipai/adapter-grok-local/server";
@@ -314,7 +314,7 @@ export async function prepareManagedAiRuntime(
           if (subscriptionFile) {
             const refreshed = await readFile(authFile, "utf8");
             if (refreshed !== value)
-              await db.transaction(async (tx) => {
+              await withAccountHomeSecretMutationLock(undefined, input.companyId, () => db.transaction(async (tx) => {
                 const [grant] = await tx
                   .select()
                   .from(connectionGrants)
@@ -374,7 +374,7 @@ export async function prepareManagedAiRuntime(
                   .update(connectionGrants)
                   .set({ updatedAt: new Date() })
                   .where(eq(connectionGrants.id, grant.id));
-              });
+              }));
           }
         } finally {
           if (home) await rm(home, { recursive: true, force: true });

server/src/services/ai-connections.ts

@@ -1,3 +1,4 @@
+import { withAccountHomeSecretMutationLock } from "@paperclipai/adapter-codex-local/server";
 import { syncConnectionCredentialBindings } from "./connection-credential-bindings.js";
 import { createHash, randomUUID } from "node:crypto";
 import { and, eq, inArray, or, sql } from "drizzle-orm";
@@ -401,7 +402,7 @@ export function aiConnectionService(db: Db) {
       } satisfies AiConnectionAttribution,
     };
   }
-  async function credential(row: Pick<Awaited<ReturnType<typeof select>>, "connection" | "grant">) {
+  async function credential(row: Pick<Awaited<ReturnType<typeof select>>, "connection" | "grant">, retry = 0): Promise<string> {
     const ref = row.grant.credentialSecretRefs.find(
       (r) => r.configPath === "ai.credential",
     );
@@ -431,6 +432,7 @@ export function aiConnectionService(db: Db) {
       responsibleUserId: row.grant.subjectUserId,
       actorType: "system" as const,
     };
+    let value: string;
     if (secret.scope === "user") {
       if (
         secret.ownerUserId !== row.grant.subjectUserId ||
@@ -448,14 +450,21 @@ export function aiConnectionService(db: Db) {
         context,
       );
       if (!result) throw unprocessable("Reconnect this AI account");
-      return result.value;
+      value = result.value;
+    } else {
+      value = await secrets.resolveSecretValue(row.connection.companyId, secret.id, "latest", context);
     }
-    return secrets.resolveSecretValue(
-      row.connection.companyId,
-      secret.id,
-      "latest",
-      context,
-    );
+    // Resolution records lastResolvedAt and can wait behind a rotating writer.
+    // Re-read after that wait instead of handing a run the pre-rotation token.
+    const [latest] = await db.select().from(companySecrets).where(and(
+      eq(companySecrets.id, secret.id), eq(companySecrets.companyId, row.connection.companyId),
+    ));
+    if (!latest || latest.status !== "active") throw unprocessable("Reconnect this AI account");
+    if (latest.latestVersion !== secret.latestVersion) {
+      if (retry >= 2) throw unprocessable("AI credentials are changing. Retry this execution.", { code: "ai_connection_busy" });
+      return credential(row, retry + 1);
+    }
+    return value;
   }
   async function save(
     companyId: string,
@@ -498,7 +507,7 @@ export function aiConnectionService(db: Db) {
       );
     const id = reconnect?.connection.id ?? randomUUID();
     const grantId = reconnect?.grant.id ?? randomUUID();
-    return db.transaction(async (tx) => {
+    return withAccountHomeSecretMutationLock(undefined, companyId, () => db.transaction(async (tx) => {
       const secrets = secretService(tx);
       if (sessionId) {
         const [session] = await tx
@@ -803,7 +812,7 @@ export function aiConnectionService(db: Db) {
         details: { provider: input.provider, method: input.method, grantId },
       });
       return { connectionId: id, grantId };
-    });
+    }));
   }
   /** A late failure must never invalidate credentials that were refreshed or reconnected meanwhile. */
   async function markAuthenticationFailed(input: {
@@ -813,7 +822,7 @@ export function aiConnectionService(db: Db) {
     runStartedAt: Date;
     attribution: AiConnectionAttribution & { identity: string };
   }) {
-    return db.transaction(async (tx) => {
+    return withAccountHomeSecretMutationLock(undefined, input.companyId, () => db.transaction(async (tx) => {
       const { attribution } = input;
       const [grant] = await tx.select().from(connectionGrants).where(and(
         eq(connectionGrants.companyId, input.companyId),
@@ -847,22 +856,24 @@ export function aiConnectionService(db: Db) {
         entityType: "tool_connection", entityId: connection.id,
         details: { provider: attribution.provider, grantId: grant.id },
       });
-    });
+    }));
   }
   /** Refresh only the selected vaulted identity, serializing rotation and reconnect. */
   async function refreshQuotaCredential(row: Parameters<typeof credential>[0], failedValue: string, signal: AbortSignal) {
-    return db.transaction(async (tx) => {
+    return withAccountHomeSecretMutationLock(undefined, row.connection.companyId, () => db.transaction(async (tx) => {
+      // Do not wait on a transaction that might itself be waiting for the file
+      // lock. All database locks must be available before consuming OAuth tokens.
       const companyId = row.connection.companyId;
       const [grant] = await tx.select().from(connectionGrants).where(and(
         eq(connectionGrants.id, row.grant.id), eq(connectionGrants.companyId, companyId),
         eq(connectionGrants.connectionId, row.connection.id),
-      )).for("update");
+      )).for("update", { noWait: true });
       if (!grant || grant.status !== "active") throw new Error("credentials_unavailable");
       const ref = grant.credentialSecretRefs.find(ref => ref.configPath === "ai.credential");
       if (!ref) throw new Error("credentials_unavailable");
       await tx.select({ id: companySecrets.id }).from(companySecrets).where(and(
         eq(companySecrets.id, ref.secretId), eq(companySecrets.companyId, companyId),
-      )).for("update");
+      )).for("update", { noWait: true });
       const current = await aiConnectionService(tx as unknown as Db).credential({ connection: row.connection, grant });
       signal.throwIfAborted();
       // Another poll, an agent, or reconnect already supplied a newer credential.
@@ -904,7 +915,7 @@ export function aiConnectionService(db: Db) {
         action: "ai_connection.credential_refreshed", entityType: "tool_connection", entityId: row.connection.id,
         details: { provider: "openai", grantId: grant.id } });
       return refreshed;
-    });
+    }));
   }
   // A quota read uses the same credential audience as execution. Operator status
   // alone never grants access to another member's personal subscription.