diff --git a/packages/db/src/migrations/0448_agent_harness_verified_digest.sql b/packages/db/src/migrations/0448_agent_harness_verified_digest.sql new file mode 100644 index 00000000000..976b5f77da7 --- /dev/null +++ b/packages/db/src/migrations/0448_agent_harness_verified_digest.sql @@ -0,0 +1,2 @@ +ALTER TABLE "agents" ADD COLUMN "harness_verified_digest" text;--> statement-breakpoint +ALTER TABLE "agents" ADD COLUMN "harness_verified_at" timestamp with time zone; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0442_snapshot.json b/packages/db/src/migrations/meta/0448_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0442_snapshot.json rename to packages/db/src/migrations/meta/0448_snapshot.json index 2944c74ed51..05c25835806 100644 --- a/packages/db/src/migrations/meta/0442_snapshot.json +++ b/packages/db/src/migrations/meta/0448_snapshot.json @@ -1,6 +1,6 @@ { - "id": "b3f3d712-65da-4b5a-bbf5-3c22fe8aae40", - "prevId": "a2ec96c6-3cf1-4c51-8ee8-b35f090bfd42", + "id": "360fef2e-1ac3-4ae8-92a2-245e544b61da", + "prevId": "76f06e29-53a8-4ecc-809a-d52b6bbb6520", "version": "7", "dialect": "postgresql", "tables": { @@ -4252,6 +4252,18 @@ "primaryKey": false, "notNull": false }, + "harness_verified_digest": { + "name": "harness_verified_digest", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "harness_verified_at": { + "name": "harness_verified_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, "reports_to": { "name": "reports_to", "type": "uuid", @@ -34856,6 +34868,12 @@ "primaryKey": false, "notNull": false }, + "controller_process": { + "name": "controller_process", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "execution_stage": { "name": "execution_stage", "type": "text", @@ -35125,6 +35143,22 @@ "method": "btree", "with": {} }, + "heartbeat_runs_legacy_controller_lease_idx": { + "name": "heartbeat_runs_legacy_controller_lease_idx", + "columns": [ + { + "expression": "controller_lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"runtime_mode\" = 'legacy' AND \"heartbeat_runs\".\"status\" = 'running' AND \"heartbeat_runs\".\"controller_boot_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_cost_accounting_pending_idx": { "name": "heartbeat_runs_cost_accounting_pending_idx", "columns": [ @@ -44766,6 +44800,12 @@ "primaryKey": false, "notNull": false }, + "lease_holder": { + "name": "lease_holder", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "controller_boot_id": { "name": "controller_boot_id", "type": "text", @@ -44930,6 +44970,54 @@ "method": "btree", "with": {} }, + "native_run_finalizations_finalizer_lease_idx": { + "name": "native_run_finalizations_finalizer_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"lease_owner\" like 'native-finalizer:%'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_run_finalizations_cleanup_lease_idx": { + "name": "native_run_finalizations_cleanup_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"lease_owner\" like 'native-cleanup:%'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_run_finalizations_accepted_result_lease_idx": { + "name": "native_run_finalizations_accepted_result_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"result_id\" is not null AND \"native_run_finalizations\".\"lease_owner\" is not null AND \"native_run_finalizations\".\"phase\" not in ('committed', 'terminal_failure', 'applied')", + "concurrently": false, + "method": "btree", + "with": {} + }, "native_run_finalizations_control_deadline_idx": { "name": "native_run_finalizations_control_deadline_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 5ee6e128f06..3d335bebd8c 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3123,6 +3123,13 @@ "when": 1794105389611, "tag": "0447_legacy_controller_process", "breakpoints": true + }, + { + "idx": 448, + "version": "7", + "when": 1794135798799, + "tag": "0448_agent_harness_verified_digest", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/agents.ts b/packages/db/src/schema/agents.ts index 03152d6dffd..490596f154f 100644 --- a/packages/db/src/schema/agents.ts +++ b/packages/db/src/schema/agents.ts @@ -31,6 +31,9 @@ export const agents = pgTable( lifecycleVersion: integer("lifecycle_version").notNull().default(0), lifecycleError: text("lifecycle_error"), lifecycleOperation: jsonb("lifecycle_operation").$type(), + /** Digest of the effective configuration and runner image whose harness verification passed. */ + harnessVerifiedDigest: text("harness_verified_digest"), + harnessVerifiedAt: timestamp("harness_verified_at", { withTimezone: true }), reportsTo: uuid("reports_to").references((): AnyPgColumn => agents.id), capabilities: text("capabilities"), adapterType: text("adapter_type").notNull().default("process"), diff --git a/packages/shared/src/types/agent-lifecycle.ts b/packages/shared/src/types/agent-lifecycle.ts index 8eece70c235..fca9e2df1de 100644 --- a/packages/shared/src/types/agent-lifecycle.ts +++ b/packages/shared/src/types/agent-lifecycle.ts @@ -12,6 +12,8 @@ export interface AgentLifecycleOperation { completedPluginIds: string[]; leaseOwner?: string; leaseUntil?: string; + /** The controller process holding the lease; restart recovery settles it once proven gone. */ + leaseHolder?: { bootId: string; pid: number; hostHash: string; processStartedAt: string | null }; retryAt?: string; /** Healthy host verification awaits a durable external producer, not a retry timer. */ waitingForSignal?: boolean; diff --git a/server/src/__tests__/agent-harness-verification-preflight.test.ts b/server/src/__tests__/agent-harness-verification-preflight.test.ts index 95f2236af58..835cdf1061d 100644 --- a/server/src/__tests__/agent-harness-verification-preflight.test.ts +++ b/server/src/__tests__/agent-harness-verification-preflight.test.ts @@ -7,8 +7,10 @@ import { eq } from "drizzle-orm"; import { agents, aiProviderDefaults, companies, companyMemberships, connectionGrants, createDb, environments, plugins } from "@paperclipai/db"; import type { AdapterEnvironmentTestResult } from "@paperclipai/adapter-utils"; import { agentHarnessVerificationService } from "../services/agent-harness-verification.js"; +import * as serverAdapters from "../adapters/index.js"; import * as environmentTests from "../services/agent-environment-test.js"; import { aiConnectionService } from "../services/ai-connections.js"; +import { secretService } from "../services/secrets.js"; import { readAiConnectionConfigurationFailure } from "../services/ai-connection-configuration-failure.js"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -167,4 +169,104 @@ describePostgres("saved harness account preflight", () => { expect(allocate).toHaveBeenCalledOnce(); expect(probe).toHaveBeenCalledOnce(); }); + + it("reuses a passed verification of the same configuration and tests again after a change", async () => { + await expect(verify()).resolves.toBe("complete"); + expect(allocate).toHaveBeenCalledOnce(); + const [verified] = await db.select().from(agents).where(eq(agents.id, agent.id)); + expect(verified!.harnessVerifiedDigest).toMatch(/^v2:hmac-sha256:[a-f0-9]{64}$/); + // A restart, or a return to verifying, never boots another test target. + await expect(verify()).resolves.toBe("complete"); + expect(allocate).toHaveBeenCalledOnce(); + expect(probe).toHaveBeenCalledOnce(); + // Account checks still run: a revoked account fails before any reuse. + const [grant] = await db.select({ status: connectionGrants.status }).from(connectionGrants).where(eq(connectionGrants.id, account.grantId)); + await db.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.id, account.grantId)); + await expect(verify()).rejects.toThrow("Reconnect or validate the selected AI account"); + await db.update(connectionGrants).set({ status: grant!.status }).where(eq(connectionGrants.id, account.grantId)); + // A changed configuration is a new configuration, including values a + // session fingerprint ignores (cwd) and literal credentials. + const model = "openrouter/anthropic/claude-sonnet-4.5"; + let calls = 1; + for (const adapterConfig of [{ model: "openrouter/openai/gpt-5.1" }, { model, cwd: "/tmp/paperclip-missing-directory" }, + { model, env: { FIXTURE_TOKEN: "first-literal" } }, { model, env: { FIXTURE_TOKEN: "second-literal" } }]) { + await db.update(agents).set({ adapterConfig }).where(eq(agents.id, agent.id)); + await expect(verify()).resolves.toBe("complete"); + expect(allocate).toHaveBeenCalledTimes(++calls); + } + // The digest is keyed: it is not a plain hash a reader could recompute. + const [stored] = await db.select().from(agents).where(eq(agents.id, agent.id)); + expect(stored!.harnessVerifiedDigest).not.toContain("second-literal"); + }); + + it("tests again after the selected AI account's credential rotates", async () => { + await expect(verify()).resolves.toBe("complete"); + const [grant] = await db.select({ refs: connectionGrants.credentialSecretRefs }).from(connectionGrants) + .where(eq(connectionGrants.id, account.grantId)); + const secretIds = (grant?.refs ?? []).map(ref => ref.secretId); + expect(secretIds.length).toBeGreaterThan(0); + await secretService(db).rotate(secretIds[0]!, { value: "fixture-rotated-key" }, { userId }); + await expect(verify()).resolves.toBe("complete"); + expect(allocate).toHaveBeenCalledTimes(2); + }); + + it("keeps a pass whose probe refreshed its own credential, never one rotated by a person meanwhile", async () => { + const [grant] = await db.select({ refs: connectionGrants.credentialSecretRefs }).from(connectionGrants) + .where(eq(connectionGrants.id, account.grantId)); + const secretId = grant!.refs[0]!.secretId; + // The subscription refresh write-back keeps the AI session epoch. + probe.mockImplementationOnce(async () => { + await secretService(db).rotate(secretId, { value: "fixture-refreshed-key", preserveAiSessionEpoch: true }, { userId }); + return passed; + }); + await expect(verify()).resolves.toBe("complete"); + await expect(verify()).resolves.toBe("complete"); + expect(allocate).toHaveBeenCalledOnce(); + // A person's rotation during the probe bumps the epoch: nothing is recorded. + await db.update(agents).set({ harnessVerifiedDigest: null }).where(eq(agents.id, agent.id)); + probe.mockImplementationOnce(async () => { + await secretService(db).rotate(secretId, { value: "fixture-person-rotated-key" }, { userId }); + return passed; + }); + await expect(verify()).resolves.toBe("complete"); + expect((await db.select().from(agents).where(eq(agents.id, agent.id)))[0]!.harnessVerifiedDigest).toBeNull(); + }); + + it("records nothing when the probe tested a different account than the digest describes (A, then B, then A)", async () => { + const other = await aiConnectionService(db).save(agent.companyId, userId, { + provider: "openrouter", method: "api_key", ownership: "personal", name: "Second account", + apiKey: "fixture-second-key", allAgents: true, agentIds: [], + }, "fixture-second-key"); + // Switch to B after verify() selected A, before the probe selects its account. + allocate.mockImplementationOnce(async () => { + await aiConnectionService(db).setDefault(agent.companyId, userId, other.grantId); + return { executionTarget: null, environmentName: "Fixture execution target", fallbackChecks: [], release }; + }); + // The probe tests B; switch back to A before verify() could look again. + probe.mockImplementationOnce(async () => { + await aiConnectionService(db).setDefault(agent.companyId, userId, account.grantId); + return passed; + }); + await expect(verify()).resolves.toBe("complete"); + expect((await db.select().from(agents).where(eq(agents.id, agent.id)))[0]!.harnessVerifiedDigest).toBeNull(); + }); + + it("never reuses a local probe that may use the host's own credentials", async () => { + const localProbe = vi.fn().mockResolvedValue(passed); + vi.spyOn(serverAdapters, "requireServerAdapter").mockReturnValue({ testEnvironment: localProbe } as never); + await db.update(agents).set({ runtimeConfig: {} }).where(eq(agents.id, agent.id)); + await expect(verify()).resolves.toBe("complete"); + await expect(verify()).resolves.toBe("complete"); + expect(localProbe).toHaveBeenCalledTimes(2); + expect((await db.select().from(agents).where(eq(agents.id, agent.id)))[0]!.harnessVerifiedDigest).toBeNull(); + }); + + it("does not record a failed verification", async () => { + probe.mockResolvedValueOnce({ ...passed, status: "fail", checks: [{ code: "opencode_hello_probe_timed_out", level: "error", message: "timed out" }] }); + await expect(verify()).rejects.toMatchObject({ code: "harness_probe_timed_out" }); + const [failed] = await db.select().from(agents).where(eq(agents.id, agent.id)); + expect(failed!.harnessVerifiedDigest).toBeNull(); + await expect(verify()).resolves.toBe("complete"); + expect(allocate).toHaveBeenCalledTimes(2); + }); }); diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 2a1e0a9457a..68c853cac8c 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -143,7 +143,7 @@ describe("report-only ownership reconciliation", () => { expect(new Set(names).size).toBe(names.length); expect(names.length).toBeGreaterThanOrEqual(40); expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)) - .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "execution_workspace_runtime_lease", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "issue_execution_lock", "issue_checkout_lock"]); + .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "environment_login_probe_lease", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "execution_workspace_runtime_lease", "legacy_shutdown_settlement_fence", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "issue_execution_lock", "issue_checkout_lock", "agent_lifecycle_lease"]); expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true); }); diff --git a/server/src/lib/agent-records.ts b/server/src/lib/agent-records.ts index c28eb81aa18..f50b1c3df8a 100644 --- a/server/src/lib/agent-records.ts +++ b/server/src/lib/agent-records.ts @@ -163,7 +163,7 @@ export function hasConfigPatchFields(data: Partial) } export function changedPendingApprovalConfigFields( - existing: Omit, + existing: Omit, data: Partial, ) { return CONFIG_REVISION_FIELDS.filter((field) => @@ -326,7 +326,8 @@ export function agentRecordQueries(db: Db) { } function normalizeAgentBaseRow(row: typeof agents.$inferSelect) { - const { lifecycleOperation, lifecycleRequiredPluginIds, ...publicRow } = row; + // The harness verification digest is server-only reuse state, never an API field. + const { lifecycleOperation, lifecycleRequiredPluginIds, harnessVerifiedDigest, harnessVerifiedAt, ...publicRow } = row; return withUrlKey({ ...publicRow, permissions: normalizeAgentPermissions(row.permissions), diff --git a/server/src/modules/agent-lifecycle/__tests__/restart-verification.test.ts b/server/src/modules/agent-lifecycle/__tests__/restart-verification.test.ts new file mode 100644 index 00000000000..966aaa0b5bf --- /dev/null +++ b/server/src/modules/agent-lifecycle/__tests__/restart-verification.test.ts @@ -0,0 +1,97 @@ +import { randomUUID } from "node:crypto"; +import { agents, companies, createDb, type Db } from "@paperclipai/db"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js"; +import { createAgentLifecycle, configureAgentLifecycle, createAgentLifecycleEffects } from "../../../services/agent-lifecycle.js"; +import { startAgentLifecycle } from "../index.js"; +import { createLifecycleStore } from "../adapters/postgres.js"; +import type { LifecycleAgent } from "../application/ports.js"; +import { createOwnershipReportState, runOwnershipReconciliationPass } from "../../../services/recovery/ownership-invariant-report.js"; +import { + currentControllerProcessStamp, + currentProcessSpaceHash, + currentProcessStartedAtMs, +} from "../../../services/controller-process-identity.js"; + +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("harness verification across a control-plane restart", () => { + let database: Awaited>; + let db: Db; + let companyId: string; + const workers: Array<{ stop(): Promise }> = []; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("paperclip-restart-verification-"); + db = createDb(database.connectionString, { maxConnections: 4 }); + }, 90_000); + afterAll(async () => { await database?.cleanup(); }); + beforeEach(async () => { + companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Restart verification", issuePrefix: `R${companyId.slice(0, 7)}` }); + }); + afterEach(async () => { + for (const worker of workers.splice(0)) await worker.stop(); + await db.update(agents).set({ lifecycleState: "terminated", status: "terminated" }).where(eq(agents.companyId, companyId)); + }); + const current = async (id: string) => (await db.select().from(agents).where(eq(agents.id, id)))[0]!; + + /** An agent whose verification a controller claimed and was running when the controller died. */ + async function verifyingHeldBy(holder: Record) { + const runHost = vi.fn(async (agent: LifecycleAgent) => agent.lifecycleState === "verifying" ? "pending" as const : "complete" as const); + const before = configureAgentLifecycle(db, { requiredPluginIds: async () => [], runHost, runPlugin: async () => "complete" as const }); + const agent = await createAgentLifecycle(db).requestHire(companyId, { name: "Verifier", adapterType: "process", adapterConfig: { command: "true" } }); + await before.process(agent.id); + await before.stop(); + expect((await current(agent.id)).lifecycleState).toBe("verifying"); + await db.update(agents).set({ lifecycleOperation: sql`${agents.lifecycleOperation} - 'retryAt'` }).where(eq(agents.id, agent.id)); + const store = createLifecycleStore(db, createAgentLifecycleEffects()); + expect(await store.claim(agent.id, "verifier-before-restart", new Date())).not.toBeNull(); + await db.update(agents).set({ lifecycleOperation: sql`${agents.lifecycleOperation} || ${JSON.stringify({ leaseHolder: holder })}::jsonb` }) + .where(eq(agents.id, agent.id)); + return agent.id; + } + + /** The restarted process: one startup reconciliation pass, then its lifecycle sweep. */ + async function restart() { + const verify = vi.fn(async (agent: LifecycleAgent) => { void agent; return "complete" as const; }); + const restarted = startAgentLifecycle(db, createAgentLifecycleEffects(), + { requiredPluginIds: async () => [], runHost: verify, runPlugin: async () => "complete" as const }, () => true); + workers.push(restarted); + const startedAt = performance.now(); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + await restarted.sweep(); + return { verify, queuedMs: performance.now() - startedAt }; + } + + it("verifies again at once when the verifier's process is proven gone, not after the lease", async () => { + const id = await verifyingHeldBy({ bootId: randomUUID(), pid: 2_000_000_001, hostHash: currentProcessSpaceHash(), + processStartedAt: new Date(Date.now() - 60_000).toISOString() }); + const { verify, queuedMs } = await restart(); + expect(verify).toHaveBeenCalledOnce(); + // Runs gate on readiness; before this change they waited out the 120 s lease. + expect((await current(id)).lifecycleState).toBe("ready"); + expect(queuedMs).toBeLessThan(10_000); + }); + + it("never takes over a verification whose process is alive", async () => { + const id = await verifyingHeldBy({ ...currentControllerProcessStamp(), bootId: randomUUID(), pid: process.pid, + processStartedAt: new Date((await currentProcessStartedAtMs())!).toISOString() }); + const before = await current(id); + const { verify } = await restart(); + expect(verify).not.toHaveBeenCalled(); + const after = await current(id); + expect(after.lifecycleState).toBe("verifying"); + expect(after.lifecycleOperation?.leaseOwner).toBe(before.lifecycleOperation?.leaseOwner); + }); + + it("stamps the claiming process on the lease and clears it with the lease", async () => { + const id = await verifyingHeldBy(currentControllerProcessStamp() as unknown as Record); + const store = createLifecycleStore(db, createAgentLifecycleEffects()); + await db.update(agents).set({ lifecycleOperation: sql`${agents.lifecycleOperation} - 'leaseOwner' - 'leaseUntil' - 'leaseHolder'` }) + .where(eq(agents.id, id)); + const claimed = await store.claim(id, "owner", new Date()); + expect(claimed?.lifecycleOperation?.leaseHolder).toMatchObject({ pid: process.pid, bootId: currentControllerProcessStamp().bootId }); + expect(await store.defer(claimed!, "owner", null, new Date())).toBe(true); + expect((await current(id)).lifecycleOperation).not.toHaveProperty("leaseHolder"); + }); +}); diff --git a/server/src/modules/agent-lifecycle/adapters/configuration.ts b/server/src/modules/agent-lifecycle/adapters/configuration.ts index cec045f49a2..38af7db7013 100644 --- a/server/src/modules/agent-lifecycle/adapters/configuration.ts +++ b/server/src/modules/agent-lifecycle/adapters/configuration.ts @@ -48,7 +48,7 @@ export async function updateAgentConfiguration(tx: Db, effects: LifecycleEffects ...(!prepareSetup && current.lifecycleHolds.includes("setup") && ["preparing", "verifying"].includes(current.lifecycleState) ? { lifecycleState: "pausing" as const } : {}), lifecycleOperation: { ...current.lifecycleOperation!, id: randomUUID(), hostComplete: false, - completedPluginIds: [], attempts: 0, leaseOwner: undefined, leaseUntil: undefined, + completedPluginIds: [], attempts: 0, leaseOwner: undefined, leaseUntil: undefined, leaseHolder: undefined, retryAt: undefined, retryRequired: undefined, verificationFailures: undefined, waitingForSignal: undefined, recheckRequested: undefined }, } : {}) }).where(eq(agents.id, id)).returning(); diff --git a/server/src/modules/agent-lifecycle/adapters/effects.ts b/server/src/modules/agent-lifecycle/adapters/effects.ts index 43eec359774..1ccec84c599 100644 --- a/server/src/modules/agent-lifecycle/adapters/effects.ts +++ b/server/src/modules/agent-lifecycle/adapters/effects.ts @@ -10,6 +10,8 @@ export interface LifecycleEffects { profile: Pick): Promise; environmentRequiresNewPlugin(db: Db, environmentId: string | null | undefined, selectedPluginIds: string[] | null): Promise; notifyWork(db: Db): Promise; + /** The calling controller process, stamped on a lifecycle lease so restart recovery can settle a dead holder. */ + leaseHolder?(): Record | null; beforeStatusChange(db: Db, companyId: string, agentId: string, before: string, after: string): Promise; beforeConnectionEligibilityRestore?(db: Db, companyId: string, agentId: string): Promise; reportFailure(context: LifecycleFailure, error: unknown): void; diff --git a/server/src/modules/agent-lifecycle/adapters/postgres.ts b/server/src/modules/agent-lifecycle/adapters/postgres.ts index 8d5f9b1f576..12da4ae5c7d 100644 --- a/server/src/modules/agent-lifecycle/adapters/postgres.ts +++ b/server/src/modules/agent-lifecycle/adapters/postgres.ts @@ -30,6 +30,32 @@ export const lifecycleSignalWait = and( /** The producer owns this outer transaction. Register its intent before either * the recheck write or the corresponding external fact commits. */ +/** + * Releases a lifecycle lease whose holder process restart recovery proved gone. + * Compare-and-swap on that exact holder and lease owner; the lifecycle state, + * operation and progress are untouched, so the worker claims the step again. + */ +export async function releaseAgentLifecycleLeaseOfHolderInTransaction( + tx: Pick, + input: { agentId: string; holder: { bootId: string | null; pid: number | null } }, + notify: () => Promise, +): Promise<{ released: false; reason: string } | { released: true; agent: LifecycleRow; leaseUntil: string }> { + if (!("nestedIndex" in tx)) throw new AgentLifecycleConflict("Lifecycle lease release requires the caller transaction"); + const [agent] = await tx.select().from(agents).where(eq(agents.id, input.agentId)).for("update"); + const operation = agent?.lifecycleOperation; + if (!agent || !operation?.leaseOwner || !operation.leaseUntil) return { released: false, reason: "lease_no_longer_held" }; + if (operation.leaseHolder?.bootId !== input.holder.bootId || operation.leaseHolder?.pid !== input.holder.pid) { + return { released: false, reason: "owner_changed_since_proof" }; + } + // Register the wake intent before the write so drain and sleep checks see it. + await notify(); + const [row] = await tx.update(agents).set({ + lifecycleOperation: sql`${agents.lifecycleOperation} - 'leaseOwner' - 'leaseUntil' - 'leaseHolder'`, + }).where(and(eq(agents.id, agent.id), sql`${agents.lifecycleOperation}->>'leaseOwner' = ${operation.leaseOwner}`)) + .returning({ id: agents.id }); + return row ? { released: true, agent, leaseUntil: operation.leaseUntil } : { released: false, reason: "lease_no_longer_held" }; +} + export async function requestAgentLifecycleRecheckInTransaction( tx: Pick, input: { companyId: string; agentId: string; expectedVersion?: number; onlyIfWaiting?: boolean }, @@ -75,7 +101,7 @@ async function changeLockedAgent(tx: Db, effects: LifecycleEffects, agent: Lifec } await effects.notifyWork(tx); const [updated] = await tx.update(agents).set({ lifecycleError: null, - lifecycleOperation: sql`${agents.lifecycleOperation} - 'retryAt' - 'retryRequired' - 'verificationFailures' - 'leaseOwner' - 'leaseUntil' - 'waitingForSignal' - 'recheckRequested'`, updatedAt: new Date() }).where(eq(agents.id, agent.id)).returning(); + lifecycleOperation: sql`${agents.lifecycleOperation} - 'retryAt' - 'retryRequired' - 'verificationFailures' - 'leaseOwner' - 'leaseUntil' - 'leaseHolder' - 'waitingForSignal' - 'recheckRequested'`, updatedAt: new Date() }).where(eq(agents.id, agent.id)).returning(); return updated; } if (command === "resume" && !["ready", "paused", "pausing", "resuming"].includes(agent.lifecycleState)) transition(agent.lifecycleState, command); @@ -180,10 +206,13 @@ export function createLifecycleStore(db: Db, effects: LifecycleEffects): Lifecyc }); }, async claim(id, owner, now) { + // leaseHolder names the claiming process, so restart recovery settles a dead holder's lease at once. + const leaseHolder = effects.leaseHolder?.() ?? null; const [agent] = await db.update(agents).set({ lifecycleOperation: sql`(coalesce(${agents.lifecycleOperation}, '{"id":"", "hostComplete":false, "completedPluginIds":[], "attempts":0}'::jsonb) - 'recheckRequested') || jsonb_build_object('leaseOwner', ${owner}::text, 'leaseUntil', ${(new Date(now.getTime() + 120_000)).toISOString()}::text, - 'attempts', coalesce((${agents.lifecycleOperation}->>'attempts')::integer, 0) + 1)`, + 'attempts', coalesce((${agents.lifecycleOperation}->>'attempts')::integer, 0) + 1) + || ${JSON.stringify(leaseHolder ? { leaseHolder } : {})}::jsonb`, }).where(and(eq(agents.id, id), pendingLifecycleWork, sql`(${agents.lifecycleState} in ('pausing', 'terminating', 'cleaning_up') or exists ( select 1 from ${companies} where ${companies.id} = ${agents.companyId} and ${companies.status} = 'active'))`, @@ -224,7 +253,7 @@ export function createLifecycleStore(db: Db, effects: LifecycleEffects): Lifecyc async defer(agent, owner, error, now, options) { if (options?.waitForSignal && agent.lifecycleState === "verifying") { const rows = await db.update(agents).set({ lifecycleError: null, - lifecycleOperation: sql`(${agents.lifecycleOperation} - 'leaseOwner' - 'leaseUntil' - 'retryAt' - 'waitingForSignal' - 'recheckRequested') + lifecycleOperation: sql`(${agents.lifecycleOperation} - 'leaseOwner' - 'leaseUntil' - 'leaseHolder' - 'retryAt' - 'waitingForSignal' - 'recheckRequested') || case when coalesce(${agents.lifecycleOperation}->>'recheckRequested', 'false') = 'true' then '{}'::jsonb else '{"waitingForSignal":true}'::jsonb end`, updatedAt: now, @@ -240,7 +269,7 @@ export function createLifecycleStore(db: Db, effects: LifecycleEffects): Lifecyc } const rows = await db.update(agents).set({ lifecycleError: error, - lifecycleOperation: sql`(${agents.lifecycleOperation} - 'leaseOwner' - 'leaseUntil' - 'retryAt' - 'retryRequired' - 'waitingForSignal' - 'recheckRequested') + lifecycleOperation: sql`(${agents.lifecycleOperation} - 'leaseOwner' - 'leaseUntil' - 'leaseHolder' - 'retryAt' - 'retryRequired' - 'waitingForSignal' - 'recheckRequested') || ${JSON.stringify(retry)}::jsonb`, updatedAt: now, }).where(and(eq(agents.id, agent.id), eq(agents.lifecycleVersion, agent.lifecycleVersion), diff --git a/server/src/modules/agent-lifecycle/index.ts b/server/src/modules/agent-lifecycle/index.ts index 8fc083de80b..017a40d68cd 100644 --- a/server/src/modules/agent-lifecycle/index.ts +++ b/server/src/modules/agent-lifecycle/index.ts @@ -13,7 +13,7 @@ import { hasCompanyHoldMismatch, projectCompanyHolds } from "./adapters/company- import { createLifecycleWorker } from "./application/worker.js"; import type { LifecycleDriver } from "./application/ports.js"; -export { applyAgentPolicyHoldsInTransaction, requestAgentLifecycleRecheckInTransaction } from "./adapters/postgres.js"; +export { applyAgentPolicyHoldsInTransaction, releaseAgentLifecycleLeaseOfHolderInTransaction, requestAgentLifecycleRecheckInTransaction } from "./adapters/postgres.js"; export { companyHoldMismatch, fenceActiveCompanyAgentStopsInTransaction } from "./adapters/company-policy.js"; export { AgentLifecycleConflict, canConfigureAgentConnection, isAgentAwaitingSetup, isAgentSetupDraft } from "./domain/policy.js"; diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index 15841c87c98..7f338f50416 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -2963,11 +2963,13 @@ export function agentRoutes( agent: T, ): T { if (!agent || typeof agent !== "object") return agent; - if (!agent.adapterConfig || typeof agent.adapterConfig !== "object") return agent; + // A raw row must not carry the server-only harness verification digest. + const { harnessVerifiedDigest: _digest, harnessVerifiedAt: _verifiedAt, ...publicAgent } = agent as Record; + if (!agent.adapterConfig || typeof agent.adapterConfig !== "object") return publicAgent as T; return { - ...agent, + ...publicAgent, adapterConfig: redactAgentAdapterConfig(agent.adapterConfig as Record), - }; + } as T; } function redactAgentConfiguration(agent: Awaited>) { diff --git a/server/src/services/agent-harness-verification.ts b/server/src/services/agent-harness-verification.ts index 4dd8b43e1b0..05f8bc8abf5 100644 --- a/server/src/services/agent-harness-verification.ts +++ b/server/src/services/agent-harness-verification.ts @@ -1,5 +1,6 @@ -import { and, eq } from "drizzle-orm"; -import { agents, companies, type Db } from "@paperclipai/db"; +import { createHmac } from "node:crypto"; +import { and, eq, inArray } from "drizzle-orm"; +import { agents, companies, companySecrets, type Db } from "@paperclipai/db"; import type { AdapterEnvironmentTestResult } from "@paperclipai/adapter-utils"; import { ADAPTER_AUTH_MISSING_CHECK_CODE, aiRuntimeConnectionBindingSchema } from "@paperclipai/shared"; import { isForbiddenConfigEnvKey, parseObject } from "@paperclipai/adapter-utils/server-utils"; @@ -18,6 +19,30 @@ import { RUNNER_REVISION_MISMATCH_CODE } from "./runner-revision-mismatch.js"; import { dotRunnerBroker } from "./dot-runner-broker.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; import type { LifecycleAgent } from "../modules/agent-lifecycle/index.js"; +import { resolveDecisionSigningSecret } from "./decision-signing.js"; +import { sandboxBinding } from "./environment-probe.js"; +import { serverVersion } from "../version.js"; + +/** + * Digest of what a passed harness test proves: the agent's fully resolved + * effective configuration (every value the test used, including cwd and + * resolved credentials), the AI account and credential generation it selected, + * its environment, and the runner image it ran in. Keyed with the instance's + * signing secret, so the stored value never verifies a guessed credential + * offline. Null when the runner image has no immutable identity: such a pass + * is never reused. + */ +export function harnessVerificationDigest(input: { + adapterType: string; config: Record; account: unknown; environmentId: string | null; + runner: Record | null; +}): string | null { + if (!input.runner) return null; + const canonical = JSON.stringify({ v: 2, adapterType: input.adapterType, config: input.config, account: input.account ?? null, + environmentId: input.environmentId, runner: input.runner }, (_key, value: unknown) => + value && typeof value === "object" && !Array.isArray(value) && !(value instanceof Date) + ? Object.fromEntries(Object.entries(value as Record).sort(([a], [b]) => a.localeCompare(b))) : value); + return `v2:hmac-sha256:${createHmac("sha256", resolveDecisionSigningSecret()).update("harness-verification-v2:").update(canonical).digest("hex")}`; +} export function assertHarnessTestPassed(result: AdapterEnvironmentTestResult) { const verified = result.checks.some(check => check.level === "info" && @@ -76,10 +101,21 @@ export function agentHarnessVerificationService(db: Db, manager: PluginWorkerMan } if (binding) config = { ...config, env: stripAiAuthBindings(config.env) }; config = (await secrets.resolveAdapterConfigForRuntime(agent.companyId, config, context, { adapterType: agent.adapterType })).config; + let runner: Record | null = providerApi ? { kind: "provider_api" } : { kind: "local", version: serverVersion }; const environmentId = providerApi ? null : await tests.resolveAdapterTestEnvironmentId(agent.companyId, agent.defaultEnvironmentId, agent.adapterType === "paperclip_runner" && config.provider === "acpx" && config.acpxAgent === "copilot"); if (environmentId) { await tests.assertAdapterTestEnvironmentForCompany(agent.companyId, environmentId, agent.adapterType); const environment = await environments.getById(environmentId); + if (environment && environment.driver !== "local") { + const image = environment.driver === "sandbox" ? await sandboxBinding(db, environment, agent.companyId).catch(() => null) : null; + // Only an immutable image digest names the image a test ran in; a snapshot + // name may be re-pointed without the allocation pinning it, so it never counts. + const immutable = image && "image" in image; + runner = immutable ? { kind: "sandbox", environmentId, ...image } : null; + } else if (environment && (await instanceSettingsService(db).getExperimental()).enableManagedSandboxOnly === true) { + // The test runs in the managed sandbox instead, whose image is not named here. + runner = null; + } // Run admission refuses these agents before any lease, so verification must too. const preconfiguredHome = agent.adapterType === "paperclip_runner" && config.homeMode === "preconfigured"; if (agent.adapterType === "paperclip_runner" && environment && (preconfiguredHome || !binding) @@ -90,24 +126,72 @@ export function agentHarnessVerificationService(db: Db, manager: PluginWorkerMan { ...context, consumerType: "environment", consumerId: environmentId }); config = { ...config, env: { ...resolved.env, ...parseObject(config.env) } }; } - if (binding) await aiConnectionService(db).select({ + const selectAccount = () => aiConnectionService(db).select({ companyId: agent.companyId, userId: responsibleUserId, agentId: agent.id, adapterType: agent.adapterType, model: config.model, runnerProvider: config.provider, - acpxAgent: config.acpxAgent, binding, + acpxAgent: config.acpxAgent, binding: binding!, }); + const selected = binding ? await selectAccount() : null; + // The account and credential generation the binding selected, never the + // credential: the selected connection and grant, and the current version + // (and AI session epoch) of every secret its credential reads. Timestamps + // that change on every use are left out, so a pass stays reusable. + const connection = selected?.connection; + const credentialRefs = [...(connection?.credentialSecretRefs ?? []), ...(selected?.grant?.credentialSecretRefs ?? [])]; + const credentialSecretIds = [...new Set(credentialRefs.map(ref => ref.secretId))]; + const readCredentialVersions = async () => credentialSecretIds.length ? await db.select({ id: companySecrets.id, + latestVersion: companySecrets.latestVersion, aiSessionEpoch: companySecrets.aiSessionEpoch }) + .from(companySecrets).where(and(eq(companySecrets.companyId, agent.companyId), inArray(companySecrets.id, credentialSecretIds))) + .then(rows => rows.sort((a, b) => a.id.localeCompare(b.id))) : []; + const accountWith = (credentialVersions: Awaited>) => selected ? { ...selected.attribution, + credentialSecretRefs: credentialRefs, externalCredential: connection?.externalCredential ?? null, + credentialVersions } : null; + const credentialVersions = await readCredentialVersions(); + // A configuration that already passed in this runner image is verified: + // a restart or a return to verifying never boots another test sandbox. + // A local probe without a managed AI connection can use the host's own + // credentials (environment variables, a CLI login) that no saved + // configuration names, so such a pass is never reused. + if (runner?.kind === "local" && !binding) runner = null; + const digest = harnessVerificationDigest({ adapterType: agent.adapterType, config, account: accountWith(credentialVersions), environmentId, runner }); + if (digest && agent.harnessVerifiedDigest === digest) return "complete"; const target = providerApi ? { executionTarget: null, environmentName: null, fallbackChecks: [], release: async (_status?: "released" | "failed") => {} } : await tests.resolveAdapterTestExecutionContext({ agentId: agent.id, companyId: agent.companyId, adapterType: agent.adapterType, environmentId }); let status: "released" | "failed" = "failed"; + // A holder, so the account the probe callback records survives control-flow narrowing. + const probed: { account: { connectionId: string; grantId: string } | null } = { account: null }; try { if (target.fallbackChecks.length) throw Object.assign(new Error("The test environment is unavailable"), { code: "test_environment_unavailable" }); const input = { companyId: agent.companyId, adapterType: agent.adapterType, config, executionTarget: target.executionTarget, environmentName: target.environmentName }; const result = binding ? await withManagedAiProbe(db, { ...input, agentId: agent.id, responsibleUserId, binding }, - managed => tests.testManagedEnvironment(agent.adapterType, { ...input, config: managed.config }, binding!, managed, agent.id)) + managed => { + // The account this probe actually uses; the recorded digest must describe it. + probed.account = { connectionId: managed.attribution.connectionId, grantId: managed.attribution.grantId }; + return tests.testManagedEnvironment(agent.adapterType, { ...input, config: managed.config }, binding!, managed, agent.id); + }) : await requireServerAdapter(agent.adapterType).testEnvironment(input); assertHarnessTestPassed(result); status = "released"; + // A passing subscription probe can save the credential it refreshed, which + // advances the version but keeps the AI session epoch; a person's rotation + // bumps the epoch. Record the generation the probe left, unless any epoch + // moved: then the pass is not recorded and the next check tests again. + const verifiedVersions = await readCredentialVersions(); + const sameEpochs = verifiedVersions.length === credentialVersions.length && verifiedVersions.every((row, index) => + row.id === credentialVersions[index]!.id && row.aiSessionEpoch === credentialVersions[index]!.aiSessionEpoch); + // The managed probe selects the account again when it starts. Record the + // pass only for the account it actually tested. + const sameAccount = !selected || (probed.account?.connectionId === selected.attribution.connectionId + && probed.account?.grantId === selected.attribution.grantId); + const verifiedDigest = digest && sameEpochs && sameAccount + ? harnessVerificationDigest({ adapterType: agent.adapterType, config, account: accountWith(verifiedVersions), environmentId, runner }) + : null; + // Record the pass before the lifecycle step completes, so a restart in + // between reuses it instead of testing again. + if (verifiedDigest) await db.update(agents).set({ harnessVerifiedDigest: verifiedDigest, harnessVerifiedAt: new Date() }) + .where(and(eq(agents.id, agent.id), eq(agents.lifecycleVersion, snapshot.lifecycleVersion))); return "complete"; } finally { await target.release(status); } } diff --git a/server/src/services/agent-lifecycle.ts b/server/src/services/agent-lifecycle.ts index 53c9081e247..63a54291ef4 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/server/src/services/agent-lifecycle.ts @@ -1,4 +1,5 @@ import { notifyMuseContactVerification } from "../modules/external-agents/index.js"; +import { currentControllerProcessStamp } from "./controller-process-identity.js"; import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; @@ -50,6 +51,7 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li return plugin !== null && !selectedPluginIds?.includes(plugin.id); }, notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle), + leaseHolder: () => currentControllerProcessStamp(), beforeConnectionEligibilityRestore: async (tx, companyId, agentId) => { await notifyMuseContactVerification(tx, { companyId, agentId }); }, diff --git a/server/src/services/environment-probe.ts b/server/src/services/environment-probe.ts index 7f15b03656e..d8b8f07bf43 100644 --- a/server/src/services/environment-probe.ts +++ b/server/src/services/environment-probe.ts @@ -34,7 +34,8 @@ export function parseSandboxPlatformFacts(output: string): PlatformFacts { libc: libc === "glibc" || libc === "musl" ? libc : "unknown", ...(libc === "glibc" ? { glibcVersion: version?.trim() ?? "unknown" } : {}) }; } -async function sandboxBinding(db: Db, environment: Environment, companyId: string) { +/** The immutable runner image or snapshot identity of a sandbox environment, or null. */ +export async function sandboxBinding(db: Db, environment: Environment, companyId: string) { const selectedTemplate: { value: SandboxTemplateIdentity } = { value: null }; const parsed = await resolveEnvironmentDriverConfigForRuntime(db, companyId, environment, { applyCustomImageTemplate: true, readOnlyCustomImageTemplate: true, diff --git a/server/src/services/recovery/agent-lifecycle-lease-settlement.ts b/server/src/services/recovery/agent-lifecycle-lease-settlement.ts new file mode 100644 index 00000000000..5a83dc6f4c5 --- /dev/null +++ b/server/src/services/recovery/agent-lifecycle-lease-settlement.ts @@ -0,0 +1,43 @@ +import type { Db } from "@paperclipai/db"; +import { releaseAgentLifecycleLeaseOfHolderInTransaction } from "../../modules/agent-lifecycle/index.js"; +import { persistActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; +import type { OwnerProof } from "../owner-proof.js"; +import type { InProgressRecord, SettleOutcome } from "./ownership-registry.js"; + +/** + * Settle of the `agent_lifecycle_lease` ownership kind. + * + * A lifecycle step (for example harness verification) holds a 120 s lease that + * its process renews. A restart killed the holder, and the next process waited + * out the lease before it could claim the step again, while the agent's runs + * stayed queued. The shared reconciler calls this only after it proved the + * stamped holder gone. The lifecycle module releases exactly that holder's + * lease (compare-and-swap), so the lifecycle worker claims the step at once. + */ +export async function settleAgentLifecycleLease(db: Db, record: InProgressRecord, + context: { now: Date; proof: OwnerProof }): Promise { + if (record.owner.kind !== "process") return { settled: false, reason: "owner_not_provable" }; + const holder = { bootId: record.owner.stamp.bootId, pid: record.owner.stamp.pid }; + let publication: ActivityPublication | undefined; + const outcome = await db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const released = await releaseAgentLifecycleLeaseOfHolderInTransaction(tx, { agentId: record.recordId, holder }, + () => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle)); + if (!released.released) return { settled: false, reason: released.reason }; + const { agent } = released; + const activity = await persistActivity(tx, { + companyId: agent.companyId, actorType: "system", actorId: "ownership-reconciler", + action: "agent.lifecycle_lease_reconciled", entityType: "agent", entityId: agent.id, agentId: agent.id, + details: { + schema: "paperclip.agent-lifecycle-lease-settlement.v1", phase: agent.lifecycleState, + operationId: agent.lifecycleOperation?.id ?? null, proof: context.proof.reason, previousLeaseUntil: released.leaseUntil, + reconciledAt: context.now.toISOString(), + }, + }); + publication = activity.publication; + return { settled: true, reason: context.proof.reason }; + }); + if (outcome.settled && publication) publishActivity(publication); + return outcome; +} diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 0dfdf3dab45..04f2e4aa6e5 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -3,6 +3,7 @@ import path from "node:path"; import { and, asc, eq, inArray, isNotNull, isNull, like, notInArray, notLike, or, sql, type SQL } from "drizzle-orm"; import { agentWakeupRequests, + agents, environmentLeases, executionWorkspaceRestorePoints, executionWorkspaceRuntimeLeases, @@ -40,6 +41,7 @@ import { copybackOwnerStamp } from "../native-runtime/native-workspace-owner-sto import { legacyShutdownSettlementOwner, settleLegacyShutdownSettlementFence } from "../legacy-shutdown-workspace-settlement.js"; import { nativeControllerOwnerStamp, runProcessOwnerStamp } from "../native-runtime/native-owner-stamps.js"; import { settleIssueLock } from "./issue-lock-settlement.js"; +import { settleAgentLifecycleLease } from "./agent-lifecycle-lease-settlement.js"; import { legacyControllerOwnerStamp, settleLegacyControllerLease } from "../legacy-controller-lease.js"; import { nativeLeaseHolderStamp, @@ -549,6 +551,31 @@ function runtimeLeaseOwner(row: typeof executionWorkspaceRuntimeLeases.$inferSel // ---------------------------------------------------------------------------- // The registry +// ---------------------------------------------------------------------------- +// agents.lifecycle_operation + +/** A live lifecycle lease (for example harness verification) and its holder. */ +const agentLifecycleLeaseHeld = and( + sql`${agents.lifecycleOperation} ? 'leaseOwner'`, sql`${agents.lifecycleOperation} ? 'leaseHolder'`, + sql`(${agents.lifecycleOperation}->>'leaseUntil')::timestamptz > now()`); + +const agentLifecycleLeaseKind: OwnableKind = { + // Shared path: a restart that kills the holder must not leave the agent's + // lifecycle step, and its queued runs, waiting out the lease. + id: "agent_lifecycle_lease", title: "Agent lifecycle step lease (harness verification and setup)", obligation: "ownership", + storage: "database", invariantChecks: [1, 2], settle: settleAgentLifecycleLease, + listInProgress: async (db, limit, offset) => { + const rows = await db.select({ id: agents.id, companyId: agents.companyId, operation: agents.lifecycleOperation, + updatedAt: agents.updatedAt }).from(agents).where(agentLifecycleLeaseHeld) + .orderBy(asc(agents.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(row => ({ + recordId: row.id, companyId: row.companyId, runId: null, issueId: null, + owner: controllerStamp(row.operation?.leaseHolder, "agent_lifecycle_lease"), updatedAt: row.updatedAt, + })) }; + }, +}; + export const OWNABLE_KINDS: readonly OwnableKind[] = [ // Native runtime and Runner nrfKind("native_executor_claim", "Executor execution claim (fresh run)", "ownership", @@ -979,6 +1006,7 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ () => ({ kind: "operator", ownerType: "run_dispatch" }), [4]), runKind("scheduled_retry_run", "Scheduled retry run", "ownership", eq(heartbeatRuns.status, "scheduled_retry"), () => ({ kind: "operator", ownerType: "retry_schedule" }), [4]), + agentLifecycleLeaseKind, ]; /** Resolve the current holder of fences that live on another row (the diff --git a/tests/restart-chaos/seed.ts b/tests/restart-chaos/seed.ts index 9c0b27044e7..8c1f53b1ae8 100644 --- a/tests/restart-chaos/seed.ts +++ b/tests/restart-chaos/seed.ts @@ -17,7 +17,8 @@ export interface ChaosLab { environmentKind: EnvironmentKind; projectId: string; fixtureRepositoryPath: string; - agents: Record; + /** `body` is the creation request, so a driver can hire an identical agent later. */ + agents: Record }>; } /** Create the `chaos-sandbox` git repository: python CLI, one failing test, bench.sh. */ @@ -179,7 +180,7 @@ export async function seedChaosLab(input: { workspacePath: fixtureRepositoryPath, secretRefs, executionId: input.waveId }); const name = `Chaos ${adapter}`; const runtimeConfig = (payload.runtimeConfig ?? {}) as Record; - const created = await api.post<{ id: string; name: string }>(`/api/companies/${company.id}/agents`, { + const body: Record = { ...payload, ...(input.adapterConfigOverrides ? { adapterConfig: { ...((payload.adapterConfig ?? {}) as Record), ...input.adapterConfigOverrides } } : {}), @@ -187,8 +188,9 @@ export async function seedChaosLab(input: { title: "Restart chaos worker", runtimeConfig: { ...runtimeConfig, heartbeat: { ...((runtimeConfig.heartbeat ?? {}) as Record), enabled: false, maxConcurrentRuns: Math.max(1, input.concurrency) } }, - }); - agents[adapter] = { id: created.id, name }; + }; + const created = await api.post<{ id: string; name: string }>(`/api/companies/${company.id}/agents`, body); + agents[adapter] = { id: created.id, name, body }; timeline.add("seed_agent", { adapter, agentId: created.id, profile: profile.id, model: profile.model, environmentId }); } diff --git a/tests/restart-chaos/ttft-baseline-stats.ts b/tests/restart-chaos/ttft-baseline-stats.ts index a0b490bcb22..9f90c0ecad6 100644 --- a/tests/restart-chaos/ttft-baseline-stats.ts +++ b/tests/restart-chaos/ttft-baseline-stats.ts @@ -16,7 +16,7 @@ export interface BaselineRun { adapter: string; environment: string; /** Driver intent: a new task, a follow-up comment on the same task, or a run crossed by a restart. */ - phase: "cold" | "warm" | "restart" | "midturn"; + phase: "cold" | "warm" | "restart" | "midturn" | "verify"; events: BaselineEvent[]; } @@ -149,7 +149,8 @@ export interface BaselineRow { adapter: string; environment: string; warmth: "co export function baselineRows(runs: readonly BaselineRun[]): BaselineRow[] { const groups = new Map(); for (const run of runs) { - if (run.phase === "restart") continue; + // Restart samples and new-agent verification samples are reported on their own. + if (run.phase === "restart" || run.phase === "verify") continue; const warmth = runWarmth(run); const key = `${run.adapter}|${run.environment}|${warmth}`; const group = groups.get(key) ?? { adapter: run.adapter, environment: run.environment, warmth, metrics: [] }; diff --git a/tests/restart-chaos/ttft-baseline.ts b/tests/restart-chaos/ttft-baseline.ts index 974226ddc8f..291a12ffd19 100644 --- a/tests/restart-chaos/ttft-baseline.ts +++ b/tests/restart-chaos/ttft-baseline.ts @@ -24,6 +24,7 @@ import { CREDENTIAL_FOR_ADAPTER } from "./config.js"; import { ACTIVE_RUN_STATUSES } from "./grader.js"; import { chaosLifecycleOverrides, seedChaosLab, type ChaosLab } from "./seed.js"; import { ChaosSupervisor } from "./supervisor.js"; +import { createDb } from "../../packages/db/src/index.js"; import { Timeline } from "./timeline.js"; import type { AdapterId, FaultKind } from "./types.js"; import { @@ -190,9 +191,16 @@ async function readyAdapters(api: ChaosApi, lab: ChaosLab, adapters: AdapterId[] function report(runs: BaselineRun[], faults: BaselineFault[]): string { const ms = (value: number | null) => value === null ? "—" : `${(value / 1000).toFixed(2)} s`; const lines = ["| Adapter | Env | Start | n | TTFT p50 | TTFT p95 | Wait behind live turn p50 / p95 (n) | TTFT from run created p50 / p95 | Boot p50 | Boot p95 | TTFT anchors |", "|---|---|---|---|---|---|---|---|---|---|---|"]; - for (const row of baselineRows(runs)) { - lines.push(`| ${row.adapter} | ${row.environment} | ${row.warmth} | ${row.ttft.n}/${row.boot.n} | ${ms(row.ttft.p50)} | ${ms(row.ttft.p95)} | ${ms(row.inputWait.p50)} / ${ms(row.inputWait.p95)} (${row.inputWait.n}) | ${ms(row.ttftAfterRunCreated.p50)} / ${ms(row.ttftAfterRunCreated.p95)} | ${ms(row.boot.p50)} | ${ms(row.boot.p95)} | ${Object.entries(row.anchors).map(([k, v]) => `${k}×${v}`).join(", ") || "—"} |`); + const rowLine = (row: ReturnType[number]) => `| ${row.adapter} | ${row.environment} | ${row.warmth} | ${row.ttft.n}/${row.boot.n} | ${ms(row.ttft.p50)} | ${ms(row.ttft.p95)} | ${ms(row.inputWait.p50)} / ${ms(row.inputWait.p95)} (${row.inputWait.n}) | ${ms(row.ttftAfterRunCreated.p50)} / ${ms(row.ttftAfterRunCreated.p95)} | ${ms(row.boot.p50)} | ${ms(row.boot.p95)} | ${Object.entries(row.anchors).map(([k, v]) => `${k}×${v}`).join(", ") || "—"} |`; + for (const row of baselineRows(runs)) lines.push(rowLine(row)); + // New agents wait for harness verification before their run is created: + // their own rows, so that wait never blends into ordinary cold starts. + const verifyRuns = runs.filter(run => run.phase === "verify").map(run => ({ ...run, phase: "cold" as const })); + if (verifyRuns.length) { + lines.push("", "New agent behind harness verification (TTFT counts from run creation; the verification wait precedes it):", "", lines[0]!, lines[1]!); + for (const row of baselineRows(verifyRuns)) lines.push(rowLine(row)); } + lines.push("", "| Pre-turn step | n | p50 | p95 | Adapters |", "|---|---|---|---|---|"); for (const step of rankPreTurnSteps(runs).slice(0, 25)) lines.push(`| \`${step.step}\` | ${step.n} | ${ms(step.p50)} | ${ms(step.p95)} | ${step.adapters.join(", ")} |`); const samples = recoverySamples(runs, faults); @@ -226,6 +234,8 @@ async function main() { "midturn-mode": { type: "string", default: "queue" }, // Fault the control plane right after a steered comment is acknowledged (K1, K2 or K6). "midturn-fault": { type: "string" }, + // verify phase: fault the control plane while a newly hired agent runs harness verification. + "verify-fault": { type: "string" }, "idle-ms": { type: "string", default: "600000" }, } }); if (values.report) { @@ -244,6 +254,8 @@ async function main() { const midturnFault = values["midturn-fault"] as FaultKind | undefined; if (midturnFault !== undefined && !["K1", "K2", "K6"].includes(midturnFault)) throw new Error("--midturn-fault must be K1, K2 or K6"); if (midturnFault && midturnMode !== "steer") throw new Error("--midturn-fault applies to --midturn-mode steer"); + const verifyFault = values["verify-fault"] as FaultKind | undefined; + if (verifyFault !== undefined && !["K1", "K2", "K6"].includes(verifyFault)) throw new Error("--verify-fault must be K1, K2 or K6"); const lifecycle = values.lifecycle; if (lifecycle !== undefined && lifecycle !== "per_turn" && lifecycle !== "warm" && lifecycle !== "default") { throw new Error("--lifecycle must be per_turn, warm or default"); @@ -319,6 +331,58 @@ async function main() { } })); + // New agent: hire an agent identical to the seeded one and assign it a task + // at once, so the task waits for harness verification. Optionally restart + // the control plane while it verifies. TTFT is comment-to-first-token. + if (phases.has("verify")) for (const adapter of adapters) { + const source = lab.agents[adapter]; + if (!source.body) continue; + for (let index = 0; index < samples; index += 1) { + const nonce = `${adapter}-v${index}-${Date.now().toString(36)}`; + const hired = await api.post(`/api/companies/${lab.companyId}/agents`, { ...source.body, name: `${source.name} verify ${nonce}` }); + const agentId = String(hired.id); + const issue = await api.post(`/api/companies/${lab.companyId}/issues`, { + title: `TTFT verify ${nonce}`, description: coldTask(nonce), status: "todo", priority: "medium", assigneeAgentId: agentId }); + const issueId = String(issue.id); + tasks.push({ issueId, adapter, phase: "verify" }); + let fault: Awaited> | null = null; + if (verifyFault) { + const deadline = Date.now() + 60_000; + let state = "unknown"; + while (Date.now() < deadline && (state = String((await api.get(`/api/agents/${agentId}`).catch(() => null))?.lifecycleState)) !== "verifying" + && state !== "ready") await delay(200); + // Fault only while a verifier holds the lifecycle lease, read from the + // instance database; otherwise the sample is skipped, not faulted. + const config = JSON.parse(await readFile(supervisor.paths.configPath, "utf8")) as { database?: { connectionString?: string } }; + const db = createDb(config.database!.connectionString!); + let held = false; + try { + while (state === "verifying" && Date.now() < deadline && !held) { + const rows = await db.$client`select 1 from agents where id = ${agentId} + and lifecycle_state = 'verifying' and lifecycle_operation ? 'leaseOwner'`; + held = rows.length > 0; + if (!held) await delay(100); + } + } finally { await db.$client.end({ timeout: 5 }).catch(() => undefined); } + timeline.add("baseline_verify_fault_point", { adapter, agentId, state, leaseHeld: held }); + if (!held) timeline.add("baseline_verify_sample_skipped", { adapter, agentId, reason: "no_held_verification_lease" }); + if (held) { + fault = await supervisor.fault(verifyFault, 0, `verify-${index}`); + } + } + const settled = await waitForRest(api, supervisor, issueId, 1); + if (settled.timedOut) { + timedOutIssues.add(issueId); + timeline.add("baseline_sample_timeout", { adapter, issueId, phase: "verify" }); + continue; + } + // The task's run is created only after verification, so name it once it + // exists: the recovery table then measures fault ready to first event. + if (fault) faults.push({ kind: verifyFault!, readyAt: fault.readyAt ?? Date.now(), runIds: settled.runs.map(runIdOf) }); + timeline.add("baseline_verify_done", { adapter, agentId, issueId, faulted: Boolean(fault), runs: settled.runs.length }); + } + } + // Mid-turn follow-up: a comment posted while the agent is inside a long // tool call. Records when the comment landed and every run of the task, // so the wait splits into "rest of the turn" and "hand-off".