fix: journal exchanged quota credentials before transactional save

Encrypt and fsync replacement OAuth tokens before vault/grant/activity writes. Retry failed saves from the journal without consuming the old refresh token again; retain durable recovery through rollback, commit failures, restart, and lost commit responses. Fence replay against revocation and reconnect.

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

doc/connections/AI-CONNECTIONS.md

@@ -540,3 +540,15 @@ the private runtime home for retry; cleanup deletes it only after a successful
 transaction or an intentional freshness/authorization discard. These retained
 homes are recovery evidence, not a background replay queue; persistent database
 or lock failures still require operator intervention.
+
+Quota OAuth replacement tokens are encrypted with the instance secrets master
+key and fsynced under `<instance-root>/quota-credential-recovery/<company-id>/`
+before vault, grant, or activity writes. Storage and encryption are checked before
+exchanging a single-use refresh token. Failed saves retry without another OAuth
+exchange; persistent failures keep the encrypted record for the next quota poll,
+including after a restart. The journal is removed only after database commit (or
+when an authorized newer credential makes it obsolete). Replay checks the grant,
+connection, secret, and original credential fingerprint and cannot reactivate a
+revoked grant or overwrite a reconnect. Back up this directory and the instance
+secrets key with the instance data. Loss of durable storage while receiving a
+provider token can still require reconnecting the account.

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

@@ -11,6 +11,7 @@ import { and, eq, sql } from "drizzle-orm";
 import { createDb, companies, agents, heartbeatRuns, companyMemberships, connectionGrants, connectionGrantDelegations, connectionGrantMembers, toolConnections, toolConnectionInstalls, aiConnectionDefaults, aiProviderDefaults, adapterAuthSessions, environments, issues, issueThreadInteractions, issueRecoveryActions, connectionIntentDeliveries, agentWakeupRequests, companySecrets, companySecretVersions, userSecretDefinitions, principalPermissionGrants } from "@paperclipai/db";
 import { startEmbeddedPostgresTestDatabase } from "@paperclipai/db/test-embedded-postgres";
 import { fetchCompanyQuotaWindows } from "../services/quota-windows.js";
+import { quotaCredentialRecoveryPath, quotaCredentialRecovery, quotaCredentialHash } from "../services/quota-credential-recovery.js";
 import { aiConnectionService } from "../services/ai-connections.js";
 import * as codexAdapter from "@paperclipai/adapter-codex-local/server";
 import { WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE } from "@paperclipai/adapter-utils/workspace-restore-merge";
@@ -104,6 +105,92 @@ describe("managed AI connections", () => {
     } finally { request.mockRestore(); }
   });
 
