diff --git a/doc/connections/AI-CONNECTIONS.md b/doc/connections/AI-CONNECTIONS.md index ec85824d7e7..1bbb8ad26fb 100644 --- a/doc/connections/AI-CONNECTIONS.md +++ b/doc/connections/AI-CONNECTIONS.md @@ -580,22 +580,33 @@ 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 `/quota-credential-recovery//` -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. New OpenAI subscription runtimes serialize credential -reads with quota exchanges and recover a matching pending replacement before -materializing an auth home. Authentication-failure handling also recovers a -matching replacement before marking the grant invalid. A failed recovery save -defers these actions and leaves both the active grant and journal intact. -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-owned quota refresh validates and locks the current grant and secret in +Postgres before exchanging a token. A failed database prerequisite prevents the +exchange. A returned replacement is saved directly into the existing secret +store. The same logical operation keeps it in memory for at most three save +attempts. A lost commit response is reconciled against the current saved value; +reconnect, revocation, secret-version and session-epoch changes win over a delayed +save. Save retries never repeat the OAuth exchange and no new recovery file is +written for this path, in cloud or local deployments. + +If all save attempts fail, the request fails. If the server process then exits, +the unsaved replacement is lost. V1 accepts signing in again after the next failed +use; it does not report successful refresh or keep a second credential vault. + +Remote Codex Runner tasks retain their private credential until secret writeback +and the lease's credential handoff commit. Provider completion follows that +handoff. Failed exports or database saves leave it pending for the existing +cleanup path, including a replacement web receiver. Its authority still checks +the company, grant, secret, version selector and session epoch. + +Existing encrypted records in +`/quota-credential-recovery//` remain readable for +cutover settlement. Do not delete unresolved files or remove the old volume +before inventory and settlement. Legacy replay cannot reactivate a revoked grant +or overwrite a reconnect. Local Google Antigravity retirement still uses its +existing encrypted recovery record and private-home ownership; remote and warm +Google subscription sessions remain unsupported. This local path is not a cloud +fallback. Cloud cutover must settle any observed legacy credential files first. ## Advanced provider routing (2026-10-02) diff --git a/server/src/__tests__/ai-connections.test.ts b/server/src/__tests__/ai-connections.test.ts index 463a6463c25..e408200aa50 100644 --- a/server/src/__tests__/ai-connections.test.ts +++ b/server/src/__tests__/ai-connections.test.ts @@ -9,7 +9,8 @@ vi.mock("../services/subscription-refresh.js", () => ({ refreshSubscriptionConne import * as localCredentials from "../services/local-ai-credentials.js"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { createHash, randomUUID } from "node:crypto"; -import { spawnSync } from "node:child_process"; +import { spawn, spawnSync } from "node:child_process"; +import { createRequire } from "node:module"; import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js"; import { promises as fs } from "node:fs"; import { mkdtemp, realpath, rm, access, readFile, writeFile, stat } from "node:fs/promises"; @@ -38,6 +39,8 @@ import { secretService } from "../services/secrets.js"; import { resolveExecutionRunAdapterConfig } from "../services/heartbeat.js"; import { aiConnectionBindingSchema, connectionPurposeTransportSchema, isAiConnectionCompatible } from "@paperclipai/shared"; import { openRouterClaudeHeaders } from "@paperclipai/shared/openrouter-attribution"; +import { createServer } from "node:http"; +import { once } from "node:events"; import express from "express"; import request from "supertest"; import { aiConnectionRoutes, canInstallSharedAiConnectionForNewAgent, responsibleUserForAiRequest } from "../routes/ai-connections.js"; @@ -385,8 +388,8 @@ describe("managed AI connections", () => { await expectSelectionFailure(service.select({ ...input, userId: owner }), "ai_connection_unavailable"); }); - it("persists first-time recovery directories before consuming a single-use token", async () => { - const owner = "quota-durable-path"; + it("commits one exchanged token without writable credential recovery storage", async () => { + const owner = "quota-database-first"; await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" }); const original = JSON.stringify({ tokens: { account_id: owner, access_token: "expired", refresh_token: "single-use" } }); await service.save(companyId, owner, { provider: "openai", method: "subscription", ownership: "personal", name: owner, loginSessionId: "fixture", agentIds: [], allAgents: true }, original); @@ -394,32 +397,32 @@ describe("managed AI connections", () => { const file = quotaCredentialRecoveryPath(companyId, row.grant.id); const recoveryRoot = path.dirname(path.dirname(file)); await expect(access(recoveryRoot)).rejects.toMatchObject({ code: "ENOENT" }); - const synced: string[] = []; - let rejectRoot = true; - const realOpen = fs.open.bind(fs); + const writes: string[] = []; + const rejectRecoveryWrite = (target: unknown) => { + if (String(target).startsWith(recoveryRoot)) { + writes.push(String(target)); + throw Object.assign(new Error("Fixture recovery storage is read-only"), { code: "EROFS" }); + } + }; + const realOpen = fs.open.bind(fs), realMkdir = fs.mkdir.bind(fs); const open = vi.spyOn(fs, "open").mockImplementation(async (...args: Parameters) => { - const handle = await realOpen(...args); - const sync = handle.sync.bind(handle); - vi.spyOn(handle, "sync").mockImplementation(async () => { - if (rejectRoot && args[0] === recoveryRoot) throw new Error("Injected recovery directory flush failure"); - await sync(); synced.push(String(args[0])); - }); - return handle; + rejectRecoveryWrite(args[0]); + return realOpen(...args); }); - const request = vi.spyOn(globalThis, "fetch").mockImplementation(async () => { - expect(synced).toContain(recoveryRoot); - expect(synced).toContain(path.dirname(recoveryRoot)); - return Response.json({ access_token: "fresh-durable", refresh_token: "replacement-durable" }); + const mkdir = vi.spyOn(fs, "mkdir").mockImplementation(async (...args: Parameters) => { + rejectRecoveryWrite(args[0]); + return realMkdir(...args); }); + const request = vi.spyOn(globalThis, "fetch").mockResolvedValue(Response.json({ access_token: "fresh-direct", refresh_token: "replacement-direct" })); try { - await expect(service.refreshQuotaCredential(row, original, new AbortController().signal)).rejects.toThrow("Injected recovery directory flush failure"); - expect(request).not.toHaveBeenCalled(); - expect(await service.credential(row)).toBe(original); - rejectRoot = false; - await service.refreshQuotaCredential(row, original, new AbortController().signal); + const refreshed = await service.refreshQuotaCredential(row, original, new AbortController().signal); + expect(JSON.parse(refreshed)).toMatchObject({ tokens: { access_token: "fresh-direct", refresh_token: "replacement-direct" } }); + expect(await service.credential(row)).toBe(refreshed); expect(request).toHaveBeenCalledTimes(1); - expect(JSON.parse(await service.credential(row))).toMatchObject({ tokens: { access_token: "fresh-durable", refresh_token: "replacement-durable" } }); - } finally { open.mockRestore(); request.mockRestore(); } + expect(writes).toEqual([]); + await expect(access(file)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(access(recoveryRoot)).rejects.toMatchObject({ code: "ENOENT" }); + } finally { open.mockRestore(); mkdir.mockRestore(); request.mockRestore(); } }); it.each(["timeout", "unexpected"] as const)("classifies a manual probe's %s lock failure before starting the provider", async (kind) => { @@ -703,7 +706,7 @@ describe("managed AI connections", () => { } finally { request.mockRestore(); } }); - it.each(["rotation", "secret_metadata", "activity", "commit"] as const)("recovers exchanged OAuth tokens after %s persistence failure without another exchange", async (failure) => { + it.each(["rotation", "secret_metadata", "activity", "commit"] as const)("fails truthfully after %s persistence failure and permits sign-in recovery without a journal", 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 } }); @@ -723,17 +726,25 @@ describe("managed AI connections", () => { 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 expect(access(file)).rejects.toMatchObject({ code: "ENOENT" }); 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}` } }); + // A new receiver has no replacement to recover. A failed use routes to the + // existing sign-in flow, rather than claiming the failed save succeeded. + const receiver = aiConnectionService(db); + const generation = createHash("sha256").update(original).digest("hex").slice(0, 16); + await receiver.markAuthenticationFailed({ companyId, runStartedAt: new Date(), attribution: { + provider: "openai", method: "subscription", connectionId: row.connection.id, grantId: row.grant.id, + mode: "responsible_user", responsibleUserId: owner, identity: `${row.grant.id}:${owner}:${generation}`, + } }); + expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, row.grant.id)))[0].status).toBe("needs_reauthorization"); + const signedIn = JSON.stringify({ tokens: { access_token: "signed-in", refresh_token: "new-login" } }); + await receiver.save(companyId, owner, { connectionId: row.connection.id, provider: "openai", method: "subscription", + ownership: "personal", name: owner, loginSessionId: "fixture", agentIds: [], allAgents: true }, signedIn); + const [current] = await receiver.quotaAccounts(companyId, owner); + expect(current.grant.status).toBe("active"); + expect(await receiver.runtimeCredential(current)).toBe(signedIn); 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(); @@ -819,6 +830,112 @@ describe("managed AI connections", () => { } finally { request.mockRestore(); await oldRun.cleanup(); await nextRun?.cleanup(); await journal.clear(); } }); + it("requires sign-in after the receiver process dies with an exchanged but unsaved token", async () => { + const owner = "quota-receiver-process-loss"; + await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" }); + const original = JSON.stringify({ tokens: { access_token: "expired-process", refresh_token: "single-use-process" } }); + 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 held = deferredSignal(), release = deferredSignal(); + const lockKey = 7192031; + const holder = db.transaction(async tx => { + await tx.execute(sql`select pg_advisory_xact_lock(${lockKey})`); + held.resolve(); await release.promise; + }); + await held.promise; + await db.execute(sql.raw(`create function hold_receiver_save() returns trigger language plpgsql as $$ begin + if new.secret_id = '${ref.secretId}'::uuid then perform pg_advisory_xact_lock(${lockKey}); end if; return new; end $$; + create trigger hold_receiver_save before insert on company_secret_versions for each row execute function hold_receiver_save();`)); + let exchanges = 0; + const provider = createServer((req, res) => { + exchanges++; req.resume(); + res.setHeader("Content-Type", "application/json"); + res.end(JSON.stringify({ access_token: "unsaved-process", refresh_token: "replacement-process" })); + }); + provider.listen(0, "127.0.0.1"); await once(provider, "listening"); + const address = provider.address() as { port: number }; + const child = spawn(process.execPath, ["--import", createRequire(import.meta.url).resolve("tsx"), new URL("./helpers/quota-credential-process.ts", import.meta.url).pathname], { + env: { ...process.env, CREDENTIAL_TEST_DATABASE_URL: database.connectionString, + CREDENTIAL_TEST_PROVIDER_URL: `http://127.0.0.1:${address.port}`, CREDENTIAL_TEST_COMPANY_ID: companyId, CREDENTIAL_TEST_OWNER: owner }, + stdio: "ignore", + }); + const exited = once(child, "exit"); + try { + let blocked = false; + const deadline = Date.now() + 25_000; + while (!blocked && Date.now() < deadline && child.exitCode === null) { + const rows = await db.execute(sql`select pid from pg_stat_activity where datname = current_database() + and wait_event_type = 'Lock' and wait_event = 'advisory'`); + blocked = exchanges === 1 && rows.length > 0; + if (!blocked) await new Promise(resolve => setTimeout(resolve, 25)); + } + expect(blocked).toBe(true); + child.kill("SIGKILL"); await exited; + } finally { + if (child.exitCode === null && child.signalCode === null) { child.kill("SIGKILL"); await exited; } + release.resolve(); await holder; + await db.execute(sql`drop trigger hold_receiver_save on company_secret_versions; drop function hold_receiver_save()`); + await new Promise(resolve => provider.close(() => resolve())); + } + expect(exchanges).toBe(1); + expect(await service.credential(row)).toBe(original); + await expect(access(quotaCredentialRecoveryPath(companyId, row.grant.id))).rejects.toMatchObject({ code: "ENOENT" }); + const receiver = aiConnectionService(db); + const generation = createHash("sha256").update(original).digest("hex").slice(0, 16); + await receiver.markAuthenticationFailed({ companyId, runStartedAt: new Date(), attribution: { + provider: "openai", method: "subscription", connectionId: row.connection.id, grantId: row.grant.id, + mode: "responsible_user", responsibleUserId: owner, identity: `${row.grant.id}:${owner}:${generation}`, + } }); + expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, row.grant.id)))[0].status).toBe("needs_reauthorization"); + const signedIn = JSON.stringify({ tokens: { access_token: "process-sign-in", refresh_token: "process-new-login" } }); + await receiver.save(companyId, owner, { connectionId: row.connection.id, provider: "openai", method: "subscription", ownership: "personal", name: owner, loginSessionId: "fixture", agentIds: [], allAgents: true }, signedIn); + const [current] = await receiver.quotaAccounts(companyId, owner); + expect(await receiver.runtimeCredential(current)).toBe(signedIn); + }, 40_000); + + it("does not exchange a token when a required database transaction fails", async () => { + const owner = "quota-db-prerequisite"; + 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" } }); + 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 request = vi.spyOn(globalThis, "fetch"); + const unavailableUrl = new URL(database.connectionString); + unavailableUrl.port = "1"; + const unavailable = createDb(unavailableUrl.toString()); + try { + await expect(aiConnectionService(unavailable).refreshQuotaCredential(row, original, new AbortController().signal)).rejects.toThrow(); + expect(request).not.toHaveBeenCalled(); + await expect(access(quotaCredentialRecoveryPath(companyId, row.grant.id))).rejects.toMatchObject({ code: "ENOENT" }); + } finally { await unavailable.$client.end(); request.mockRestore(); } + }); + + it.each(["reconnect", "same_value_new_epoch", "revoked"] as const)("lets %s win after a returned token's save rolls back", async change => { + const owner = `quota-delayed-save-${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" } }); + 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 transaction = db.transaction.bind(db); + const interrupted = vi.spyOn(db, "transaction").mockImplementationOnce(async (work, config) => { + await transaction(async tx => { await work(tx); throw new Error("fixture rollback"); }, config).catch(() => {}); + if (change === "revoked") await db.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.id, row.grant.id)); + else await secretService(db).rotate(ref.secretId, { value: change === "reconnect" ? "new-login" : original }); + throw new Error("fixture lost response"); + }); + const request = vi.spyOn(globalThis, "fetch").mockResolvedValue(Response.json({ access_token: "late", refresh_token: "late-refresh" })); + try { + const refresh = service.refreshQuotaCredential(row, original, new AbortController().signal); + if (change === "revoked") await expect(refresh).rejects.toThrow("credentials_unavailable"); + else expect(await refresh).toBe(change === "reconnect" ? "new-login" : original); + expect(await service.credential(row)).toBe(change === "reconnect" ? "new-login" : original); + expect(request).toHaveBeenCalledTimes(1); + await expect(access(quotaCredentialRecoveryPath(companyId, row.grant.id))).rejects.toMatchObject({ code: "ENOENT" }); + } finally { interrupted.mockRestore(); request.mockRestore(); } + }); + 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" }); @@ -863,7 +980,10 @@ describe("managed AI connections", () => { triggerInstalled = true; await expect(service.refreshQuotaCredential(row, current, new AbortController().signal)).rejects.toThrow(); expect(await service.credential(row)).toBe(current); - expect(JSON.parse((await journal.read())!.value).tokens.access_token).toBe("C"); + expect(await journal.read()).toBeNull(); + // Only observed records from the earlier release use the legacy reader. + await journal.write({ companyId, grantId: row.grant.id, connectionId: row.connection.id, secretId: ref.secretId, + baseHash: quotaCredentialHash(current), value: JSON.stringify({ tokens: { access_token: "C", refresh_token: "refresh-C" } }) }); await db.execute(sql`drop trigger delayed_quota_commit_failure on company_secret_versions; drop function delayed_quota_commit_failure()`); triggerInstalled = false; const recovered = await service.refreshQuotaCredential(row, original, new AbortController().signal); diff --git a/server/src/__tests__/helpers/quota-credential-process.ts b/server/src/__tests__/helpers/quota-credential-process.ts new file mode 100644 index 00000000000..3914dcfe7e5 --- /dev/null +++ b/server/src/__tests__/helpers/quota-credential-process.ts @@ -0,0 +1,19 @@ +// A real receiver process for the credential-loss boundary test. No provider +// credentials leave this process through logs, IPC, or diagnostic exceptions. +import { createDb } from "@paperclipai/db"; +import { aiConnectionService } from "../../services/ai-connections.js"; + +const db = createDb(process.env.CREDENTIAL_TEST_DATABASE_URL!); +const provider = process.env.CREDENTIAL_TEST_PROVIDER_URL!; +const request = globalThis.fetch; +globalThis.fetch = (url, init) => { + if (String(url) !== "https://auth.openai.com/oauth/token") throw new Error("Unexpected fixture network request"); + return request(provider, init); +}; +try { + const service = aiConnectionService(db); + const [row] = await service.quotaAccounts(process.env.CREDENTIAL_TEST_COMPANY_ID!, process.env.CREDENTIAL_TEST_OWNER!); + await service.refreshQuotaCredential(row, await service.credential(row), new AbortController().signal); + process.exitCode = 0; +} catch { process.exitCode = 1; } +finally { await db.$client.end(); } diff --git a/server/src/__tests__/plugin-task-credential-writeback.test.ts b/server/src/__tests__/plugin-task-credential-writeback.test.ts index 8ef383854c9..0603d09f59a 100644 --- a/server/src/__tests__/plugin-task-credential-writeback.test.ts +++ b/server/src/__tests__/plugin-task-credential-writeback.test.ts @@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto"; import { mkdtemp, realpath, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { and, eq } from "drizzle-orm"; +import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, companies, companyMemberships, createDb, environmentLeases, environments, heartbeatRuns, plugins } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -90,6 +90,26 @@ suite("durable plugin task credential handoff", () => { expect(await f.credential()).toBe(auth("fixture-refreshed", 2021)); expect(f.calls.at(-1)).toBe("complete"); }); + it("retains the Runner credential after a database save failure until a fresh receiver commits", async () => { + const f = await fixture(); + await db.execute(sql.raw(`create function reject_runner_credential_save() returns trigger language plpgsql as $$ begin + if new.secret_id = '${f.managed.credentialWriteback!.secretId}'::uuid then raise exception 'fixture receiver save unavailable'; end if; return new; end $$; + create constraint trigger reject_runner_credential_save after insert on company_secret_versions + deferrable initially deferred for each row execute function reject_runner_credential_save();`)); + try { + await expect(f.settle()).rejects.toThrow(); + expect(f.calls).toContain("credentials"); + expect(f.calls).not.toContain("complete"); + expect(await f.credential()).toBe(auth("fixture-original")); + expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "pending" }); + } finally { await db.execute(sql`drop trigger reject_runner_credential_save on company_secret_versions; drop function reject_runner_credential_save()`); } + const receiver = createDb(database.connectionString); + expect(await settleDispatchedEnvironmentTask(receiver, f.workers, { companyId: f.companyId, id: f.leaseId })).toBe(true); + expect(await f.credential()).toBe(auth("fixture-refreshed", 2021)); + expect(f.calls.filter(call => call === "credentials")).toHaveLength(2); + expect(f.calls.at(-1)).toBe("complete"); + expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "complete" }); + }); it("does not overwrite an explicit account replacement", async () => { const f = await fixture(); await secretService(db).rotate(f.managed.credentialWriteback!.secretId, { value: auth("fixture-reconnected", 2022) }, { userId: f.owner }); diff --git a/server/src/services/ai-connections.ts b/server/src/services/ai-connections.ts index c03b40e3ced..d76f9606c9e 100644 --- a/server/src/services/ai-connections.ts +++ b/server/src/services/ai-connections.ts @@ -1105,7 +1105,7 @@ export function aiConnectionService(db: Db) { async function refreshQuotaCredential(row: Parameters[0], failedValue: string, signal: AbortSignal) { return withAccountHomeSecretMutationLock(undefined, row.connection.companyId, async () => { const recovery = quotaCredentialRecovery(row.connection.companyId, row.grant.id); - let savingReplacement = false; + let replacement: { value: string; secretId: string; version: number; epoch: number; base: string; grantUpdatedAt: number; versionSelector: "latest" | number } | undefined; 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. @@ -1117,11 +1117,20 @@ export function aiConnectionService(db: Db) { 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( + const [secret] = await tx.select({ id: companySecrets.id, version: companySecrets.latestVersion, epoch: companySecrets.aiSessionEpoch }).from(companySecrets).where(and( eq(companySecrets.id, ref.secretId), eq(companySecrets.companyId, companyId), )).for("update", { noWait: true }); + if (!secret) throw new Error("credentials_unavailable"); const current = await aiConnectionService(tx as unknown as Db).credential({ connection: row.connection, grant }); - signal.throwIfAborted(); + if (replacement) { + // A lost commit response is reconciled from the authoritative secret. + // Reconnect, revocation and a competing saved generation always win. + if (current === replacement.value) return current; + if (ref.secretId !== replacement.secretId || secret.version !== replacement.version + || secret.epoch !== replacement.epoch || current !== replacement.base + || grant.updatedAt.getTime() !== replacement.grantUpdatedAt + || (ref.versionSelector ?? "latest") !== replacement.versionSelector) return current; + } else signal.throwIfAborted(); let pending = await recovery.read(); if (pending && (pending.secretId !== ref.secretId || pending.connectionId !== row.connection.id || pending.baseHash !== quotaCredentialHash(current))) { // Reconnect may reuse this grant. Once the current authorized token @@ -1141,10 +1150,9 @@ export function aiConnectionService(db: Db) { 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 (active && !pending && !replacement) throw new Error("provider_unavailable"); + let refreshed = replacement?.value ?? 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"); @@ -1169,12 +1177,12 @@ export function aiConnectionService(db: Db) { ...(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; + // Only this logical refresh owns the replacement. Save retries must not + // consume another single-use token. Process loss before commit can require + // signing in again in v1; no secondary credential vault is created. + replacement ??= { value: refreshed, secretId: ref.secretId, version: secret.version, epoch: secret.epoch, base: current, + grantUpdatedAt: grant.updatedAt.getTime(), versionSelector: ref.versionSelector ?? "latest" }; // 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, preserveAiSessionEpoch: true }, { userId: grant.subjectUserId }); @@ -1191,9 +1199,9 @@ export function aiConnectionService(db: Db) { 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. + if (!replacement || attempt >= 2) throw error; + // Retry only the returned value. Required database reads still fail + // closed; exhaustion is an error, never a successful refresh. } } }); diff --git a/server/src/services/quota-credential-recovery.ts b/server/src/services/quota-credential-recovery.ts index 2158c760e93..990673528be 100644 --- a/server/src/services/quota-credential-recovery.ts +++ b/server/src/services/quota-credential-recovery.ts @@ -28,7 +28,9 @@ export const quotaCredentialHash = (value: string) => createHash("sha256").updat export const quotaCredentialRecoveryPath = (companyId: string, grantId: string) => path.join(idleQuotaRecoveryPath(), companyId, `${grantId}.json`); -/** Called only while holding the company credential mutation lock. The local +/** Legacy settlement and local-only Google retirement. Server quota refresh + * must never create a new record here. 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) {