mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(runtime): persist managed Runner credential refreshes across execution targets (#20580)
This commit is contained in:
1 parent
85dec04677
commit
17136a6c25
23 files changed
+883
-238
No files matched your search
@@ -5208,3 +5208,33 @@ failure remains retryable even when the clone directory already exists. Later
|
||||
workspace saves retain repository coverage only for directories present in the
|
||||
captured archive. Deleted checkouts lose coverage and can be acquired again.
|
||||
A failed batch upload marks the affected repository rows as needing repair.
|
||||
|
||||
### Native Runner credential refreshes
|
||||
|
||||
Core handles PRP `auth.credentials.updated` notifications for managed Codex and
|
||||
Grok subscriptions across local, SSH, sandbox, and supported plugin Runner tasks.
|
||||
After validating the company, agent, run, session, and Runner binding, it exports
|
||||
the selected account and writes newer credentials directly to the tenant secret.
|
||||
Grant, secret, version-selector, session-epoch, and provider freshness checks are
|
||||
shared across execution targets. Explicit reconnects and revoked grants win over
|
||||
old refreshes. Providers that do not emit this event keep their existing flow;
|
||||
operator-owned homes do not acquire tenant writeback authority.
|
||||
|
||||
The transport's credential handler rebinds with each warm invocation, including
|
||||
its current target exporter. Idle refreshes and recovered controller handles do
|
||||
not depend on a completed invocation's temporary home. Detached handles cannot
|
||||
process new updates. Native/ACPX Codex and Grok use bounded, descriptor-based file
|
||||
reads; SSH and sandbox readers use the Node runtime already required by remote
|
||||
isolated-home preparation. Plugin tasks use their account-bound export operation.
|
||||
|
||||
A notification is acknowledged only after writeback succeeds or its authority is
|
||||
intentionally discarded. Export and database failures replay through PRP. An
|
||||
unchanged managed-subscription turn does no credential export or database write.
|
||||
Retirement keeps a final export for legacy and recovery paths; a failed final
|
||||
export retains the provider's credential files for retry. After verified provider
|
||||
exit, ACPX Codex and Grok preserve a private `auth-refresh.json` handoff before
|
||||
scrubbing `auth.json`; Core removes it after retirement writeback. Plugin retirement also
|
||||
records rotation, unchanged credential, changed binding, undispatched task, or a
|
||||
null export. No secret bytes appear in PRP events or lease metadata. The plugin
|
||||
provider must support live account-bound exports before enabling the paired Core
|
||||
handler.
|
||||
@@ -230,11 +230,15 @@ impl CredentialObserver {
|
||||
self.watch_directory();
|
||||
let state = self.state.as_ref().expect("configured observer");
|
||||
let snapshot = read_snapshot(&state.provider, &state.path).or_else(|| {
|
||||
(state.provider == "grok"
|
||||
&& fs::symlink_metadata(&state.path)
|
||||
.is_err_and(|error| error.kind() == std::io::ErrorKind::NotFound))
|
||||
.then(|| read_snapshot("grok", &state.path.with_file_name("auth-refresh.json")))
|
||||
.flatten()
|
||||
fs::symlink_metadata(&state.path)
|
||||
.is_err_and(|error| error.kind() == std::io::ErrorKind::NotFound)
|
||||
.then(|| {
|
||||
read_snapshot(
|
||||
&state.provider,
|
||||
&state.path.with_file_name("auth-refresh.json"),
|
||||
)
|
||||
})
|
||||
.flatten()
|
||||
});
|
||||
let Some(snapshot) = snapshot else {
|
||||
return Ok(());
|
||||
@@ -671,6 +675,21 @@ mod tests {
|
||||
assert_eq!(f.observer.pending(), Some(update_event("grok", 2)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn codex_close_handoff_preserves_shutdown_refresh() {
|
||||
let mut f = Fixture::new("codex", &codex("seed"));
|
||||
f.write("auth.json", &codex("seed"));
|
||||
f.observer.activate().unwrap();
|
||||
assert!(f.observer.pending().is_none());
|
||||
f.write("auth-refresh.json", &codex("close"));
|
||||
fs::remove_file(f.root.join("auth.json")).unwrap();
|
||||
f.observer.poll(true).unwrap();
|
||||
assert_eq!(f.observer.pending(), Some(update_event("codex", 1)));
|
||||
f.observer.acknowledge(1).unwrap();
|
||||
f.observer.poll(true).unwrap();
|
||||
assert!(f.observer.pending().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_symlinks_unsafe_permissions_and_oversized_stores() {
|
||||
let f = Fixture::new("codex", &codex("seed"));
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
import { constants } from "node:fs";
|
||||
import { open, rename, rm, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import type { AcpxProviderLifetimeLease } from "./codex-credentials.js";
|
||||
|
||||
/** Called only by owners that close their credential lease after provider exit. */
|
||||
export async function retainAcpxCredentialRefresh(input: {
|
||||
lease: AcpxProviderLifetimeLease;
|
||||
agentHomeDirectory: string;
|
||||
retainRefresh: () => boolean;
|
||||
beforeClose?: () => Promise<void>;
|
||||
}): Promise<AcpxProviderLifetimeLease> {
|
||||
const { lease } = input;
|
||||
const refresh = join(input.agentHomeDirectory, "auth-refresh.json");
|
||||
await rm(refresh, { force: true }).catch(async (error) => { await lease.close(); throw error; });
|
||||
let prepared = false;
|
||||
let closed = false;
|
||||
let closing: Promise<void> | null = null;
|
||||
return {
|
||||
lifetimeFenceCandidates: lease.lifetimeFenceCandidates,
|
||||
lifetimeFenceFds: lease.lifetimeFenceFds,
|
||||
activateLifetimeOwner: (pid) => lease.activateLifetimeOwner(pid),
|
||||
close() {
|
||||
if (closed) return Promise.resolve();
|
||||
if (closing) return closing;
|
||||
closing = (async () => {
|
||||
// The runtime host has verified provider exit. Keep the final snapshot
|
||||
// private until Core persists it, before auth.json is scrubbed.
|
||||
// Do not touch a successor's files if lease cleanup needs a retry.
|
||||
if (!prepared) {
|
||||
if (input.retainRefresh()) {
|
||||
const auth = await open(join(input.agentHomeDirectory, "auth.json"), constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK).catch((error) => {
|
||||
if (error.code === "ENOENT") return null;
|
||||
throw error;
|
||||
});
|
||||
if (auth) {
|
||||
try {
|
||||
const stat = await auth.stat();
|
||||
if (!stat.isFile() || stat.uid !== process.getuid?.() || (stat.mode & 0o777) !== 0o600 || stat.size > 256 * 1024) throw new Error("ACPX refreshed credential is invalid");
|
||||
const buffer = Buffer.alloc(256 * 1024 + 1);
|
||||
let size = 0;
|
||||
while (size < buffer.length) {
|
||||
const read = await auth.read(buffer, size, buffer.length - size, size);
|
||||
if (!read.bytesRead) break;
|
||||
size += read.bytesRead;
|
||||
}
|
||||
if (size > 256 * 1024) { buffer.fill(0); throw new Error("ACPX refreshed credential exceeds its bound"); }
|
||||
const bytes = buffer.subarray(0, size);
|
||||
const temporary = `${refresh}.tmp`;
|
||||
try {
|
||||
await rm(temporary, { force: true });
|
||||
await writeFile(temporary, bytes, { flag: "wx", mode: 0o600 });
|
||||
await rename(temporary, refresh);
|
||||
} finally { bytes.fill(0); await rm(temporary, { force: true }); }
|
||||
} finally { await auth.close(); }
|
||||
}
|
||||
}
|
||||
await input.beforeClose?.();
|
||||
prepared = true;
|
||||
}
|
||||
await lease.close();
|
||||
closed = true;
|
||||
})().finally(() => { closing = null; });
|
||||
return closing;
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -1,7 +1,8 @@
|
||||
import { mkdir, mkdtemp, readFile, rm, stat, symlink, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { retainAcpxCredentialRefresh } from "./credential-refresh.js";
|
||||
import { stageManagedGrokCredential } from "./grok-credentials.js";
|
||||
import { createSanitizedAcpxSpawnInput } from "./environment.js";
|
||||
import { resolveQualifiedAcpxProfile } from "./qualified-profiles.js";
|
||||
@@ -75,6 +76,19 @@ describe("Grok isolated credentials", () => {
|
||||
await writeFile(join(directory, "auth.json"), credential, { mode: 0o600 });
|
||||
await lease.close();
|
||||
});
|
||||
it("retries a scrub failure without overwriting the completed handoff with a successor", async () => {
|
||||
const directory = await home();
|
||||
await writeFile(join(directory, "auth.json"), credential, { mode: 0o600 });
|
||||
const close = vi.fn().mockRejectedValueOnce(new Error("scrub failed")).mockResolvedValue(undefined);
|
||||
const lease = await retainAcpxCredentialRefresh({ agentHomeDirectory: directory, retainRefresh: () => true,
|
||||
lease: { lifetimeFenceCandidates: [49152, 49153, 49154], lifetimeFenceFds: [1, 2], activateLifetimeOwner: async () => {}, close } });
|
||||
await expect(lease.close()).rejects.toThrow("scrub failed");
|
||||
const successor = credential.replace("test-token", "successor-token");
|
||||
await writeFile(join(directory, "auth.json"), successor, { mode: 0o600 });
|
||||
await lease.close();
|
||||
expect(await readFile(join(directory, "auth-refresh.json"), "utf8")).toBe(credential);
|
||||
expect(await readFile(join(directory, "auth.json"), "utf8")).toBe(successor);
|
||||
});
|
||||
it("never passes other providers, host homes or the inline credential to Grok", () => {
|
||||
const result = createSanitizedAcpxSpawnInput({ XAI_API_KEY: "selected", HOME: "/host", GROK_HOME: "/host/.grok", OPENAI_API_KEY: "other", PAPERCLIP_ACPX_GROK_AUTH_JSON_SECRET: credential }, "grok");
|
||||
expect(result.env).toEqual({ XAI_API_KEY: "selected" });
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
import { constants } from "node:fs";
|
||||
import { open, rename, rm, writeFile } from "node:fs/promises";
|
||||
import { rm } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { stageManagedCodexCredential, type AcpxProviderLifetimeLease } from "./codex-credentials.js";
|
||||
|
||||
import { retainAcpxCredentialRefresh } from "./credential-refresh.js";
|
||||
|
||||
export const GROK_AUTH_REFRESH_FILE = "auth-refresh.json";
|
||||
|
||||
/** Shares the established credential staging, crash cleanup and process fence.
|
||||
@@ -37,56 +38,10 @@ export async function stageManagedGrokCredential(input: {
|
||||
agentHomeDirectory: input.agentHomeDirectory,
|
||||
environment: inline ? { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: inline } : { OPENAI_API_KEY: apiKey },
|
||||
});
|
||||
const refresh = join(input.agentHomeDirectory, GROK_AUTH_REFRESH_FILE);
|
||||
await rm(refresh, { force: true }).catch(async (error) => { await lease.close(); throw error; });
|
||||
let closed = false;
|
||||
let closing: Promise<void> | null = null;
|
||||
return {
|
||||
lifetimeFenceCandidates: lease.lifetimeFenceCandidates,
|
||||
lifetimeFenceFds: lease.lifetimeFenceFds,
|
||||
activateLifetimeOwner: (pid) => lease.activateLifetimeOwner(pid),
|
||||
close() {
|
||||
if (closed) return Promise.resolve();
|
||||
if (closing) return closing;
|
||||
closing = (async () => {
|
||||
// The runtime host calls close only after verified provider exit. Retain
|
||||
// a private refresh handoff for the controller's existing Grok merge
|
||||
// predicate, then scrub auth.json before releasing the process fence.
|
||||
if (inline && (input.retainRefresh?.() ?? true)) {
|
||||
const auth = await open(join(input.agentHomeDirectory, "auth.json"), constants.O_RDONLY | constants.O_NOFOLLOW).catch((error) => {
|
||||
if (error.code === "ENOENT") return null;
|
||||
throw error;
|
||||
});
|
||||
if (auth) {
|
||||
try {
|
||||
const stat = await auth.stat();
|
||||
if (!stat.isFile() || stat.uid !== process.getuid?.() || (stat.mode & 0o777) !== 0o600 || stat.size > 256 * 1024) throw new Error("Grok refreshed credential is invalid");
|
||||
const buffer = Buffer.alloc(256 * 1024 + 1);
|
||||
let size = 0;
|
||||
while (size < buffer.length) {
|
||||
const read = await auth.read(buffer, size, buffer.length - size, size);
|
||||
if (!read.bytesRead) break;
|
||||
size += read.bytesRead;
|
||||
}
|
||||
if (size > 256 * 1024) { buffer.fill(0); throw new Error("Grok refreshed credential exceeds its bound"); }
|
||||
const bytes = buffer.subarray(0, size);
|
||||
const temporary = `${refresh}.tmp`;
|
||||
try {
|
||||
await rm(temporary, { force: true });
|
||||
await writeFile(temporary, bytes, { flag: "wx", mode: 0o600 });
|
||||
await rename(temporary, refresh);
|
||||
} finally { bytes.fill(0); await rm(temporary, { force: true }); }
|
||||
} finally { await auth.close(); }
|
||||
}
|
||||
}
|
||||
// Grok diagnostics can contain account/authentication fields. They are
|
||||
// disposable, unlike session history, and must not outlive this owner.
|
||||
// close is invoked only after the provider has been contained.
|
||||
await rm(join(input.agentHomeDirectory, "logs"), { recursive: true, force: true });
|
||||
await lease.close();
|
||||
closed = true;
|
||||
})().finally(() => { closing = null; });
|
||||
return closing;
|
||||
},
|
||||
};
|
||||
return retainAcpxCredentialRefresh({
|
||||
lease, agentHomeDirectory: input.agentHomeDirectory,
|
||||
retainRefresh: () => Boolean(inline) && (input.retainRefresh?.() ?? true),
|
||||
// Diagnostics may contain auth fields; session history remains resumable.
|
||||
beforeClose: () => rm(join(input.agentHomeDirectory, "logs"), { recursive: true, force: true }),
|
||||
});
|
||||
}
|
||||
@@ -1413,6 +1413,32 @@ describe("ACPX runtime host", () => {
|
||||
expect(fixture.commandClose).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it.each(["subscription", "api_key"])("retains only the final subscription snapshot after Codex exits: %s", async mode => {
|
||||
const fixture = await hostFixture();
|
||||
let home = "";
|
||||
const runtime = runtimePort({ onClose: async () => {
|
||||
// A refresh during provider shutdown must survive the lease's auth scrub.
|
||||
await writeFile(join(home, "auth.json"), '{"tokens":{"access_token":"final-refresh"}}', { mode: 0o600 });
|
||||
} });
|
||||
const host = await AcpxRuntimeHost.open({
|
||||
...fixture.options, agent: "codex", model: "gpt-5.6-sol", permissionMode: "approve-all",
|
||||
environment: mode === "subscription"
|
||||
? { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"tokens":{"access_token":"initial"}}' }
|
||||
: { OPENAI_API_KEY: "synthetic-key" },
|
||||
}, fixture.dependencies({ openRuntime: async options => {
|
||||
home = options.launchEnvironment.CODEX_HOME!;
|
||||
return runtime;
|
||||
} }));
|
||||
await host.close({ reason: "final snapshot verified" });
|
||||
await expect(readFile(join(home, "auth.json"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
if (mode === "subscription") {
|
||||
expect(await readFile(join(home, "auth-refresh.json"), "utf8")).toContain("final-refresh");
|
||||
expect((await stat(join(home, "auth-refresh.json"))).mode & 0o777).toBe(0o600);
|
||||
} else {
|
||||
await expect(readFile(join(home, "auth-refresh.json"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
}
|
||||
});
|
||||
|
||||
it("cleans credentials and command leases when provider open fails", async () => {
|
||||
const fixture = await hostFixture();
|
||||
let authPath = "";
|
||||
@@ -1437,6 +1463,7 @@ describe("ACPX runtime host", () => {
|
||||
),
|
||||
).rejects.toThrow("provider failed");
|
||||
await expect(readFile(authPath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
await expect(readFile(join(authPath, "..", "auth-refresh.json"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(fixture.commandClose).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { retainAcpxCredentialRefresh } from "./credential-refresh.js";
|
||||
import { assertAcpxExecutionAdmission, parseAcpxCandidateStartupAdmission, type AcpxCandidateStartupAdmission } from "./candidate-admission.js";
|
||||
import { resolveGrokDisableWebSearch } from "./grok-web-search.js";
|
||||
import { recordAcpxAdmissionFailure, type ACPX_ADMISSION_CODES } from "./admission-diagnostics.js";
|
||||
@@ -505,12 +506,17 @@ export class AcpxRuntimeHost {
|
||||
if (options.agent === "codex" && !preconfigured) {
|
||||
credential = await acquireAbortableAdmissionResource({
|
||||
signal: options.signal,
|
||||
acquire: () =>
|
||||
(dependencies.stageCredential ?? stageManagedCodexCredential)({
|
||||
acquire: async () => {
|
||||
const lease = await (dependencies.stageCredential ?? stageManagedCodexCredential)({
|
||||
agentHomeDirectory: sandbox.agentHomeDirectory,
|
||||
environment: options.environment,
|
||||
sourcePath: options.managedCodexCredentialSourcePath,
|
||||
}),
|
||||
});
|
||||
return retainAcpxCredentialRefresh({
|
||||
lease, agentHomeDirectory: sandbox.agentHomeDirectory,
|
||||
retainRefresh: () => admissionSucceeded && lease.mode !== "api_key",
|
||||
});
|
||||
},
|
||||
resource: "credential",
|
||||
releaseLate: (lateCredential) => lateCredential.close(),
|
||||
reportFailure: (failure) =>
|
||||
|
||||
@@ -8,9 +8,12 @@ import { agents, companies, companyMemberships, createDb, environmentLeases, env
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { aiConnectionService } from "../services/ai-connections.js";
|
||||
import { prepareManagedAiRuntime } from "../services/ai-connection-runtime.js";
|
||||
import { registerPluginTaskCredentialHandoff, settlePluginTaskCredentialHandoff } from "../services/plugin-task-credential-writeback.js";
|
||||
import { registerPluginTaskCredentialHandoff, settlePluginTaskCredentialHandoff, syncPluginTaskCredentialHandoff } from "../services/plugin-task-credential-writeback.js";
|
||||
import { executeEnvironmentTask, settleDispatchedEnvironmentTask } from "../services/environment-task-runtime.js";
|
||||
import { secretService } from "../services/secrets.js";
|
||||
import { syncManagedAiCredentials } from "../services/managed-ai-credential-writeback.js";
|
||||
import { appendNativeCredentialUpdate } from "../services/native-runtime/native-credential-updates.js";
|
||||
import type { PrpEvent } from "../vendor/paperclip-runner/index.js";
|
||||
import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
|
||||
|
||||
vi.mock("../services/subscription-refresh.js", () => ({ refreshSubscriptionConnection: async () => {} }));
|
||||
@@ -26,17 +29,24 @@ suite("durable plugin task credential handoff", () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("plugin-task-credential-"); db = createDb(database.connectionString);
|
||||
}, 90_000);
|
||||
afterAll(async () => { await database?.cleanup(); vi.unstubAllEnvs(); if (home) await rm(home, { recursive: true, force: true }); });
|
||||
async function fixture(dispatched = true) {
|
||||
async function fixture(dispatched = true, provider: "codex" | "grok" = "codex") {
|
||||
const selectedProvider = provider === "codex" ? "openai" as const : "xai" as const;
|
||||
const credential = (token: string, year = 2020) => provider === "codex" ? auth(token, year) : JSON.stringify({
|
||||
"https://issuer.example.test::11111111-1111-4111-8111-111111111111": {
|
||||
key: "fixture-access", refresh_token: token, expires_at: `${year}-01-01T00:00:00Z`,
|
||||
},
|
||||
});
|
||||
const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), pluginId = randomUUID(), environmentId = randomUUID(), leaseId = randomUUID();
|
||||
const owner = `fixture-${randomUUID()}`;
|
||||
await db.insert(companies).values({ id: companyId, name: "Task credential fixture", issuePrefix: `C${companyId.slice(0, 6)}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", adapterType: "paperclip_runner" });
|
||||
await db.insert(companyMemberships).values({ companyId, principalId: owner, principalType: "user", status: "active", membershipRole: "member" });
|
||||
const service = aiConnectionService(db);
|
||||
await service.save(companyId, owner, { provider: "openai", method: "subscription", ownership: "personal", name: "Fixture Codex", loginSessionId: "fixture", agentIds: [], allAgents: true }, auth("fixture-original"));
|
||||
const [selection] = await service.quotaAccounts(companyId, owner);
|
||||
await service.save(companyId, owner, { provider: selectedProvider, method: "subscription", ownership: "personal", name: "Fixture account", loginSessionId: "fixture", agentIds: [], allAgents: true }, credential("fixture-original"));
|
||||
const [selection] = await service.quotaAccounts(companyId, owner, true);
|
||||
const managed = await prepareManagedAiRuntime(db, { companyId, agentId, responsibleUserId: owner, adapterType: "paperclip_runner",
|
||||
binding: { mode: "responsible_user", provider: "openai", method: "subscription" }, config: { provider: "codex" } });
|
||||
binding: { mode: "responsible_user", provider: selectedProvider, method: "subscription" },
|
||||
config: provider === "codex" ? { provider: "codex" } : { provider: "acpx", acpxAgent: "grok" } });
|
||||
expect(managed.credentialWriteback).toBeDefined();
|
||||
await db.insert(environments).values({ id: environmentId, name: `Fixture ${environmentId}`, driver: "plugin", status: "active", config: { pluginKey: pluginId, driverKey: "tasks", driverConfig: {} } });
|
||||
await db.insert(plugins).values({ id: pluginId, pluginKey: pluginId, packageName: "fixture-tasks", version: "1.0.0", status: "ready", manifestJson: {
|
||||
@@ -49,7 +59,7 @@ suite("durable plugin task credential handoff", () => {
|
||||
// Recovery must not depend on the original controller or its private home.
|
||||
await rm(managed.home!, { recursive: true, force: true });
|
||||
const calls: string[] = [];
|
||||
let failExport = false;
|
||||
let failExport = false, exported: string | null = auth("fixture-refreshed", 2021);
|
||||
const workers = { getWorker: () => ({ supportedMethods: ["environmentTask"] }),
|
||||
call: async (_plugin: string, _method: string, params: any) => {
|
||||
calls.push(params.operation.kind);
|
||||
@@ -58,15 +68,126 @@ suite("durable plugin task credential handoff", () => {
|
||||
if (params.operation.kind === "credentials") {
|
||||
expect(params.operation.accountIdentity).toBe(managed.sessionIdentity);
|
||||
if (failExport) throw new Error("fixture export unavailable");
|
||||
return { kind: "credentials", taskId: "fixture-task", credential: { type: "codex-auth-json", authJson: auth("fixture-refreshed", 2021) } };
|
||||
return { kind: "credentials", taskId: "fixture-task", credential: exported === null ? null : { type: "codex-auth-json", authJson: exported } };
|
||||
}
|
||||
return { kind: "accepted", taskId: "fixture-task" };
|
||||
} } as unknown as PluginWorkerManager;
|
||||
const lease = () => db.select().from(environmentLeases).where(and(eq(environmentLeases.id, leaseId), eq(environmentLeases.companyId, companyId))).then(rows => rows[0]!);
|
||||
return { companyId, agentId, runId, environmentId, leaseId, owner, selection: selection!, managed, workers, calls, lease,
|
||||
return { accountCredential: credential, companyId, agentId, runId, environmentId, leaseId, owner, selection: selection!, managed, workers, calls, lease,
|
||||
synchronize: () => syncPluginTaskCredentialHandoff(db, { companyId, leaseId, expectedTaskId: "fixture-task" }, async accountIdentity => {
|
||||
const result = await executeEnvironmentTask(db, workers, { companyId, leaseId, expectedTaskId: "fixture-task", operation: { kind: "credentials", accountIdentity } });
|
||||
if (result.kind !== "credentials") throw new Error("Unexpected credential result");
|
||||
return result.credential;
|
||||
}),
|
||||
exportCredential(value: string | null) { exported = value; },
|
||||
failExport(value: boolean) { failExport = value; }, credential: () => service.credential(selection!),
|
||||
settle: (executionStopObserved = false) => settleDispatchedEnvironmentTask(db, workers, { companyId, id: leaseId }, { executionStopObserved }) };
|
||||
}
|
||||
it.each(["codex", "grok"] as const)("writes managed %s refreshes without any plugin or invocation home", async provider => {
|
||||
const f = await fixture(true, provider);
|
||||
const read = vi.fn(async () => f.accountCredential("fixture-refreshed", 2021));
|
||||
expect(await syncManagedAiCredentials(db, f.companyId, f.managed.credentialWriteback!, provider, read)).toBe("rotated");
|
||||
expect(await f.credential()).toBe(f.accountCredential("fixture-refreshed", 2021));
|
||||
expect(await syncManagedAiCredentials(db, f.companyId, f.managed.credentialWriteback!, provider, read)).toBe("unchanged");
|
||||
await secretService(db).rotate(f.managed.credentialWriteback!.secretId, { value: f.accountCredential("fixture-reconnected", 2022) }, { userId: f.owner });
|
||||
read.mockClear();
|
||||
expect(await syncManagedAiCredentials(db, f.companyId, f.managed.credentialWriteback!, provider, read)).toBe("binding_changed");
|
||||
expect(read).not.toHaveBeenCalled();
|
||||
expect(await f.credential()).toBe(f.accountCredential("fixture-reconnected", 2022));
|
||||
expect(f.calls).toEqual([]);
|
||||
});
|
||||
it.each((["codex", "grok"] as const).flatMap(provider => ["{partial", "{}"].map(snapshot => [provider, snapshot] as const)))(
|
||||
"rejects unusable %s snapshots for retry: %s", async (provider, snapshot) => {
|
||||
const f = await fixture(true, provider);
|
||||
await expect(syncManagedAiCredentials(db, f.companyId, f.managed.credentialWriteback!, provider, async () => snapshot))
|
||||
.rejects.toThrow("snapshot invalid");
|
||||
expect(await f.credential()).toBe(f.accountCredential("fixture-original"));
|
||||
expect(await syncManagedAiCredentials(db, f.companyId, f.managed.credentialWriteback!, provider,
|
||||
async () => f.accountCredential("fixture-refreshed", 2021))).toBe("rotated");
|
||||
},
|
||||
);
|
||||
it.each(["codex", "grok"] as const)("rechecks %s authority after the exporter yields", async provider => {
|
||||
const f = await fixture(true, provider);
|
||||
const result = await syncManagedAiCredentials(db, f.companyId, f.managed.credentialWriteback!, provider, async () => {
|
||||
await secretService(db).rotate(f.managed.credentialWriteback!.secretId, { value: f.accountCredential("fixture-reconnected", 2022) }, { userId: f.owner });
|
||||
return f.accountCredential("fixture-refreshed", 2021);
|
||||
});
|
||||
expect(result).toBe("binding_changed");
|
||||
expect(await f.credential()).toBe(f.accountCredential("fixture-reconnected", 2022));
|
||||
});
|
||||
it("replays a validated PRP refresh into the tenant secret after export failure", async () => {
|
||||
const f = await fixture(), normalizedSessionId = randomUUID(), runnerInstanceId = randomUUID();
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", runtimeMode: "native", nativeSessionId: normalizedSessionId, runnerInstanceId })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
await db.update(environmentLeases).set({ status: "retained" }).where(eq(environmentLeases.id, f.leaseId));
|
||||
const event: PrpEvent = {
|
||||
schema: "paperclip.prp.event.v3", schemaVersion: 3, eventType: "auth.credentials.updated", priority: 0,
|
||||
sourceKind: "runner", sourceInstanceId: runnerInstanceId, sourceEventId: "live-refresh", sourceSeq: 1,
|
||||
runId: f.runId, normalizedSessionId, emittedAt: "2020-01-01T00:00:00Z",
|
||||
payload: { schema: "paperclip.auth.credentials.updated.v1", provider: "codex", bindingId: "harness-auth", revision: 1, detection: "credential_store" },
|
||||
};
|
||||
const notification = { companyId: f.companyId, agentId: f.agentId, runnerInstanceId, normalizedSessionId,
|
||||
event, synchronizeCredentials: f.synchronize };
|
||||
f.failExport(true);
|
||||
await expect(appendNativeCredentialUpdate(db, notification)).rejects.toThrow();
|
||||
expect(await f.credential()).toBe(auth("fixture-original"));
|
||||
f.failExport(false);
|
||||
await appendNativeCredentialUpdate(db, notification);
|
||||
expect(await f.credential()).toBe(auth("fixture-refreshed", 2021));
|
||||
expect(f.calls).toEqual(["credentials", "credentials"]);
|
||||
expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "pending" });
|
||||
});
|
||||
it("writes repeated live refreshes without stopping or completing the warm task", async () => {
|
||||
const f = await fixture();
|
||||
expect(await f.synchronize()).toBe("rotated");
|
||||
expect(await f.credential()).toBe(auth("fixture-refreshed", 2021));
|
||||
expect(await f.synchronize()).toBe("unchanged");
|
||||
f.exportCredential(auth("fixture-next-refresh", 2022));
|
||||
expect(await f.synchronize()).toBe("rotated");
|
||||
expect(await f.credential()).toBe(auth("fixture-next-refresh", 2022));
|
||||
expect(f.calls).toEqual(["credentials", "credentials", "credentials"]);
|
||||
expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "pending" });
|
||||
expect(await f.settle()).toBe("completed");
|
||||
expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "complete", outcome: "unchanged" });
|
||||
});
|
||||
it("keeps live export failures replayable and rejects missing snapshots", async () => {
|
||||
const f = await fixture(); f.failExport(true);
|
||||
await expect(f.synchronize()).rejects.toThrow();
|
||||
f.failExport(false); f.exportCredential(null);
|
||||
await expect(f.synchronize()).rejects.toThrow("snapshot unavailable");
|
||||
expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "pending" });
|
||||
expect(await f.credential()).toBe(auth("fixture-original"));
|
||||
f.exportCredential(auth("fixture-refreshed", 2021));
|
||||
expect(await f.synchronize()).toBe("rotated");
|
||||
expect(f.calls).not.toContain("complete");
|
||||
});
|
||||
it("fences live writeback after explicit reconnect and physical task replacement", async () => {
|
||||
const f = await fixture();
|
||||
await secretService(db).rotate(f.managed.credentialWriteback!.secretId, { value: auth("fixture-reconnected", 2022) }, { userId: f.owner });
|
||||
expect(await f.synchronize()).toBe("binding_changed");
|
||||
expect(f.calls).toEqual([]);
|
||||
expect(await f.credential()).toBe(auth("fixture-reconnected", 2022));
|
||||
await db.update(environmentLeases).set({ providerLeaseId: "replacement-task" }).where(eq(environmentLeases.id, f.leaseId));
|
||||
await expect(f.synchronize()).rejects.toThrow("lease unavailable");
|
||||
expect(f.calls).toEqual([]);
|
||||
});
|
||||
it("records a null retirement export without treating it as a secret rotation", async () => {
|
||||
const f = await fixture(); f.exportCredential(null);
|
||||
expect(await f.settle()).toBe("completed");
|
||||
expect(f.calls).toContain("credentials");
|
||||
expect(await f.credential()).toBe(auth("fixture-original"));
|
||||
expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "complete", outcome: "no_credentials" });
|
||||
});
|
||||
it("exports a dispatched stopped task even after the dispatch flag was cleared", async () => {
|
||||
const f = await fixture();
|
||||
const metadata = { ...(await f.lease()).metadata, runnerTaskDispatched: false, runnerTaskExecutionStopped: true,
|
||||
runnerTaskIdentity: { runnerId: "fixture-runner" } };
|
||||
await db.update(environmentLeases).set({ metadata }).where(eq(environmentLeases.id, f.leaseId));
|
||||
expect(await f.settle()).toBe("stopped");
|
||||
expect(f.calls).toEqual(["credentials"]);
|
||||
expect(await f.credential()).toBe(auth("fixture-refreshed", 2021));
|
||||
expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "complete", outcome: "rotated" });
|
||||
});
|
||||
it("persists refreshed credentials before completing a stopped task, after controller files disappear", async () => {
|
||||
const f = await fixture();
|
||||
expect(await f.settle()).toBe("completed");
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { environmentTaskResultSchema } from "@paperclipai/plugin-sdk";
|
||||
import type { PluginTaskCredentialWritebackBinding } from "./plugin-task-credential-writeback.js";
|
||||
import type { ManagedAiCredentialWritebackBinding } from "./managed-ai-credential-writeback.js";
|
||||
import { OPENROUTER_ATTRIBUTION_HEADERS, openRouterClaudeHeaders } from "@paperclipai/shared/openrouter-attribution";
|
||||
import { createHash } from "node:crypto";
|
||||
import { HttpError, unprocessable } from "../errors.js";
|
||||
@@ -606,9 +606,9 @@ export async function prepareManagedAiRuntime(
|
||||
return {
|
||||
sessionIdentity,
|
||||
...(retainedCleanupScope ? { retainedCleanupScope } : {}),
|
||||
...(subscriptionFile && input.binding.provider === "openai" && credentialRef && freshness ? {
|
||||
...(subscriptionFile && (input.binding.provider === "openai" || input.binding.provider === "xai") && credentialRef && freshness ? {
|
||||
credentialWriteback: { grantId: selection.grant.id, secretId: credentialRef.secretId, sessionEpoch: freshness.epoch,
|
||||
versionSelector: credentialRef.versionSelector ?? "latest" } satisfies PluginTaskCredentialWritebackBinding,
|
||||
versionSelector: credentialRef.versionSelector ?? "latest" } satisfies ManagedAiCredentialWritebackBinding,
|
||||
} : {}),
|
||||
config: {
|
||||
...input.config,
|
||||
|
||||
@@ -498,12 +498,18 @@ export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWor
|
||||
const [current] = await db.update(environmentLeases).set({
|
||||
metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"runnerTaskRetiring":true}'::jsonb`,
|
||||
}).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId)))
|
||||
.returning({ metadata: environmentLeases.metadata, heartbeatRunId: environmentLeases.heartbeatRunId });
|
||||
.returning({ metadata: environmentLeases.metadata, heartbeatRunId: environmentLeases.heartbeatRunId, taskId: environmentLeases.providerLeaseId });
|
||||
const startup = current?.metadata?.runnerTaskStartup;
|
||||
const startupCleanup = current?.metadata?.runnerTaskStartupCleanup as { requiresPreparationStop?: boolean } | undefined;
|
||||
if (current?.metadata?.runnerTaskDispatched !== true && !startup && !startupCleanup) {
|
||||
if (current?.metadata) await settlePluginTaskCredentialHandoff(db, { companyId: lease.companyId, leaseId: lease.id }, current.metadata,
|
||||
async () => { throw new Error("Undispatched task must not request credentials"); });
|
||||
if (current?.metadata) await settlePluginTaskCredentialHandoff(db, {
|
||||
companyId: lease.companyId, leaseId: lease.id, expectedTaskId: current.taskId ?? undefined,
|
||||
}, current.metadata, async accountIdentity => {
|
||||
const snapshot = await executeEnvironmentTask(db, workers, { companyId: lease.companyId, leaseId: lease.id,
|
||||
expectedTaskId: current.taskId ?? undefined, operation: { kind: "credentials", accountIdentity } });
|
||||
if (snapshot.kind !== "credentials") throw new Error("Plugin Runner credential snapshot unavailable");
|
||||
return snapshot.credential;
|
||||
});
|
||||
return current?.metadata?.runnerTaskExecutionStopped === true ? "stopped" : false;
|
||||
}
|
||||
const scope = { companyId: lease.companyId, leaseId: lease.id };
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
import { z } from "zod";
|
||||
import { aiConnectionMetadataSchema } from "@paperclipai/shared";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { companySecrets, connectionGrants, toolConnections, type Db } from "@paperclipai/db";
|
||||
import { mkdtemp, writeFile, rm, realpath } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { decideCodexAuthMerge, readSubscriptionAccountId, withAccountHomeSecretMutationLock } from "@paperclipai/adapter-codex-local/server";
|
||||
import { decideGrokAuthMerge, parseGrokAuthPayload, hasUsableGrokAuthValue } from "@paperclipai/adapter-grok-local/server";
|
||||
import { WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE } from "@paperclipai/adapter-utils/workspace-restore-merge";
|
||||
import { aiConnectionService } from "./ai-connections.js";
|
||||
import { secretService } from "./secrets.js";
|
||||
|
||||
export const managedAiCredentialWritebackBindingSchema = z.object({
|
||||
grantId: z.string().uuid(), secretId: z.string().uuid(), sessionEpoch: z.number().int().nonnegative(),
|
||||
versionSelector: z.union([z.literal("latest"), z.number().int().positive()]),
|
||||
}).strict();
|
||||
export type ManagedAiCredentialWritebackBinding = z.infer<typeof managedAiCredentialWritebackBindingSchema>;
|
||||
|
||||
/** Credential ownership survives the invocation home; reconnects and revocations win. */
|
||||
export async function syncManagedAiCredentials(
|
||||
db: Db,
|
||||
companyId: string,
|
||||
binding: ManagedAiCredentialWritebackBinding,
|
||||
provider: "codex" | "grok",
|
||||
fetchCredential: () => Promise<string | null>,
|
||||
) {
|
||||
managedAiCredentialWritebackBindingSchema.parse(binding);
|
||||
const home = await mkdtemp(join(await realpath(tmpdir()), "paperclip-ai-credential-"));
|
||||
try {
|
||||
// Revocation or an explicit account replacement intentionally discards an old refresh.
|
||||
const account = async (database: Db) => {
|
||||
const [grant] = await database.select().from(connectionGrants).where(and(
|
||||
eq(connectionGrants.companyId, companyId), eq(connectionGrants.id, binding.grantId),
|
||||
)).for("update");
|
||||
if (!grant || grant.status !== "active") return null;
|
||||
const ref = grant.credentialSecretRefs.find(ref => ref.configPath === "ai.credential");
|
||||
if (!ref || ref.secretId !== binding.secretId ||
|
||||
(ref.versionSelector ?? "latest") !== binding.versionSelector) return null;
|
||||
const [secret] = await database.select().from(companySecrets).where(and(
|
||||
eq(companySecrets.companyId, companyId), eq(companySecrets.id, binding.secretId),
|
||||
)).for("update");
|
||||
if (!secret || secret.aiSessionEpoch !== binding.sessionEpoch) return null;
|
||||
const [connection] = await database.select().from(toolConnections).where(and(
|
||||
eq(toolConnections.companyId, companyId), eq(toolConnections.id, grant.connectionId),
|
||||
));
|
||||
const ai = aiConnectionMetadataSchema.safeParse(connection?.config.ai);
|
||||
if (!connection || !ai.success || ai.data.provider !== (provider === "codex" ? "openai" : "xai") || ai.data.method !== "subscription") return null;
|
||||
return { connection, grant };
|
||||
};
|
||||
const mutate = <T>(operation: (database: Db) => Promise<T>) => withAccountHomeSecretMutationLock(
|
||||
undefined, companyId, () => db.transaction(tx => operation(tx as unknown as Db)),
|
||||
);
|
||||
const current = await mutate(account);
|
||||
if (!current) return "binding_changed" as const;
|
||||
// Do not hold database or account locks while the provider exports its credential.
|
||||
const credential = await fetchCredential();
|
||||
if (credential === null) return "no_credentials" as const;
|
||||
// A partial provider write must replay, not be acknowledged as unchanged.
|
||||
let usable = false;
|
||||
if (provider === "codex") usable = readSubscriptionAccountId(Buffer.from(credential)) !== null;
|
||||
else {
|
||||
try {
|
||||
const parsed = parseGrokAuthPayload(JSON.parse(credential));
|
||||
usable = parsed !== null && hasUsableGrokAuthValue(parsed.value);
|
||||
} catch { /* Do not expose credential bytes through JSON parser errors. */ }
|
||||
}
|
||||
if (!usable) throw new Error("Managed credential snapshot invalid");
|
||||
const incoming = join(home, "incoming.json"), destination = join(home, "current.json");
|
||||
await writeFile(incoming, credential, { mode: 0o600 });
|
||||
const writeBack = () => mutate(async database => {
|
||||
const selected = await account(database);
|
||||
if (!selected) return "binding_changed" as const;
|
||||
const value = await aiConnectionService(database).credential(selected);
|
||||
await writeFile(destination, value, { mode: 0o600 });
|
||||
if (await (provider === "codex" ? decideCodexAuthMerge : decideGrokAuthMerge)(incoming, destination, { errorLabel: "AI account refresh" }) !== 10) return "unchanged" as const;
|
||||
await secretService(database).rotate(binding.secretId, {
|
||||
value: credential, preserveAiSessionEpoch: true,
|
||||
}, { userId: selected.grant.subjectUserId });
|
||||
return "rotated" as const;
|
||||
});
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
try {
|
||||
return await writeBack();
|
||||
} catch (error) {
|
||||
if (attempt >= 2 || (error as NodeJS.ErrnoException)?.code !== WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE) throw error;
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await rm(home, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
@@ -109,3 +109,18 @@ describe("managed native credential turns", () => {
|
||||
expect(remove).toHaveBeenCalledOnce();
|
||||
});
|
||||
});
|
||||
|
||||
it("defers the durable final export until retirement and retains it for retry", async () => {
|
||||
const session = { close: vi.fn(async () => {}) } as unknown as NativeSession;
|
||||
const copyBack = vi.fn(async () => {}), remove = vi.fn(async () => {});
|
||||
const retire = vi.fn().mockRejectedValueOnce(new Error("export unavailable")).mockResolvedValue(undefined);
|
||||
bindManagedNativeCredentialTurn(session, { copyBack, retire, remove });
|
||||
await completeManagedNativeCredentialTurn(session);
|
||||
expect(retire).not.toHaveBeenCalled();
|
||||
await expect(session.close({ reason: "idle" })).rejects.toThrow("export unavailable");
|
||||
expect(remove).not.toHaveBeenCalled();
|
||||
await session.close({ reason: "retry retirement" });
|
||||
await session.close({ reason: "already retired" });
|
||||
expect(retire).toHaveBeenCalledTimes(2);
|
||||
expect(remove).toHaveBeenCalledOnce();
|
||||
});
|
||||
@@ -1,10 +1,13 @@
|
||||
import type { NativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
|
||||
type CredentialTurn = {
|
||||
/** Copy refreshed credentials only into this invocation's private home. */
|
||||
/** Legacy turn copy-back; event-backed bindings need no per-turn read. */
|
||||
copyBack: () => Promise<void>;
|
||||
/** Remove launch credentials after the provider owner has stopped. */
|
||||
remove: () => Promise<void>;
|
||||
/** Save the final refresh directly through durable authority before removal. */
|
||||
retire?: () => Promise<void>;
|
||||
retirement?: Promise<void>;
|
||||
detached: boolean;
|
||||
closing?: boolean;
|
||||
completed: boolean;
|
||||
@@ -17,7 +20,7 @@ const credentialTurns = new WeakMap<NativeSession, CredentialTurn>();
|
||||
/** A warm session outlives an invocation's temporary credential destination. */
|
||||
export function bindManagedNativeCredentialTurn(
|
||||
session: NativeSession,
|
||||
callbacks: Pick<CredentialTurn, "copyBack" | "remove">,
|
||||
callbacks: Pick<CredentialTurn, "copyBack" | "remove" | "retire">,
|
||||
): NativeSession {
|
||||
const previous = credentialTurns.get(session);
|
||||
// A successor controller must attach a new handle. Reusing a detached or
|
||||
@@ -49,6 +52,10 @@ export function bindManagedNativeCredentialTurn(
|
||||
if (owner) owner.closing = true;
|
||||
await close(input);
|
||||
if (!owner || owner.detached || credentialTurns.get(session) !== owner) return;
|
||||
if (owner.retire && !owner.removed) {
|
||||
owner.retirement ??= owner.retire().catch(error => { owner.retirement = undefined; throw error; });
|
||||
await owner.retirement;
|
||||
}
|
||||
try {
|
||||
await completeCredentialTurn(owner);
|
||||
} finally {
|
||||
@@ -69,7 +76,7 @@ async function completeCredentialTurn(turn: CredentialTurn) {
|
||||
await turn.completion;
|
||||
}
|
||||
|
||||
/** Flush refreshes after each turn, without closing a healthy warm provider. */
|
||||
/** Complete a turn without closing the provider; event-backed bindings flush through PRP. */
|
||||
export async function completeManagedNativeCredentialTurn(session: NativeSession | undefined) {
|
||||
if (!session) return;
|
||||
const turn = credentialTurns.get(session);
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { mkdtemp, realpath, mkdir, writeFile, rm, symlink, chmod } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { tmpdir } from "node:os";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { createNativeCredentialStore } from "./native-credential-store.js";
|
||||
|
||||
const execute = promisify(execFile), roots: string[] = [];
|
||||
afterEach(async () => { await Promise.all(roots.splice(0).map(root => rm(root, { recursive: true, force: true }))); });
|
||||
async function fixture(target: "local" | "ssh" | "sandbox", provider: "codex" | "grok" = "codex") {
|
||||
const root = await realpath(await mkdtemp(join(tmpdir(), "native-credential-store-"))); roots.push(root);
|
||||
const localHome = join(root, "local"), remoteHome = join(root, "remote");
|
||||
await mkdir(localHome); await mkdir(remoteHome);
|
||||
const home = target === "local" ? localHome : remoteHome;
|
||||
const runner = { execute: vi.fn(async (request: { command: string; args: string[]; bypassSession?: boolean; timeoutMs?: number }) => {
|
||||
expect(request.bypassSession).toBe(true); expect(request.timeoutMs).toBe(10_000);
|
||||
try {
|
||||
const result = await execute(request.command, request.args, { timeout: request.timeoutMs });
|
||||
return { exitCode: 0, stdout: result.stdout, stderr: result.stderr };
|
||||
} catch (error) {
|
||||
return { exitCode: (error as { code: number }).code, stdout: "", stderr: "" };
|
||||
}
|
||||
}) };
|
||||
return { home, runner, store: createNativeCredentialStore({ provider, localHome,
|
||||
...(target === "local" ? {} : { remoteHome, runner: runner as never }) }) };
|
||||
}
|
||||
|
||||
it.each(["local", "ssh", "sandbox"] as const)("exports %s snapshots only on demand and removes them after retirement", async target => {
|
||||
const { home, runner, store } = await fixture(target);
|
||||
expect(runner.execute).not.toHaveBeenCalled();
|
||||
await writeFile(join(home, "auth.json"), '{"tokens":{"refresh_token":"synthetic"}}', { mode: 0o600 });
|
||||
expect(await store.read()).toContain("synthetic");
|
||||
await store.remove();
|
||||
await expect(store.read()).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it.each((["codex", "grok"] as const).flatMap(provider => (["local", "ssh", "sandbox"] as const).map(target => ({ provider, target }))))("uses the $provider close handoff for $target when auth.json is absent", async ({ target, provider }) => {
|
||||
const { home, store } = await fixture(target, provider);
|
||||
await writeFile(join(home, "auth-refresh.json"), '{"synthetic":"refresh"}', { mode: 0o600 });
|
||||
expect(await store.read()).toBe('{"synthetic":"refresh"}');
|
||||
await store.remove();
|
||||
await expect(store.read()).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it.each(["local", "ssh", "sandbox"] as const)("rejects unsafe %s files without falling back to another credential", async target => {
|
||||
const { home, store } = await fixture(target, "grok");
|
||||
await writeFile(join(home, "auth-refresh.json"), "must-not-read", { mode: 0o600 });
|
||||
await writeFile(join(home, "auth.json"), "unsafe", { mode: 0o644 });
|
||||
await expect(store.read()).rejects.toThrow();
|
||||
await chmod(join(home, "auth.json"), 0o600);
|
||||
await writeFile(join(home, "auth.json"), "x".repeat(65_537));
|
||||
await expect(store.read()).rejects.toThrow();
|
||||
await rm(join(home, "auth.json"));
|
||||
await symlink(join(home, "auth-refresh.json"), join(home, "auth.json"));
|
||||
await expect(store.read()).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("rejects truncated remote replies and timeouts", async () => {
|
||||
const runner = { execute: vi.fn(async () => ({ exitCode: 0, stdout: "not-base64!", timedOut: false })) };
|
||||
const store = createNativeCredentialStore({ provider: "codex", localHome: "/unused", remoteHome: "/owned", runner: runner as never });
|
||||
await expect(store.read()).rejects.toThrow("export invalid");
|
||||
runner.execute.mockResolvedValue({ exitCode: 0, stdout: "", timedOut: true });
|
||||
await expect(store.read()).rejects.toThrow("export unavailable");
|
||||
});
|
||||
@@ -0,0 +1,76 @@
|
||||
import { rm } from "node:fs/promises";
|
||||
import { join, posix } from "node:path";
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
import { readLocalAiCredentialFile } from "../local-ai-credential-file.js";
|
||||
|
||||
// Descriptor reads keep credentials private, bounded, and independent of the workspace.
|
||||
const READ_CREDENTIAL = `
|
||||
const fs = require('node:fs'), path = require('node:path');
|
||||
let fd;
|
||||
try {
|
||||
const file = process.argv[1];
|
||||
let parent = path.dirname(file);
|
||||
while (true) {
|
||||
if (!fs.lstatSync(parent).isDirectory()) throw Error('directory');
|
||||
const next = path.dirname(parent);
|
||||
if (next === parent) break;
|
||||
parent = next;
|
||||
}
|
||||
fd = fs.openSync(file, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW | fs.constants.O_NONBLOCK);
|
||||
const before = fs.fstatSync(fd);
|
||||
if (!before.isFile() || before.uid !== process.getuid() || (before.mode & 511) !== 384 || before.size > 65536) throw Error('credential');
|
||||
const bytes = Buffer.alloc(65537);
|
||||
let size = 0;
|
||||
while (size < bytes.length) {
|
||||
const count = fs.readSync(fd, bytes, size, bytes.length - size, size);
|
||||
if (!count) break;
|
||||
size += count;
|
||||
}
|
||||
const after = fs.fstatSync(fd);
|
||||
if (size > 65536 || ['dev', 'ino', 'size', 'mtimeMs', 'ctimeMs'].some(key => before[key] !== after[key])) throw Error('credential');
|
||||
process.stdout.write(bytes.subarray(0, size).toString('base64'));
|
||||
bytes.fill(0);
|
||||
} catch (error) {
|
||||
process.exitCode = error.code === 'ENOENT' ? 66 : 1;
|
||||
} finally {
|
||||
if (fd !== undefined) fs.closeSync(fd);
|
||||
}`;
|
||||
|
||||
export function createNativeCredentialStore(input: {
|
||||
provider: "codex" | "grok";
|
||||
localHome: string;
|
||||
remoteHome?: string;
|
||||
runner?: Pick<CommandManagedRuntimeRunner, "execute">;
|
||||
}) {
|
||||
if (Boolean(input.remoteHome) !== Boolean(input.runner)) throw new Error("native_credential_store_target_unavailable");
|
||||
const names = ["auth.json", "auth-refresh.json"];
|
||||
const read = async (name: string): Promise<string> => {
|
||||
if (!input.runner || !input.remoteHome) return readLocalAiCredentialFile(join(input.localHome, name));
|
||||
// Remote isolated-home preparation already requires Node. Read only on a refresh or retirement.
|
||||
const result = await input.runner.execute({ command: "node", args: ["-e", READ_CREDENTIAL, posix.join(input.remoteHome, name)], bypassSession: true, timeoutMs: 10_000 });
|
||||
if (result.exitCode !== 0 || result.timedOut) throw Object.assign(new Error("Native credential export unavailable"), {
|
||||
code: result.exitCode === 66 && !result.timedOut ? "ENOENT" : "INVALID_CREDENTIAL",
|
||||
});
|
||||
const encoded = result.stdout.trim(), bytes = Buffer.from(encoded, "base64");
|
||||
if (bytes.length > 65_536 || bytes.toString("base64") !== encoded) throw new Error("Native credential export invalid");
|
||||
return bytes.toString("utf8");
|
||||
};
|
||||
return {
|
||||
read: async () => {
|
||||
try { return await read(names[0]!); }
|
||||
catch (error) {
|
||||
if (names.length !== 2 || (error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
|
||||
return read(names[1]!);
|
||||
}
|
||||
},
|
||||
remove: async () => {
|
||||
for (const name of [...names, "auth-refresh.json.tmp"]) {
|
||||
await rm(join(input.localHome, name), { force: true });
|
||||
if (input.runner && input.remoteHome) {
|
||||
const result = await input.runner.execute({ command: "rm", args: ["-f", "--", posix.join(input.remoteHome, name)], bypassSession: true, timeoutMs: 10_000 });
|
||||
if (result.exitCode !== 0 || result.timedOut) throw new Error("Native credential cleanup failed");
|
||||
}
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
import { expect, it, vi } from "vitest";
|
||||
import type { NativeSession, PrpEvent } from "../../vendor/paperclip-runner/index.js";
|
||||
import { createNativeCredentialUpdateOwner } from "./native-credential-updates.js";
|
||||
|
||||
const event = { eventType: "auth.credentials.updated" } as PrpEvent;
|
||||
it("handles bootstrap, idle refreshes, and successive warm invocation exporters", async () => {
|
||||
const first = vi.fn(async () => {}), next = vi.fn(async () => {});
|
||||
const owner = createNativeCredentialUpdateOwner(first);
|
||||
await owner.onCredentialsUpdated(event);
|
||||
const session = { close: vi.fn(async () => {}) } as unknown as NativeSession;
|
||||
owner.bindSession(session);
|
||||
// Completing a turn and deleting its invocation home does not retire this owner.
|
||||
await owner.onCredentialsUpdated(event);
|
||||
createNativeCredentialUpdateOwner(next).bindSession(session);
|
||||
await owner.onCredentialsUpdated(event);
|
||||
expect(first).toHaveBeenCalledTimes(2);
|
||||
expect(next).toHaveBeenCalledExactlyOnceWith(event);
|
||||
});
|
||||
|
||||
it("keeps the handler available for trailing close events and rejects closing-handle rebind", async () => {
|
||||
const handle = vi.fn(async () => {}), owner = createNativeCredentialUpdateOwner(handle);
|
||||
const session = { close: async () => { await owner.onCredentialsUpdated(event); } } as unknown as NativeSession;
|
||||
owner.bindSession(session);
|
||||
await session.close({ reason: "retirement" });
|
||||
expect(handle).toHaveBeenCalledExactlyOnceWith(event);
|
||||
expect(() => createNativeCredentialUpdateOwner(handle).bindSession(session)).toThrow("owner_retired");
|
||||
});
|
||||
|
||||
it("fences detached controllers and gives recovered handles independent ownership", async () => {
|
||||
const handle = vi.fn(async () => {}), owner = createNativeCredentialUpdateOwner(handle);
|
||||
const session = { close: async () => {}, detachControllerForRestart: async () => {} } as unknown as NativeSession;
|
||||
owner.bindSession(session);
|
||||
await session.detachControllerForRestart!();
|
||||
await expect(owner.onCredentialsUpdated(event)).rejects.toThrow("owner_detached");
|
||||
expect(() => createNativeCredentialUpdateOwner(handle).bindSession(session)).toThrow("owner_retired");
|
||||
expect(handle).not.toHaveBeenCalled();
|
||||
const recovered = createNativeCredentialUpdateOwner(handle);
|
||||
recovered.bindSession({ close: async () => {} } as unknown as NativeSession);
|
||||
await recovered.onCredentialsUpdated(event);
|
||||
expect(handle).toHaveBeenCalledOnce();
|
||||
});
|
||||
@@ -1,5 +1,5 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { afterAll, beforeAll, expect, it } from "vitest";
|
||||
import { afterAll, beforeAll, expect, it, vi } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
|
||||
import type { PrpEvent } from "../../vendor/paperclip-runner/index.js";
|
||||
@@ -61,3 +61,19 @@ it("rejects unbound runs and credential payloads with secret fields", async () =
|
||||
payload: { ...event.payload, access_token: "synthetic-secret" },
|
||||
} })).rejects.toThrow("native_credential_update_binding_mismatch");
|
||||
});
|
||||
|
||||
it("retries credential writeback after the event is durably recorded", async () => {
|
||||
const synchronizeCredentials = vi.fn().mockRejectedValueOnce(new Error("synthetic writeback failure"))
|
||||
.mockResolvedValue(undefined);
|
||||
const update = { ...event, sourceSeq: 10, sourceEventId: "credential_update_retry", payload: { ...event.payload, revision: 2 } };
|
||||
await expect(appendNativeCredentialUpdate(db, { ...binding, event: update, synchronizeCredentials }))
|
||||
.rejects.toThrow("synthetic writeback failure");
|
||||
await appendNativeCredentialUpdate(db, { ...binding, event: update, synchronizeCredentials });
|
||||
expect(synchronizeCredentials).toHaveBeenCalledTimes(2);
|
||||
const rows = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId));
|
||||
expect(rows.filter(row => row.sourceEventId === `credentials:${update.sourceEventId}`)).toHaveLength(1);
|
||||
synchronizeCredentials.mockClear();
|
||||
await expect(appendNativeCredentialUpdate(db, { ...binding, companyId: randomUUID(), event: update, synchronizeCredentials }))
|
||||
.rejects.toThrow("binding_mismatch");
|
||||
expect(synchronizeCredentials).not.toHaveBeenCalled();
|
||||
});
|
||||
@@ -1,6 +1,6 @@
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { validatePrpEvent, type PrpEvent } from "../../vendor/paperclip-runner/index.js";
|
||||
import { validatePrpEvent, type PrpEvent, type NativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
|
||||
|
||||
export async function appendNativeCredentialUpdate(db: Db, input: {
|
||||
@@ -9,6 +9,7 @@ export async function appendNativeCredentialUpdate(db: Db, input: {
|
||||
normalizedSessionId: string;
|
||||
runnerInstanceId: string;
|
||||
event: PrpEvent;
|
||||
synchronizeCredentials?: () => Promise<unknown>;
|
||||
}): Promise<void> {
|
||||
const parsed = validatePrpEvent(input.event);
|
||||
if (!parsed.ok || parsed.event.eventType !== "auth.credentials.updated"
|
||||
@@ -42,4 +43,32 @@ export async function appendNativeCredentialUpdate(db: Db, input: {
|
||||
},
|
||||
});
|
||||
});
|
||||
await input.synchronizeCredentials?.();
|
||||
}
|
||||
|
||||
type CredentialUpdateOwner = { handle: (event: PrpEvent) => Promise<void>; detached: boolean; closing: boolean };
|
||||
const credentialUpdateOwners = new WeakMap<NativeSession, CredentialUpdateOwner>();
|
||||
|
||||
/** The transport keeps its owner while each warm invocation rebinds the exporter. */
|
||||
export function createNativeCredentialUpdateOwner(handle: CredentialUpdateOwner["handle"]) {
|
||||
const owner: CredentialUpdateOwner = { handle, detached: false, closing: false };
|
||||
return {
|
||||
onCredentialsUpdated: async (event: PrpEvent) => {
|
||||
if (owner.detached) throw new Error("native_credential_update_owner_detached");
|
||||
await owner.handle(event);
|
||||
},
|
||||
bindSession: (session: NativeSession) => {
|
||||
const previous = credentialUpdateOwners.get(session);
|
||||
if (previous) {
|
||||
if (previous.detached || previous.closing) throw new Error("native_credential_update_owner_retired");
|
||||
previous.handle = handle;
|
||||
return;
|
||||
}
|
||||
credentialUpdateOwners.set(session, owner);
|
||||
const close = session.close.bind(session);
|
||||
session.close = async options => { owner.closing = true; await close(options); };
|
||||
const detach = session.detachControllerForRestart?.bind(session);
|
||||
if (detach) session.detachControllerForRestart = async () => { owner.detached = true; await detach(); };
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import type { NativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
import { currentControllerProcess } from "../controller-process-identity.js";
|
||||
import { activeNativeExecutionRunIds } from "./native-active-runs.js";
|
||||
import { onRetainedNativeSessionCleanupRelease } from "../retained-native-session-cleanup-notifications.js";
|
||||
@@ -125,6 +126,10 @@ import { canonicalProviderPackJson } from "./remote-provider-pack-verification.j
|
||||
import * as providerPackAdmission from "./native-provider-pack-admission.js";
|
||||
import * as providerArtifactStorage from "./native-provider-artifacts.js";
|
||||
import * as providerArtifactCapture from "./native-provider-artifact-capture.js";
|
||||
import * as nativeCredentialEvents from "./native-credential-updates.js";
|
||||
import * as nativeCredentialStores from "./native-credential-store.js";
|
||||
import * as managedCredentialWriteback from "../managed-ai-credential-writeback.js";
|
||||
import { completeManagedNativeCredentialTurn } from "./managed-native-credentials.js";
|
||||
import * as nativeExecutionDriver from "./native-runner-execution-driver.js";
|
||||
|
||||
vi.mock("node:fs", async (importOriginal) => {
|
||||
@@ -180,6 +185,7 @@ type BackendFactoryOptions = {
|
||||
type RunnerTransportOptions = {
|
||||
authorityRunId?: string;
|
||||
onAccountingEvent?: (event: PrpEvent) => Promise<void>;
|
||||
onCredentialsUpdated?: (event: PrpEvent) => Promise<void>;
|
||||
onRecoveryAccountingEvent?: (event: PrpEvent, identity: import("../../vendor/paperclip-runner/index.js").DurableRecoveryIdentity) => Promise<void>;
|
||||
environment?: NodeJS.ProcessEnv;
|
||||
acpxRuntimeDirectory?: string;
|
||||
@@ -2430,6 +2436,7 @@ describe("remote provider pack manifest", () => {
|
||||
runnerArtifact?: { version: string; digest: string };
|
||||
};
|
||||
expect(transport.runnerBinary).toBe(state.bundledRunnerBinary);
|
||||
expect(transport.onCredentialsUpdated).toBeUndefined();
|
||||
if (opencode) {
|
||||
expect(state.createAssignedMcpTools).toHaveBeenCalledWith(expect.objectContaining({ gatewayPublicId: "assigned-fixture", bearerToken: "private-fixture", delivery: "on_demand" }));
|
||||
for (const name of ["PAPERCLIP_NATIVE_MCP_NAME", "PAPERCLIP_NATIVE_MCP_URL", "PAPERCLIP_NATIVE_MCP_TOKEN"]) expect(transport.environment?.[name]).toBeUndefined();
|
||||
@@ -16824,6 +16831,74 @@ describe("runnerd provider runtime wiring", () => {
|
||||
expect(task.mock.calls.some(([op]) => ["prepare", "submit", "stop"].includes(op.kind))).toBe(false);
|
||||
});
|
||||
|
||||
it.each(["codex", "acpx-codex", "grok"] as const)("writes managed %s events through the current warm exporter without per-turn reads", async provider => {
|
||||
const root = await realpath(await mkdtemp(join(tmpdir(), "native-managed-refresh-")));
|
||||
const previousRoot = process.env.PAPERCLIP_RUNNER_STATE_DIR;
|
||||
process.env.PAPERCLIP_RUNNER_STATE_DIR = join(root, "runner");
|
||||
const firstHome = join(root, "first"), nextHome = join(root, "next");
|
||||
await mkdir(firstHome); await mkdir(nextHome);
|
||||
for (const home of [firstHome, nextHome]) await writeFile(join(home, "auth.json"), "{}", { mode: 0o600 });
|
||||
const current = scopedNativeExecution(`managed-refresh-${provider}`);
|
||||
current.session.lifecyclePolicy = { mode: "warm", idleTimeoutMs: 60_000 };
|
||||
if (provider !== "codex") {
|
||||
current.provider = { kind: "acpx", agent: provider === "grok" ? "grok" : "codex", model: null, permissionMode: "deny-all" } as never;
|
||||
current.session.driverKind = "acpx_runtime";
|
||||
}
|
||||
const firstStore = { read: vi.fn(async () => "first-refresh"), remove: vi.fn(async () => {}) };
|
||||
const nextStore = { read: vi.fn(async () => "next-refresh"), remove: vi.fn(async () => {}) };
|
||||
const stores = vi.spyOn(nativeCredentialStores, "createNativeCredentialStore").mockReturnValueOnce(firstStore).mockReturnValueOnce(nextStore);
|
||||
const persist = vi.spyOn(managedCredentialWriteback, "syncManagedAiCredentials").mockImplementation(async (_db, _company, _binding, _provider, read) => {
|
||||
await read(); return "rotated";
|
||||
});
|
||||
const append = vi.spyOn(nativeCredentialEvents, "appendNativeCredentialUpdate").mockImplementation(async (_db, input) => { await input.synchronizeCredentials?.(); });
|
||||
const binding = { grantId: "10000000-0000-4000-8000-000000000001", secretId: "10000000-0000-4000-8000-000000000002", sessionEpoch: 1, versionSelector: "latest" as const };
|
||||
const session = { close: vi.fn(async () => {}) } as unknown as NativeSession;
|
||||
try {
|
||||
const backend = await createRunnerdBackend({ db: leaseDb(current), execution: current, runnerInstanceId: "managed-runner",
|
||||
managedAiCredentialHome: firstHome, managedAiCredentialWriteback: binding });
|
||||
state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!();
|
||||
const retainedHook = state.createTransport.mock.calls.at(-1)![0].onCredentialsUpdated!;
|
||||
backend.bindManagedSession(session);
|
||||
await completeManagedNativeCredentialTurn(session);
|
||||
expect(persist).not.toHaveBeenCalled(); expect(firstStore.read).not.toHaveBeenCalled();
|
||||
await rm(firstHome, { recursive: true, force: true });
|
||||
const notification: PrpEvent = {
|
||||
schema: "paperclip.prp.event.v3", schemaVersion: 3, eventType: "auth.credentials.updated", sourceKind: "runner",
|
||||
sourceInstanceId: "managed-runner", sourceEventId: "refresh", sourceSeq: 1, runId: current.binding.runId,
|
||||
normalizedSessionId: current.session.normalizedSessionId!, priority: 0, emittedAt: "2020-01-01T00:00:00Z",
|
||||
payload: { schema: "paperclip.auth.credentials.updated.v1", bindingId: "harness-auth", revision: 1,
|
||||
detection: "credential_store", provider: provider === "grok" ? "grok" : "codex" },
|
||||
};
|
||||
await retainedHook(notification);
|
||||
expect(firstStore.read).toHaveBeenCalledOnce();
|
||||
// The mocked transport does not publish a durable Runner identity.
|
||||
await rm(state.createTransport.mock.calls.at(-1)![0].stateDirectory!, { recursive: true, force: true });
|
||||
const next = { ...current, binding: { ...current.binding, runId: `${current.binding.runId}-next` } };
|
||||
const nextBinding = { ...binding };
|
||||
const successor = await createRunnerdBackend({ db: leaseDb(next), execution: next, runnerInstanceId: "managed-runner",
|
||||
managedAiCredentialHome: nextHome, managedAiCredentialWriteback: nextBinding });
|
||||
successor.bindManagedSession(session);
|
||||
await completeManagedNativeCredentialTurn(session);
|
||||
await rm(nextHome, { recursive: true, force: true });
|
||||
await retainedHook(notification);
|
||||
expect(persist).toHaveBeenLastCalledWith(expect.anything(), current.binding.companyId, nextBinding,
|
||||
provider === "grok" ? "grok" : "codex", nextStore.read);
|
||||
expect(persist.mock.calls.at(-1)![2]).toBe(nextBinding);
|
||||
await session.close({ reason: "idle retirement" });
|
||||
expect(nextStore.read).toHaveBeenCalledTimes(2);
|
||||
expect(firstStore.remove).not.toHaveBeenCalled(); expect(nextStore.remove).toHaveBeenCalledOnce();
|
||||
expect(stores.mock.calls[0]![0].localHome).toContain(provider === "grok" ? "grok-home" : "codex-home");
|
||||
persist.mockClear();
|
||||
await expect(retainedHook({ ...notification, payload: { provider: "unexpected" } })).rejects.toThrow("provider_mismatch");
|
||||
expect(persist).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
stores.mockRestore(); persist.mockRestore(); append.mockRestore();
|
||||
if (previousRoot === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR;
|
||||
else process.env.PAPERCLIP_RUNNER_STATE_DIR = previousRoot;
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["codex", "opencode", "acpx"] as const)("uses a task-only %s driver without exposing Core paths or transferring the agent home", async (kind) => {
|
||||
const taskExecution = { ...execution, provider: kind === "codex" ? execution.provider
|
||||
: kind === "opencode" ? { kind, model: "openai/test", permissionMode: "deny" }
|
||||
@@ -16836,7 +16911,7 @@ describe("runnerd provider runtime wiring", () => {
|
||||
state.toolAuthorityDefinitions.mockImplementation(async binding => binding.readRemoteTaskFile ? [{ name: "register_deliverable" }] as never : []);
|
||||
const readWorkspaceFile = vi.fn(async () => Buffer.from("proof"));
|
||||
const release = vi.fn(async () => {}), resume = vi.fn(async () => "resumed" as const);
|
||||
const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, release, resume, isRetired: () => false, retiredBeforeClaim: () => false, rebind:vi.fn() });
|
||||
const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue({ launcher, registration, driver, readWorkspaceFile, release, resume, isRetired: () => false, retiredBeforeClaim: () => false, credentialsUpdated: vi.fn(), rebind:vi.fn() });
|
||||
const task = vi.fn(async (operation: unknown) => {
|
||||
environmentTaskOperationSchema.parse(operation);
|
||||
return { kind: "description", launchPreparationVersion: 1, projectAuthCheckVersion: 1, releaseResumeVersion: 1, ...(kind === "codex" ? { workspaceFileReadVersion: 1 } : {}), taskId: "task", runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } };
|
||||
@@ -16863,6 +16938,9 @@ describe("runnerd provider runtime wiring", () => {
|
||||
const expectedTools = kind === "codex" ? [{ name: "register_deliverable" }] : [];
|
||||
expect(state.createBackend.mock.calls.at(-1)![1]).toMatchObject({ dynamicTools: expectedTools });
|
||||
expect(options).toMatchObject({ resumeDynamicTools: expectedTools });
|
||||
const notification = { eventType: "auth.credentials.updated" } as never;
|
||||
await options.onCredentialsUpdated!(notification);
|
||||
expect(owner.mock.results.at(-1)!.value.credentialsUpdated).toHaveBeenCalledWith(notification);
|
||||
expect(options).toMatchObject({ ...driver, runnerArtifact: { version: "fixture" },
|
||||
prpIdentity: { environmentLeaseId: "plugin-task-lease" } });
|
||||
if (kind === "acpx") expect(options).toMatchObject({ acpxAgent: "claude", claudeEffort: "high" });
|
||||
|
||||
@@ -53,7 +53,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notificati
|
||||
import { writeExecutionControl } from "../execution-control-notifications.js";
|
||||
import { notifyExecutionRunWork } from "../execution-control-notifications.js";
|
||||
import { createRunUsageRecorder } from "../usage-receipts.js";
|
||||
import { registerPluginTaskCredentialHandoff, type PluginTaskCredentialWritebackBinding } from "../plugin-task-credential-writeback.js";
|
||||
import { syncPluginTaskCredentialHandoff, registerPluginTaskCredentialHandoff } from "../plugin-task-credential-writeback.js";
|
||||
import { resolvePluginTaskResume } from "../environment-task-runtime.js";
|
||||
import { environmentTaskOperationSchema, environmentTaskResultSchema } from "@paperclipai/plugin-sdk";
|
||||
import { preparePluginRunnerCredentials } from "./plugin-runner-credentials.js";
|
||||
@@ -105,6 +105,8 @@ import { RUNNERD_CONTROL_PLANE_STATE_SCHEMAS, runnerdStateIsIncompleteBootstrap,
|
||||
import { createNativeDeliverableTextObservation } from "./native-deliverable-text.js";
|
||||
import { createNativeGitHubAccess, type NativeGitHubAccess } from "./native-github-access.js";
|
||||
import { resolveGitHubOperationCredentials } from "../github-operation-credentials.js";
|
||||
import { syncManagedAiCredentials, type ManagedAiCredentialWritebackBinding } from "../managed-ai-credential-writeback.js";
|
||||
import { createNativeCredentialStore } from "./native-credential-store.js";
|
||||
import { bindManagedNativeCredentialTurn, completeManagedNativeCredentialTurn } from "./managed-native-credentials.js";
|
||||
import { createLocalNativeQuestionBridge } from "./local-native-question-bridge.js";
|
||||
import { createArtifactInstaller } from "./artifact-transfer.js";
|
||||
@@ -245,7 +247,7 @@ import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js";
|
||||
import { assertNativeRetainedSandboxBinding, deferNativeControllerRecovery, NativeControllerRecoveryWaitingError, readNativeControllerRecovery, resolveNativeControllerRecovery } from "./native-controller-recovery.js";
|
||||
import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js";
|
||||
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
|
||||
import { appendNativeCredentialUpdate } from "./native-credential-updates.js";
|
||||
import { createNativeCredentialUpdateOwner, appendNativeCredentialUpdate } from "./native-credential-updates.js";
|
||||
import { canonicalNativeJson, nativeSha256 } from "./canonical.js";
|
||||
import { lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js";
|
||||
import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js";
|
||||
@@ -9437,7 +9439,7 @@ export async function executePaperclipNativeSession(input: {
|
||||
managedAiCredentialHome?: string;
|
||||
managedAiCredentialIdentity?: string;
|
||||
managedAiAccountIdentity?: string;
|
||||
managedAiCredentialWriteback?: PluginTaskCredentialWritebackBinding;
|
||||
managedAiCredentialWriteback?: ManagedAiCredentialWritebackBinding;
|
||||
billingIdentity?: Pick<AdapterExecutionResult, "provider" | "biller" | "billingType">;
|
||||
onTaskDispatched?: () => Promise<void> | void;
|
||||
runnerExecutionTarget?: AdapterExecutionTarget | null;
|
||||
@@ -14423,7 +14425,7 @@ export async function createRunnerdBackend(input: {
|
||||
managedAiCredentialHome?: string;
|
||||
managedAiCredentialIdentity?: string;
|
||||
managedAiAccountIdentity?: string;
|
||||
managedAiCredentialWriteback?: PluginTaskCredentialWritebackBinding;
|
||||
managedAiCredentialWriteback?: ManagedAiCredentialWritebackBinding;
|
||||
onTaskDispatched?: () => Promise<void> | void;
|
||||
runnerExecutionTarget?: AdapterExecutionTarget | null;
|
||||
/** Resolved per-run authorization; not an independent instance setting. */
|
||||
@@ -14912,7 +14914,20 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
const ownerInput: Parameters<typeof createPluginRunnerTaskOwner>[0] = { ...credentials, taskId: description.taskId,
|
||||
executionContext: input.executionContext?.pluginSupported ? input.executionContext : undefined,
|
||||
// Warm retirement can outlive the run's temporary credential home; complete uses the durable handoff.
|
||||
onCredentials: input.execution.session.lifecyclePolicy.mode === "warm" ? undefined : credentials.onCredentials,
|
||||
onCredentials: input.managedAiCredentialWriteback || input.execution.session.lifecyclePolicy.mode === "warm" ? undefined : credentials.onCredentials,
|
||||
onCredentialsUpdated: event => appendNativeCredentialUpdate(input.db, {
|
||||
companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId,
|
||||
normalizedSessionId: nativeSessionKey(input.execution),
|
||||
runnerInstanceId: adoptionIdentity?.runnerInstanceId ?? prior?.runnerInstanceId ?? effectiveRunnerInstanceId,
|
||||
event,
|
||||
synchronizeCredentials: event.payload.provider === "codex" ? () => syncPluginTaskCredentialHandoff(input.db, {
|
||||
companyId: input.execution.binding.companyId, leaseId: remoteTarget.leaseId!, expectedTaskId: description.taskId,
|
||||
}, async accountIdentity => {
|
||||
const snapshot = environmentTaskResultSchema.parse(await pluginTask({ kind: "credentials", accountIdentity }, { expectedTaskId: description.taskId }));
|
||||
if (snapshot.kind !== "credentials") throw new Error("Plugin Runner credential snapshot unavailable");
|
||||
return snapshot.credential;
|
||||
}) : undefined,
|
||||
}),
|
||||
warmTaskVersion: description.warmTaskVersion,
|
||||
releaseResumeVersion: description.releaseResumeVersion,
|
||||
authorityRunId: description.releaseResumeVersion === 1 ? input.execution.binding.runId : undefined,
|
||||
@@ -14983,6 +14998,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
measureOperation: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? measureRunnerOperation : undefined,
|
||||
recordTimingReceipt: process.env.OTEL_EXPORTER_OTLP_ENDPOINT?.trim() ? recordRunnerTimingReceipt : undefined,
|
||||
...owner.driver, get authorityRunId() { return owner.driver.authorityRunId; }, runnerArtifact: input.controllerPersistence?.runnerArtifact ?? description.runner,
|
||||
onCredentialsUpdated: event => owner.credentialsUpdated(event),
|
||||
controlPlaneRegistration: input.controllerPersistence ? async (authority, identity, signal) => {
|
||||
await nativeRunnerCheckpointRegistration({
|
||||
...nativeRunnerCheckpointPersistence(input.db, input.controllerPersistence!.bootId,
|
||||
@@ -16468,6 +16484,33 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
owner => logger.warn({ companyId: owner.companyId, runId: owner.runId }, "Native session diagnostic recording failed"));
|
||||
const diagnosticLease = diagnostics.activate(diagnosticOwner, input.onLog);
|
||||
input.onDiagnosticLease?.(diagnosticLease);
|
||||
const credentialProvider = input.execution.provider.kind === "codex" ? "codex"
|
||||
: input.execution.provider.kind === "acpx" && (input.execution.provider.agent === "codex" || input.execution.provider.agent === "grok")
|
||||
? input.execution.provider.agent : null;
|
||||
const acpxCredential = input.execution.provider.kind === "acpx";
|
||||
const credentialRelativeHome = credentialProvider && acpxCredential
|
||||
? `acpx/acpx/${acpxRuntimeSessionDirectoryName(nativeSessionKey(input.execution))}/${credentialProvider}-home` : "codex-home";
|
||||
const credentialStore = credentialProvider && !preconfiguredHome ? createNativeCredentialStore({
|
||||
provider: credentialProvider,
|
||||
localHome: acpxCredential ? resolve(resolvePaperclipInstanceRoot(), "runtime", "paperclip-runner", credentialRelativeHome) : join(root, "codex-home"),
|
||||
...(remoteRunnerFilesystemRoot && remoteCommandRunner ? {
|
||||
remoteHome: posix.join(remoteRunnerFilesystemRoot, credentialRelativeHome), runner: remoteCommandRunner,
|
||||
} : {}),
|
||||
}) : null;
|
||||
const synchronizeCredentials = input.managedAiCredentialWriteback && credentialProvider && credentialStore
|
||||
? async () => {
|
||||
const outcome = await syncManagedAiCredentials(input.db, input.execution.binding.companyId,
|
||||
input.managedAiCredentialWriteback!, credentialProvider, credentialStore.read);
|
||||
if (outcome === "no_credentials") throw new Error("Native credential snapshot unavailable");
|
||||
} : undefined;
|
||||
const credentialUpdates = credentialProvider ? createNativeCredentialUpdateOwner(event => appendNativeCredentialUpdate(input.db, {
|
||||
companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId,
|
||||
normalizedSessionId: nativeSessionKey(input.execution), runnerInstanceId: effectiveRunnerInstanceId, event,
|
||||
synchronizeCredentials: synchronizeCredentials ? async () => {
|
||||
if (event.payload.provider !== credentialProvider) throw new Error("native_credential_update_provider_mismatch");
|
||||
await synchronizeCredentials();
|
||||
} : undefined,
|
||||
})) : null;
|
||||
const backend = createNativeSessionBackend(runnerExecution, {
|
||||
retainedSystemInstructions,
|
||||
runnerInstanceId:
|
||||
@@ -16510,13 +16553,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
...(input.execution.provider.kind === "codex" ? { disableCodexWebSearch: "webSearch" in input.execution.provider && input.execution.provider.webSearch === "disabled" } : {}),
|
||||
homeMode: preconfiguredHome ? "preconfigured" : "isolated",
|
||||
...accountingTransportOptions,
|
||||
onCredentialsUpdated: (event) => appendNativeCredentialUpdate(input.db, {
|
||||
companyId: input.execution.binding.companyId,
|
||||
agentId: input.execution.binding.agentId,
|
||||
normalizedSessionId: nativeSessionKey(input.execution),
|
||||
runnerInstanceId: effectiveRunnerInstanceId,
|
||||
event,
|
||||
}),
|
||||
onCredentialsUpdated: credentialUpdates?.onCredentialsUpdated,
|
||||
provider:
|
||||
input.execution.provider.kind === "cursor_cloud" ? "cursor_cloud" : input.execution.provider.kind === "codex"
|
||||
? "codex"
|
||||
@@ -16829,6 +16866,7 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
});
|
||||
const boundManagedSessions = new WeakSet<NativeSession>();
|
||||
const wrapManagedSession = (session: NativeSession): NativeSession => {
|
||||
credentialUpdates?.bindSession(session);
|
||||
if (cursorCloudRuntime && !boundManagedSessions.has(session)) {
|
||||
boundManagedSessions.add(session);
|
||||
const close = session.close.bind(session);
|
||||
@@ -16846,61 +16884,18 @@ async function createRunnerdBackendWithinSessionClaim(
|
||||
managedHome: antigravityCredential.home, credential: antigravityCredential.credential, runtimeHome,
|
||||
});
|
||||
}
|
||||
if (isGrok && grokCredential?.home && !boundManagedSessions.has(session)) {
|
||||
boundManagedSessions.add(session);
|
||||
// The launch runtime directory already ends in "acpx"; ACPX adds its
|
||||
// own namespace beneath it in resolveAcpxRuntimeRoot.
|
||||
const relativeHome = `acpx/acpx/${acpxRuntimeSessionDirectoryName(nativeSessionKey(input.execution))}/grok-home`;
|
||||
const localHome = resolve(resolvePaperclipInstanceRoot(), "runtime", "paperclip-runner", relativeHome);
|
||||
const remoteHome = remoteRunnerFilesystemRoot ? posix.join(remoteRunnerFilesystemRoot, relativeHome) : null;
|
||||
const readAuth = async (name: string): Promise<Buffer> => {
|
||||
if (!remoteHome || !remoteCommandRunner) return Buffer.from(await readLocalAiCredentialFile(join(localHome, name)));
|
||||
const script = `const fs=require('node:fs'),path=require('node:path');let fd;try{const file=process.argv[1];let parent=path.dirname(file);while(true){if(!fs.lstatSync(parent).isDirectory())throw Error('directory');const next=path.dirname(parent);if(next===parent)break;parent=next;}fd=fs.openSync(file,fs.constants.O_RDONLY|fs.constants.O_NOFOLLOW);const st=fs.fstatSync(fd);if(!st.isFile()||st.uid!==process.getuid()||(st.mode&511)!==384||st.size>65536)throw Error('credential');const b=Buffer.alloc(65537);let n=0;while(n<b.length){const k=fs.readSync(fd,b,n,b.length-n,n);if(!k)break;n+=k;}if(n>65536)throw Error('size');process.stdout.write(b.subarray(0,n).toString('base64'));b.fill(0);}catch(e){process.exitCode=e.code==='ENOENT'?66:1;}finally{if(fd!==undefined)fs.closeSync(fd);}`;
|
||||
const result = await remoteCommandRunner.execute({ command: "node", args: ["-e", script, posix.join(remoteHome, name)], bypassSession: true, timeoutMs: 10000 });
|
||||
if (result.exitCode !== 0 || result.timedOut) throw Object.assign(new Error("Grok credential refresh handoff unavailable"), { code: result.exitCode === 66 ? "ENOENT" : "INVALID_CREDENTIAL" });
|
||||
return Buffer.from(result.stdout, "base64");
|
||||
};
|
||||
return bindManagedNativeCredentialTurn(session, {
|
||||
copyBack: async () => {
|
||||
await copyBackGrokAuth({ hostHomeDir: grokCredential.home!, log: () => {},
|
||||
readSandboxAuth: () => readAuth("auth.json").catch((error) => {
|
||||
if (error.code !== "ENOENT") throw error;
|
||||
return readAuth("auth-refresh.json");
|
||||
}),
|
||||
});
|
||||
},
|
||||
remove: async () => {
|
||||
for (const name of ["auth.json", "auth-refresh.json", "auth-refresh.json.tmp"]) {
|
||||
rmSync(join(localHome, name), { force: true });
|
||||
if (remoteHome && remoteCommandRunner) {
|
||||
const result = await remoteCommandRunner.execute({ command: "rm", args: ["-f", "--", posix.join(remoteHome, name)], bypassSession: true, timeoutMs: 10000 });
|
||||
if (result.exitCode !== 0 || result.timedOut) throw new Error("Grok credential cleanup failed");
|
||||
}
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
if (preconfiguredHome || !input.managedAiCredentialHome || input.execution.provider.kind !== "codex" || boundManagedSessions.has(session)) return session;
|
||||
const destinationHome = credentialProvider === "grok" ? grokCredential?.home : input.managedAiCredentialHome;
|
||||
if (!credentialProvider || !credentialStore || !destinationHome || boundManagedSessions.has(session)) return session;
|
||||
boundManagedSessions.add(session);
|
||||
const remoteAuth = remoteRunnerFilesystemRoot ? posix.join(remoteRunnerFilesystemRoot, "codex-home", "auth.json") : null;
|
||||
const localAuth = join(root, "codex-home", "auth.json");
|
||||
return bindManagedNativeCredentialTurn(session, {
|
||||
copyBack: async () => {
|
||||
await copyBackCodexAuth({
|
||||
hostAuthPath: join(input.managedAiCredentialHome!, "auth.json"),
|
||||
readSandboxAuth: async () => {
|
||||
if (!remoteAuth || !remoteCommandRunner) return readFileSync(localAuth);
|
||||
const result = await remoteCommandRunner.execute({ command: "base64", args: [remoteAuth], bypassSession: true, timeoutMs: 10000 });
|
||||
if (result.exitCode !== 0 || result.timedOut) throw new Error("AI credential copy-back failed");
|
||||
return Buffer.from(result.stdout, "base64");
|
||||
},
|
||||
log: () => {},
|
||||
});
|
||||
},
|
||||
remove: async () => {
|
||||
rmSync(localAuth, { force: true });
|
||||
if (remoteAuth && remoteCommandRunner) await remoteCommandRunner.execute({ command: "rm", args: ["-f", "--", remoteAuth], bypassSession: true, timeoutMs: 10000 });
|
||||
copyBack: synchronizeCredentials ? async () => {} : async () => {
|
||||
if (credentialProvider === "grok") await copyBackGrokAuth({ hostHomeDir: destinationHome,
|
||||
readSandboxAuth: async () => Buffer.from(await credentialStore.read()), log: () => {} });
|
||||
else await copyBackCodexAuth({ hostAuthPath: join(destinationHome, "auth.json"),
|
||||
readSandboxAuth: async () => Buffer.from(await credentialStore.read()), log: () => {} });
|
||||
},
|
||||
retire: synchronizeCredentials,
|
||||
remove: credentialStore.remove,
|
||||
});
|
||||
};
|
||||
input.executionContext?.signal.throwIfAborted();
|
||||
|
||||
@@ -856,3 +856,18 @@ it.each(["codex", "claude"] as const)("rejects a foreign %s source tuple before
|
||||
expect(task).not.toHaveBeenCalled(); expect(connect).not.toHaveBeenCalled();
|
||||
} finally { await registration.release!("detach"); }
|
||||
});
|
||||
|
||||
it("routes credential notifications to the current warm lease after rebind", async () => {
|
||||
const { task } = provider();
|
||||
const first = vi.fn(async () => {}), successor = vi.fn(async () => {});
|
||||
const input = { taskId: "task", task, projectIds: [], harness: "codex", adoptIdentity: requirements.identity,
|
||||
onCredentialsUpdated: first };
|
||||
const owner = createPluginRunnerTaskOwner(input);
|
||||
const event = { eventType: "auth.credentials.updated" } as never;
|
||||
await owner.credentialsUpdated(event);
|
||||
owner.rebind({ ...input, onCredentialsUpdated: successor });
|
||||
await owner.credentialsUpdated(event);
|
||||
expect(first).toHaveBeenCalledExactlyOnceWith(event);
|
||||
expect(successor).toHaveBeenCalledExactlyOnceWith(event);
|
||||
expect(task).not.toHaveBeenCalled();
|
||||
});
|
||||
@@ -4,7 +4,7 @@ import { setTimeout as delay } from "node:timers/promises";
|
||||
import { environmentTaskOperationSchema, environmentTaskResultSchema, parseEnvironmentTaskResult, type PluginEnvironmentTaskOperation, type PluginEnvironmentTaskResult, type PluginEnvironmentTaskCredential } from "@paperclipai/plugin-sdk";
|
||||
import type { RunnerIngressEndpoint } from "@paperclipai/adapter-utils/runner-connectivity";
|
||||
import { RUNNER_PRP_PROTOCOL_MIN, RUNNER_PRP_PROTOCOL_MAX, type RunnerProcessHandle, type RunnerProcessLaunchSpec,
|
||||
type DurablePrpControlPlane, type RunnerExecutionDriver, type RunnerLaunchPreparer, type DurableRecoveryIdentity } from "../../vendor/paperclip-runner/index.js";
|
||||
type DurablePrpControlPlane, type RunnerExecutionDriver, type RunnerLaunchPreparer, type DurableRecoveryIdentity, type PrpEvent } from "../../vendor/paperclip-runner/index.js";
|
||||
import { connectRunnerPrpIngress } from "../../realtime/runner-prp-outbound.js";
|
||||
import type { NativeExecutionContext } from "./native-execution-context.js";
|
||||
|
||||
@@ -43,6 +43,7 @@ export function createPluginRunnerTaskOwner(input: {
|
||||
accountIdentity?: string;
|
||||
credential?: PluginEnvironmentTaskCredential;
|
||||
onCredentials?: (credential: PluginEnvironmentTaskCredential) => Promise<void>;
|
||||
onCredentialsUpdated?: (event: PrpEvent) => Promise<void>;
|
||||
onSessionRetained?: (state: Record<string, unknown>) => Promise<void>;
|
||||
onDispatched?: () => Promise<void> | void;
|
||||
pollMs?: number;
|
||||
@@ -421,6 +422,8 @@ export function createPluginRunnerTaskOwner(input: {
|
||||
return result.outcome;
|
||||
};
|
||||
return { launcher, registration, driver, readWorkspaceFile, release, resume,
|
||||
// A warm transport outlives its first lease and invocation's private home.
|
||||
credentialsUpdated: async (event: PrpEvent) => { await input.onCredentialsUpdated?.(event); },
|
||||
// This is a lease-bound Launcher receipt, not a terminal phase or a sent
|
||||
// stop request. Completion also proves credential/writeback obligations ran.
|
||||
isRetired: () => stoppedState?.executionStopped === true && completed && !detached,
|
||||
|
||||
@@ -1,27 +1,14 @@
|
||||
import { aiConnectionMetadataSchema } from "@paperclipai/shared";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { companySecrets, connectionGrants, environmentLeases, toolConnections, type Db } from "@paperclipai/db";
|
||||
import { environmentLeases, type Db } from "@paperclipai/db";
|
||||
import { z, type PluginEnvironmentTaskCredential } from "@paperclipai/plugin-sdk";
|
||||
import { mkdtemp, writeFile, rm, realpath } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { isDeepStrictEqual } from "node:util";
|
||||
import { decideCodexAuthMerge, withAccountHomeSecretMutationLock } from "@paperclipai/adapter-codex-local/server";
|
||||
import { WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE } from "@paperclipai/adapter-utils/workspace-restore-merge";
|
||||
import { aiConnectionService } from "./ai-connections.js";
|
||||
import { secretService } from "./secrets.js";
|
||||
import { managedAiCredentialWritebackBindingSchema, syncManagedAiCredentials, type ManagedAiCredentialWritebackBinding } from "./managed-ai-credential-writeback.js";
|
||||
|
||||
const bindingSchema = z.object({
|
||||
grantId: z.string().uuid(),
|
||||
secretId: z.string().uuid(),
|
||||
sessionEpoch: z.number().int().nonnegative(),
|
||||
versionSelector: z.union([z.literal("latest"), z.number().int().positive()]),
|
||||
}).strict();
|
||||
export type PluginTaskCredentialWritebackBinding = z.infer<typeof bindingSchema>;
|
||||
const handoffSchema = z.object({
|
||||
accountIdentity: z.string().min(1),
|
||||
binding: bindingSchema,
|
||||
binding: managedAiCredentialWritebackBindingSchema,
|
||||
state: z.enum(["pending", "complete"]),
|
||||
outcome: z.enum(["rotated", "unchanged", "binding_changed", "not_dispatched", "no_credentials"]).optional(),
|
||||
}).strict();
|
||||
const key = "runnerTaskCredentialHandoff";
|
||||
type Scope = { companyId: string; leaseId: string; expectedTaskId?: string };
|
||||
@@ -31,7 +18,7 @@ export async function registerPluginTaskCredentialHandoff(
|
||||
db: Db,
|
||||
scope: Scope,
|
||||
accountIdentity: string,
|
||||
binding: PluginTaskCredentialWritebackBinding,
|
||||
binding: ManagedAiCredentialWritebackBinding,
|
||||
) {
|
||||
const handoff = handoffSchema.parse({ accountIdentity, binding, state: "pending" });
|
||||
await db.transaction(async tx => {
|
||||
@@ -55,6 +42,39 @@ export async function registerPluginTaskCredentialHandoff(
|
||||
});
|
||||
}
|
||||
|
||||
/** Live PRP updates use durable authority, including while a warm task is idle. */
|
||||
export async function syncPluginTaskCredentialHandoff(
|
||||
db: Db,
|
||||
scope: Scope,
|
||||
fetchCredential: (accountIdentity: string) => Promise<PluginEnvironmentTaskCredential | null>,
|
||||
) {
|
||||
const [lease] = await db.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, scope.companyId), eq(environmentLeases.id, scope.leaseId),
|
||||
scope.expectedTaskId === undefined ? undefined : eq(environmentLeases.providerLeaseId, scope.expectedTaskId),
|
||||
));
|
||||
if (!lease) throw new Error("Plugin task credential lease unavailable");
|
||||
if (!lease.metadata?.[key]) return;
|
||||
const handoff = handoffSchema.parse(lease.metadata[key]);
|
||||
if (handoff.state === "complete") return;
|
||||
const outcome = await writeBackCredentials(db, scope.companyId, handoff, fetchCredential);
|
||||
if (outcome === "no_credentials") throw new Error("Plugin task credential snapshot unavailable");
|
||||
return outcome;
|
||||
}
|
||||
|
||||
async function writeBackCredentials(
|
||||
db: Db,
|
||||
companyId: string,
|
||||
handoff: z.infer<typeof handoffSchema>,
|
||||
fetchCredential: (accountIdentity: string) => Promise<PluginEnvironmentTaskCredential | null>,
|
||||
) {
|
||||
return syncManagedAiCredentials(db, companyId, handoff.binding, "codex", async () => {
|
||||
const credential = await fetchCredential(handoff.accountIdentity);
|
||||
if (credential === null) return null;
|
||||
if (credential.type !== "codex-auth-json") throw new Error("Plugin task credential provider mismatch");
|
||||
return credential.authJson;
|
||||
});
|
||||
}
|
||||
|
||||
/** Runs for live completion and the cleanup sweep, including after Core restarts. */
|
||||
export async function settlePluginTaskCredentialHandoff(
|
||||
db: Db,
|
||||
@@ -65,81 +85,29 @@ export async function settlePluginTaskCredentialHandoff(
|
||||
if (!metadata[key]) return;
|
||||
const handoff = handoffSchema.parse(metadata[key]);
|
||||
if (handoff.state === "complete") return;
|
||||
if (metadata.runnerTaskDispatched === true && metadata.runnerTaskExecutionStopped !== true) {
|
||||
const dispatched = metadata.runnerTaskDispatched === true || !!metadata.runnerTaskIdentity;
|
||||
if (dispatched && metadata.runnerTaskExecutionStopped !== true) {
|
||||
throw new Error("Plugin task credential process termination is unconfirmed");
|
||||
}
|
||||
const binding = handoff.binding;
|
||||
const home = await mkdtemp(join(await realpath(tmpdir()), "paperclip-task-credential-"));
|
||||
try {
|
||||
// Revocation or an explicit account replacement intentionally discards an old refresh.
|
||||
const account = async (database: Db) => {
|
||||
const [grant] = await database.select().from(connectionGrants).where(and(
|
||||
eq(connectionGrants.companyId, scope.companyId), eq(connectionGrants.id, binding.grantId),
|
||||
)).for("update");
|
||||
if (!grant || grant.status !== "active") return null;
|
||||
const ref = grant.credentialSecretRefs.find(ref => ref.configPath === "ai.credential");
|
||||
if (!ref || ref.secretId !== binding.secretId ||
|
||||
(ref.versionSelector ?? "latest") !== binding.versionSelector) return null;
|
||||
const [secret] = await database.select().from(companySecrets).where(and(
|
||||
eq(companySecrets.companyId, scope.companyId), eq(companySecrets.id, binding.secretId),
|
||||
)).for("update");
|
||||
if (!secret || secret.aiSessionEpoch !== binding.sessionEpoch) return null;
|
||||
const [connection] = await database.select().from(toolConnections).where(and(
|
||||
eq(toolConnections.companyId, scope.companyId), eq(toolConnections.id, grant.connectionId),
|
||||
));
|
||||
const ai = aiConnectionMetadataSchema.safeParse(connection?.config.ai);
|
||||
if (!connection || !ai.success || ai.data.provider !== "openai" || ai.data.method !== "subscription") return null;
|
||||
return { connection, grant };
|
||||
};
|
||||
const mutate = <T>(operation: (database: Db) => Promise<T>) => withAccountHomeSecretMutationLock(
|
||||
undefined, scope.companyId, () => db.transaction(tx => operation(tx as unknown as Db)),
|
||||
);
|
||||
const current = metadata.runnerTaskDispatched === true && await mutate(account);
|
||||
if (current) {
|
||||
// Do not hold database or account locks while the provider exports its credential.
|
||||
const credential = await fetchCredential(handoff.accountIdentity);
|
||||
if (credential !== null) {
|
||||
if (credential.type !== "codex-auth-json") throw new Error("Plugin task credential provider mismatch");
|
||||
const incoming = join(home, "incoming.json"), destination = join(home, "current.json");
|
||||
await writeFile(incoming, credential.authJson, { mode: 0o600 });
|
||||
const writeBack = () => mutate(async database => {
|
||||
const selected = await account(database);
|
||||
if (!selected) return;
|
||||
const value = await aiConnectionService(database).credential(selected);
|
||||
await writeFile(destination, value, { mode: 0o600 });
|
||||
if (await decideCodexAuthMerge(incoming, destination, { errorLabel: "AI account refresh" }) !== 10) return;
|
||||
await secretService(database).rotate(binding.secretId, {
|
||||
value: credential.authJson, preserveAiSessionEpoch: true,
|
||||
}, { userId: selected.grant.subjectUserId });
|
||||
});
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
try {
|
||||
await writeBack();
|
||||
break;
|
||||
} catch (error) {
|
||||
if (attempt >= 2 || (error as NodeJS.ErrnoException)?.code !== WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE) throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
const completed = await db.update(environmentLeases).set({
|
||||
metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ [key]: { ...handoff, state: "complete" } })}::jsonb`,
|
||||
}).where(and(
|
||||
const outcome = dispatched
|
||||
? await writeBackCredentials(db, scope.companyId, handoff, fetchCredential)
|
||||
: "not_dispatched";
|
||||
const settled = { ...handoff, state: "complete" as const, outcome };
|
||||
const completed = await db.update(environmentLeases).set({
|
||||
metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ [key]: settled })}::jsonb`,
|
||||
}).where(and(
|
||||
eq(environmentLeases.companyId, scope.companyId), eq(environmentLeases.id, scope.leaseId),
|
||||
scope.expectedTaskId === undefined ? undefined : eq(environmentLeases.providerLeaseId, scope.expectedTaskId),
|
||||
sql`${environmentLeases.metadata}->${key} = ${JSON.stringify(handoff)}::jsonb`,
|
||||
)).returning({ id: environmentLeases.id });
|
||||
if (completed.length !== 1) {
|
||||
const [current] = await db.select({ metadata: environmentLeases.metadata, taskId: environmentLeases.providerLeaseId }).from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, scope.companyId), eq(environmentLeases.id, scope.leaseId),
|
||||
scope.expectedTaskId === undefined ? undefined : eq(environmentLeases.providerLeaseId, scope.expectedTaskId),
|
||||
sql`${environmentLeases.metadata}->${key} = ${JSON.stringify(handoff)}::jsonb`,
|
||||
)).returning({ id: environmentLeases.id });
|
||||
if (completed.length !== 1) {
|
||||
const [current] = await db.select({ metadata: environmentLeases.metadata, taskId: environmentLeases.providerLeaseId }).from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, scope.companyId), eq(environmentLeases.id, scope.leaseId),
|
||||
));
|
||||
if (scope.expectedTaskId !== undefined && current?.taskId !== scope.expectedTaskId ||
|
||||
!isDeepStrictEqual(handoffSchema.safeParse(current?.metadata?.[key]).data, { ...handoff, state: "complete" })) {
|
||||
throw new Error("Plugin task credential handoff changed");
|
||||
}
|
||||
));
|
||||
const saved = handoffSchema.safeParse(current?.metadata?.[key]);
|
||||
if (scope.expectedTaskId !== undefined && current?.taskId !== scope.expectedTaskId ||
|
||||
!saved.success || !isDeepStrictEqual({ ...saved.data, outcome: settled.outcome }, settled)) {
|
||||
throw new Error("Plugin task credential handoff changed");
|
||||
}
|
||||
} finally {
|
||||
// The provider retains the recoverable credential until completion succeeds.
|
||||
await rm(home, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user