+  it.each(["rotation", "grant", "activity", "commit"] as const)("recovers exchanged OAuth tokens after %s persistence failure without another exchange", async (failure) => {
+    const owner = `quota-rollback-${failure}`;
+    await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" });
+    const original = JSON.stringify({ tokens: { access_token: `expired-${failure}`, refresh_token: "single-use", account_id: owner } });
+    await service.save(companyId, owner, { provider: "openai", method: "subscription", ownership: "personal", name: owner, loginSessionId: "fixture", agentIds: [], allAgents: true }, original);
+    const [row] = await service.quotaAccounts(companyId, owner);
+    const ref = row.grant.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!;
+    const table = failure === "grant" ? "connection_grants" : failure === "activity" ? "activity_log" : "company_secret_versions";
+    const condition = failure === "grant" ? `new.id = '${row.grant.id}'::uuid` : failure === "activity" ? `new.action = 'ai_connection.credential_refreshed'` : `new.secret_id = '${ref.secretId}'::uuid`;
+    await db.execute(sql.raw(`create function quota_save_failure() returns trigger language plpgsql as $$ begin
+      if ${condition} then raise exception 'injected quota save failure'; end if; return new; end $$;
+      create ${failure === "commit" ? "constraint " : ""}trigger quota_save_failure ${failure === "commit" ? "after" : "before"} ${failure === "grant" ? "update" : "insert"} on ${table}
+      ${failure === "commit" ? "deferrable initially deferred" : ""} for each row execute function quota_save_failure();`));
+    const request = vi.spyOn(globalThis, "fetch").mockResolvedValue(Response.json({ access_token: `fresh-${failure}`, refresh_token: `replacement-${failure}` }));
+    const file = quotaCredentialRecoveryPath(companyId, row.grant.id);
+    let triggerDropped = false;
+    try {
+      await expect(service.refreshQuotaCredential(row, original, new AbortController().signal)).rejects.toThrow();
+      expect(request).toHaveBeenCalledTimes(1);
+      expect(await service.credential(row)).toBe(original);
+      const encrypted = await readFile(file, "utf8");
+      expect(encrypted).not.toContain(`fresh-${failure}`);
+      expect(encrypted).not.toContain(`replacement-${failure}`);
+      await db.execute(sql.raw(`drop trigger quota_save_failure on ${table}; drop function quota_save_failure();`));
+      triggerDropped = true;
+      // A new service instance recovers from disk, without an in-memory cache.
+      const recovered = aiConnectionService(db);
+      await recovered.refreshQuotaCredential(row, original, new AbortController().signal);
+      expect(JSON.parse(await recovered.credential(row))).toMatchObject({ tokens: { access_token: `fresh-${failure}`, refresh_token: `replacement-${failure}` } });
+      expect(request).toHaveBeenCalledTimes(1);
+      await expect(access(file)).rejects.toThrow();
+      expect((await db.select().from(companySecretVersions).where(eq(companySecretVersions.secretId, ref.secretId))).map(version => version.version).sort()).toEqual([1, 2]);
+    } finally {
+      request.mockRestore();
+      if (!triggerDropped) await db.execute(sql.raw(`drop trigger quota_save_failure on ${table}; drop function quota_save_failure();`));
+    }
+  });
+
+  it.each(["rollback", "lost_commit_reply"] as const)("automatically saves one exchanged token across %s", async (failure) => {
+    const owner = `quota-once-${failure}`;
+    await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" });
+    const original = JSON.stringify({ tokens: { access_token: "expired", refresh_token: "single-use", account_id: owner } });
+    await service.save(companyId, owner, { provider: "openai", method: "subscription", ownership: "personal", name: owner, loginSessionId: "fixture", agentIds: [], allAgents: true }, original);
+    const [row] = await service.quotaAccounts(companyId, owner);
+    const transaction = db.transaction.bind(db);
+    const interrupted = vi.spyOn(db, "transaction").mockImplementationOnce(async (work, config) => {
+      await transaction(async (tx) => {
+        await work(tx);
+        if (failure === "rollback") throw new Error("Injected rollback after writes");
+      }, config);
+      throw new Error("Injected lost commit response");
+    });
+    const request = vi.spyOn(globalThis, "fetch").mockResolvedValue(Response.json({ access_token: "fresh-once", refresh_token: "replacement-once" }));
+    try {
+      await service.refreshQuotaCredential(row, original, new AbortController().signal);
+      expect(JSON.parse(await service.credential(row))).toMatchObject({ tokens: { access_token: "fresh-once", refresh_token: "replacement-once" } });
+      expect(request).toHaveBeenCalledTimes(1);
+      await expect(access(quotaCredentialRecoveryPath(companyId, row.grant.id))).rejects.toThrow();
+    } finally { interrupted.mockRestore(); request.mockRestore(); }
+  });
+
+  it.each(["reconnected", "revoked"] as const)("does not apply a pending refresh over a %s identity", async (change) => {
+    const owner = `quota-replay-${change}`;
+    await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" });
+    const original = JSON.stringify({ tokens: { access_token: "expired", refresh_token: "single-use", account_id: owner } });
+    await service.save(companyId, owner, { provider: "openai", method: "subscription", ownership: "personal", name: owner, loginSessionId: "fixture", agentIds: [], allAgents: true }, original);
+    const [row] = await service.quotaAccounts(companyId, owner);
+    const ref = row.grant.credentialSecretRefs.find(ref => ref.configPath === "ai.credential")!;
+    const journal = quotaCredentialRecovery(companyId, row.grant.id);
+    await journal.write({ companyId, grantId: row.grant.id, connectionId: row.connection.id, secretId: ref.secretId, baseHash: quotaCredentialHash(original), value: "old-pending-replacement" });
+    const request = vi.spyOn(globalThis, "fetch");
+    try {
+      if (change === "reconnected") {
+        await secretService(db).rotate(ref.secretId, { value: "new-authorized-identity" });
+        expect(await service.refreshQuotaCredential(row, original, new AbortController().signal)).toBe("new-authorized-identity");
+        expect(await service.credential(row)).toBe("new-authorized-identity");
+        expect(await journal.read()).toBeNull();
+      } else {
+        await db.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.id, row.grant.id));
+        await expect(service.refreshQuotaCredential(row, original, new AbortController().signal)).rejects.toThrow("credentials_unavailable");
+        expect(await journal.read()).not.toBeNull();
+      }
+      expect(request).not.toHaveBeenCalled();
+    } finally { request.mockRestore(); await journal.clear(); }
+  });
+
   it("saves exchanged tokens when the quota deadline expires during response-body reading", async () => {
     const owner = "quota-body-deadline-owner";
     await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" });

server/src/services/ai-connections.ts

@@ -34,6 +34,8 @@ import {
   type AiConnectionLoginIntent,
 } from "@paperclipai/shared";
 import { forbidden, notFound, unprocessable } from "../errors.js";
+import { quotaCredentialRecovery, quotaCredentialHash } from "./quota-credential-recovery.js";
+import { logger } from "../middleware/logger.js";
 import { logActivity } from "./activity-log.js";
 import { secretService } from "./secrets.js";
 import { probeAiConnectionUsage } from "./ai-connection-usage.js";
@@ -860,66 +862,96 @@ export function aiConnectionService(db: Db) {
   }
   /** Refresh only the selected vaulted identity, serializing rotation and reconnect. */
   async function refreshQuotaCredential(row: Parameters<typeof credential>[0], failedValue: string, signal: AbortSignal) {
-    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", { 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", { 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.
-      if (current !== failedValue) return current;
-      // An active provider may rotate this single-use token itself. Defer polling
-      // until its credential write-back instead of invalidating its live copy.
-      const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
-        .innerJoin(agents, and(eq(agents.id, heartbeatRuns.agentId), eq(agents.companyId, companyId))).where(and(
-        eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.status, ["queued", "running"]),
-        or(sql`${heartbeatRuns.contextSnapshot}->'aiConnection'->>'grantId' = ${grant.id}`,
-          sql`${agents.runtimeConfig}->'aiConnection'->>'provider' = 'openai'`, eq(agents.adapterType, "codex_local")),
-      )).limit(1);
-      if (active) throw new Error("provider_unavailable");
-      const auth = JSON.parse(current);
-      const refreshToken = auth.tokens?.refresh_token;
-      if (typeof refreshToken !== "string" || !refreshToken) throw new Error("authentication_required");
-      // Once exchange starts, its lifetime is independent of the dashboard's
-      // shorter deadline. A consumed single-use token must still be read/saved.
-      signal.throwIfAborted();
-      const refreshSignal = AbortSignal.timeout(60_000);
-      // Matches the Codex OAuth client (openai/codex, login/src/auth/manager.rs).
-      const response = await fetch("https://auth.openai.com/oauth/token", {
-        method: "POST", redirect: "error", signal: refreshSignal,
-        headers: { "Content-Type": "application/json" },
-        body: JSON.stringify({ client_id: "app_EMoamEEZ73f0CkXaXp7hrann", grant_type: "refresh_token", refresh_token: refreshToken }),
+    return withAccountHomeSecretMutationLock(undefined, row.connection.companyId, async () => {
+      const recovery = quotaCredentialRecovery(row.connection.companyId, row.grant.id);
+      let savingReplacement = false;
+      const save = () => 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", { 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", { 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.
+        if (current !== failedValue) return current;
+        const pending = await recovery.read();
+        if (pending && (pending.secretId !== ref.secretId || pending.connectionId !== row.connection.id || pending.baseHash !== quotaCredentialHash(current))) {
+          throw new Error("credentials_unavailable");
+        }
+        // An active provider may rotate this single-use token itself. Defer polling
+        // until its credential write-back instead of invalidating its live copy.
+        const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
+          .innerJoin(agents, and(eq(agents.id, heartbeatRuns.agentId), eq(agents.companyId, companyId))).where(and(
+          eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.status, ["queued", "running"]),
+          or(sql`${heartbeatRuns.contextSnapshot}->'aiConnection'->>'grantId' = ${grant.id}`,
+            sql`${agents.runtimeConfig}->'aiConnection'->>'provider' = 'openai'`, eq(agents.adapterType, "codex_local")),
+        )).limit(1);
+        if (active && !pending) throw new Error("provider_unavailable");
+        let refreshed = pending?.value;
+        if (!refreshed) {
+          await recovery.prepare();
+          const auth = JSON.parse(current);
+          const refreshToken = auth.tokens?.refresh_token;
+          if (typeof refreshToken !== "string" || !refreshToken) throw new Error("authentication_required");
+          // Once exchange starts, its lifetime is independent of the dashboard's
+          // shorter deadline. A consumed single-use token must still be read/saved.
+          signal.throwIfAborted();
+          const refreshSignal = AbortSignal.timeout(60_000);
+          // Matches the Codex OAuth client (openai/codex, login/src/auth/manager.rs).
+          const response = await fetch("https://auth.openai.com/oauth/token", {
+            method: "POST", redirect: "error", signal: refreshSignal,
+            headers: { "Content-Type": "application/json" },
+            body: JSON.stringify({ client_id: "app_EMoamEEZ73f0CkXaXp7hrann", grant_type: "refresh_token", refresh_token: refreshToken }),
+          });
+          if (!response.ok) {
+            await response.body?.cancel();
+            throw new Error(response.status === 400 || response.status === 401 ? "authentication_required" : "provider_unavailable");
+          }
+          const tokens = await response.json() as Record<string, unknown>;
+          if (typeof tokens.access_token !== "string" || !tokens.access_token) throw new Error("provider_unavailable");
+          refreshed = JSON.stringify({ ...auth, tokens: { ...auth.tokens,
+            access_token: tokens.access_token,
+            ...(typeof tokens.refresh_token === "string" && tokens.refresh_token ? { refresh_token: tokens.refresh_token } : {}),
+            ...(typeof tokens.id_token === "string" && tokens.id_token ? { id_token: tokens.id_token } : {}),
+          }, last_refresh: new Date().toISOString() });
+          // This fsynced encrypted journal survives transaction rollback and host
+          // restart. Never exchange the consumed token again after a failed save.
+          await recovery.write({ companyId, grantId: grant.id, connectionId: row.connection.id,
+            secretId: ref.secretId, baseHash: quotaCredentialHash(current), value: refreshed });
+        }
+        savingReplacement = true;
+        // Persist the rotated token even if the quota deadline expires after the
+        // successful exchange; losing it would strand all subsequent executions.
+        await secretService(tx).rotate(ref.secretId, { value: refreshed }, { userId: grant.subjectUserId });
+        await tx.update(connectionGrants).set({ updatedAt: new Date() }).where(eq(connectionGrants.id, grant.id));
+        await logActivity(tx as unknown as Db, { companyId, actorType: "system", actorId: "quota",
+          action: "ai_connection.credential_refreshed", entityType: "tool_connection", entityId: row.connection.id,
+          details: { provider: "openai", grantId: grant.id } });
+        return refreshed;
       });
-      if (!response.ok) {
-        await response.body?.cancel();
-        throw new Error(response.status === 400 || response.status === 401 ? "authentication_required" : "provider_unavailable");
+      for (let attempt = 0; ; attempt++) {
+        try {
+          const value = await save();
+          // Clear only after commit. A committed-but-lost response is detected
+          // by the current credential comparison on the next attempt.
+          await recovery.clear().catch((error) => logger.warn({ err: error, grantId: row.grant.id }, "Saved quota credential recovery cleanup pending"));
+          return value;
+        } catch (error) {
+          if (!savingReplacement || attempt >= 2) throw error;
+          // Saving retries have no provider side effect. Persistent failures
+          // retain the journal for the next quota poll, including after restart.
+        }
       }
-      const tokens = await response.json() as Record<string, unknown>;
-      if (typeof tokens.access_token !== "string" || !tokens.access_token) throw new Error("provider_unavailable");
-      const refreshed = JSON.stringify({ ...auth, tokens: { ...auth.tokens,
-        access_token: tokens.access_token,
-        ...(typeof tokens.refresh_token === "string" && tokens.refresh_token ? { refresh_token: tokens.refresh_token } : {}),
-        ...(typeof tokens.id_token === "string" && tokens.id_token ? { id_token: tokens.id_token } : {}),
-      }, last_refresh: new Date().toISOString() });
-      // Persist the rotated token even if the quota deadline expires after the
-      // successful exchange; losing it would strand all subsequent executions.
-      await secretService(tx).rotate(ref.secretId, { value: refreshed }, { userId: grant.subjectUserId });
-      await tx.update(connectionGrants).set({ updatedAt: new Date() }).where(eq(connectionGrants.id, grant.id));
-      await logActivity(tx as unknown as Db, { companyId, actorType: "system", actorId: "quota",
-        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.

server/src/services/quota-credential-recovery.ts

@@ -0,0 +1,59 @@
+import { createHash, randomUUID } from "node:crypto";
+import { promises as fs } from "node:fs";
+import path from "node:path";
+import { z } from "zod";
+import { resolvePaperclipInstanceRoot } from "../home-paths.js";
+import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js";
+
+const recoverySchema = z.object({
+  companyId: z.string().uuid(), grantId: z.string().uuid(), connectionId: z.string().uuid(), secretId: z.string().uuid(),
+  baseHash: z.string().length(64), value: z.string().min(1),
+});
+type Recovery = z.infer<typeof recoverySchema>;
+export const quotaCredentialHash = (value: string) => createHash("sha256").update(value).digest("hex");
+export const quotaCredentialRecoveryPath = (companyId: string, grantId: string) =>
+  path.join(resolvePaperclipInstanceRoot(), "quota-credential-recovery", companyId, `${grantId}.json`);
+
+async function syncDirectory(directory: string) {
+  if (process.platform === "win32") return;
+  const handle = await fs.open(directory, "r");
+  try { await handle.sync(); } finally { await handle.close(); }
+}
+
+/** Called only while holding the company credential mutation lock. The local
+ * encrypted provider uses the instance secrets master key, including when the
+ * destination vault is remote. No plaintext token enters this journal. */
+export function quotaCredentialRecovery(companyId: string, grantId: string) {
+  const file = quotaCredentialRecoveryPath(companyId, grantId);
+  const directory = path.dirname(file);
+  async function writeEncrypted(target: string, value: string) {
+    const encrypted = await localEncryptedProvider.createVersion({ value });
+    await fs.mkdir(directory, { recursive: true, mode: 0o700 });
+    const temporary = `${target}.${randomUUID()}.tmp`;
+    const handle = await fs.open(temporary, "wx", 0o600);
+    try {
+      try { await handle.writeFile(JSON.stringify(encrypted.material)); await handle.sync(); }
+      finally { await handle.close(); }
+      await fs.rename(temporary, target);
+      await syncDirectory(directory);
+    } catch (error) { await fs.rm(temporary, { force: true }); throw error; }
+  }
+  return {
+    // Verify encryption, storage, and fsync before exchanging a single-use token.
+    prepare: async () => {
+      const probe = `${file}.${randomUUID()}.probe`;
+      try { await writeEncrypted(probe, "credential-recovery-probe"); }
+      finally { await fs.rm(probe, { force: true }); }
+    },
+    read: async (): Promise<Recovery | null> => {
+      let material: string;
+      try { material = await fs.readFile(file, "utf8"); }
+      catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; throw error; }
+      const pending = recoverySchema.parse(JSON.parse(await localEncryptedProvider.resolveVersion({ material: JSON.parse(material), externalRef: null })));
+      if (pending.companyId !== companyId || pending.grantId !== grantId) throw new Error("credentials_unavailable");
+      return pending;
+    },
+    write: (input: Recovery) => writeEncrypted(file, JSON.stringify(recoverySchema.parse(input))),
+    clear: async () => { await fs.rm(file, { force: true }); },
+  };
+}