diff --git a/packages/db/src/client.test.ts b/packages/db/src/client.test.ts index 491cd31db85..96ffb1d3f9d 100644 --- a/packages/db/src/client.test.ts +++ b/packages/db/src/client.test.ts @@ -302,6 +302,9 @@ describeEmbeddedPostgres("merged Muse migration histories", () => { const beforeHistory = baseline === "fresh" ? [] : await historyRows(sql); const retainedRows = baseline === "fresh" ? null : await seedRetainedRows(sql, baseline === "muse0348"); const beforeRows = await retainedRows?.(); + const expectedRetainedRows = beforeRows + ? { ...beforeRows, run: { ...beforeRows.run, legacy_process_settlement: null } } + : beforeRows; const state = await inspectMigrations(connectionString); expect(state.status).toBe("needsMigrations"); const laterMigrations = journal.entries.filter(entry => entry.idx > 353).map(entry => `${entry.tag}.sql`); @@ -327,7 +330,7 @@ describeEmbeddedPostgres("merged Muse migration histories", () => { const currentHash = await migrationHash(entry.file); expect(afterHistory.filter(row => row.hash === currentHash)).toHaveLength(1); } - if (retainedRows) expect(retainedColumns(await retainedRows(), beforeRows!)).toEqual(beforeRows); + if (retainedRows) expect(retainedColumns(await retainedRows(), expectedRetainedRows!)).toEqual(expectedRetainedRows); const indexes = await sql` SELECT c.relname, i.indisvalid, i.indisready, pg_get_indexdef(c.oid) AS definition FROM pg_index i JOIN pg_class c ON c.oid = i.indexrelid JOIN pg_namespace n ON n.oid = c.relnamespace @@ -344,7 +347,7 @@ describeEmbeddedPostgres("merged Muse migration histories", () => { expect((await inspectMigrations(connectionString)).status).toBe("upToDate"); expect(await historyRows(sql)).toEqual(afterHistory); expect(await schemaShape(sql)).toEqual(finalShape); - if (retainedRows) expect(retainedColumns(await retainedRows(), beforeRows!)).toEqual(beforeRows); + if (retainedRows) expect(retainedColumns(await retainedRows(), expectedRetainedRows!)).toEqual(expectedRetainedRows); } finally { await Promise.all([sql.end(), reference.end()]); } diff --git a/packages/db/src/migrations/0439_cultured_impossible_man.sql b/packages/db/src/migrations/0439_cultured_impossible_man.sql new file mode 100644 index 00000000000..5ace6155e80 --- /dev/null +++ b/packages/db/src/migrations/0439_cultured_impossible_man.sql @@ -0,0 +1 @@ +ALTER TABLE "heartbeat_runs" ADD COLUMN "legacy_process_settlement" jsonb; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0434_snapshot.json b/packages/db/src/migrations/meta/0439_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0434_snapshot.json rename to packages/db/src/migrations/meta/0439_snapshot.json index 34fa20956fd..58aa1726b71 100644 --- a/packages/db/src/migrations/meta/0434_snapshot.json +++ b/packages/db/src/migrations/meta/0439_snapshot.json @@ -1,6 +1,6 @@ { - "id": "5b40e9ac-12aa-4757-aac7-36218c7500a8", - "prevId": "10b23b32-3586-4a30-a993-618e7d3ca893", + "id": "d290d707-7d29-43a0-a697-06eecbdb743d", + "prevId": "69e6941f-890f-43d8-af7c-2e3e24ebfcd5", "version": "7", "dialect": "postgresql", "tables": { @@ -4058,6 +4058,28 @@ "method": "btree", "with": {} }, + "agent_wakeup_requests_tree_hold_resume_uq": { + "name": "agent_wakeup_requests_tree_hold_resume_uq", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "idempotency_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"agent_wakeup_requests\".\"idempotency_key\" LIKE 'tree-hold-resume:%' AND \"agent_wakeup_requests\".\"status\" <> 'skipped'", + "concurrently": true, + "method": "btree", + "with": {} + }, "agent_wakeup_requests_chat_completion_uq": { "name": "agent_wakeup_requests_chat_completion_uq", "columns": [ @@ -28962,6 +28984,22 @@ "method": "btree", "with": {} }, + "environments_managed_plugin_environment_idx": { + "name": "environments_managed_plugin_environment_idx", + "columns": [ + { + "expression": "(\"metadata\" ->> 'managedPluginEnvironment')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "jsonb_typeof(\"environments\".\"metadata\" -> 'managedPluginEnvironment') = 'string'", + "concurrently": false, + "method": "btree", + "with": {} + }, "environments_name_idx": { "name": "environments_name_idx", "columns": [ @@ -34842,6 +34880,12 @@ "primaryKey": false, "notNull": false }, + "legacy_process_settlement": { + "name": "legacy_process_settlement", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "last_output_at": { "name": "last_output_at", "type": "timestamp with time zone", @@ -35053,6 +35097,28 @@ "method": "btree", "with": {} }, + "heartbeat_runs_tree_hold_suppression_idx": { + "name": "heartbeat_runs_tree_hold_suppression_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"result_json\"->'treeHoldSuppression'->>'holdId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"status\" = 'cancelled' and \"heartbeat_runs\".\"result_json\" ? 'treeHoldSuppression'", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_cost_accounting_pending_idx": { "name": "heartbeat_runs_cost_accounting_pending_idx", "columns": [ @@ -44397,6 +44463,256 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.native_provider_artifact_references": { + "name": "native_provider_artifact_references", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "artifact_id": { + "name": "artifact_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "run_id": { + "name": "run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "backup_id": { + "name": "backup_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "native_provider_artifact_references_artifact_idx": { + "name": "native_provider_artifact_references_artifact_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "artifact_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_provider_artifact_references_run_idx": { + "name": "native_provider_artifact_references_run_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "artifact_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"native_provider_artifact_references\".\"backup_id\" is null", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_provider_artifact_references_backup_idx": { + "name": "native_provider_artifact_references_backup_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "artifact_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "backup_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"native_provider_artifact_references\".\"backup_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "native_provider_artifact_references_artifact_id_native_provider_artifacts_id_fk": { + "name": "native_provider_artifact_references_artifact_id_native_provider_artifacts_id_fk", + "tableFrom": "native_provider_artifact_references", + "tableTo": "native_provider_artifacts", + "columnsFrom": [ + "artifact_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "restrict", + "onUpdate": "no action" + }, + "native_provider_artifact_references_backup_id_native_harness_backups_id_fk": { + "name": "native_provider_artifact_references_backup_id_native_harness_backups_id_fk", + "tableFrom": "native_provider_artifact_references", + "tableTo": "native_harness_backups", + "columnsFrom": [ + "backup_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.native_provider_artifacts": { + "name": "native_provider_artifacts", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "closure_digest": { + "name": "closure_digest", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "upload_token": { + "name": "upload_token", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "descriptor": { + "name": "descriptor", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "reference": { + "name": "reference", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "native_provider_artifacts_company_digest_idx": { + "name": "native_provider_artifacts_company_digest_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "closure_digest", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"native_provider_artifacts\".\"status\" = 'ready'", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.native_run_finalizations": { "name": "native_run_finalizations", "schema": "", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 57b0de3aa33..f22ae4ca85f 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3060,6 +3060,13 @@ "when": 1793996706927, "tag": "0438_mighty_outlaw_kid", "breakpoints": true + }, + { + "idx": 439, + "version": "7", + "when": 1794024254666, + "tag": "0439_cultured_impossible_man", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index c989342ac96..1d2e25466f0 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -96,6 +96,7 @@ export const heartbeatRuns = pgTable( processPid: integer("process_pid"), processGroupId: integer("process_group_id"), processStartedAt: timestamp("process_started_at", { withTimezone: true }), + legacyProcessSettlement: jsonb("legacy_process_settlement").$type>(), lastOutputAt: timestamp("last_output_at", { withTimezone: true }), lastOutputSeq: integer("last_output_seq").notNull().default(0), lastOutputStream: text("last_output_stream"), diff --git a/server/src/__tests__/helpers/native-restart-fixture.ts b/server/src/__tests__/helpers/native-restart-fixture.ts index 2607aba738b..2b6ced75946 100644 --- a/server/src/__tests__/helpers/native-restart-fixture.ts +++ b/server/src/__tests__/helpers/native-restart-fixture.ts @@ -1,9 +1,9 @@ +import { createNativeRestartRuntimeRoot } from "./native-restart-workspace.js"; import { startRecoveryPostgres } from "./recovery-postgres-fixture.js"; import { randomUUID } from "node:crypto"; import { existsSync } from "node:fs"; -import { mkdtemp, rm } from "node:fs/promises"; +import { rm } from "node:fs/promises"; import { createServer, type Server } from "node:http"; -import { tmpdir } from "node:os"; import { resolve } from "node:path"; import { eq } from "drizzle-orm"; import { agents, companies, closeRegisteredClients, createDb, heartbeatRuns, issues, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; @@ -90,7 +90,7 @@ export function nativeRestartFixture(options: { dedicatedDatabaseUrl?: string } beforeAll(async () => { temporary = await startRecoveryPostgres("native-runner-restart-recovery-", options); - runtimeRoot = await mkdtemp(resolve(tmpdir(), "native-restart-runtime-")); + runtimeRoot = await createNativeRestartRuntimeRoot(); isolatedHome = createIsolatedTestHome(); paperclipHome = isolatedHome.home; process.env.PAPERCLIP_HOME = paperclipHome; diff --git a/server/src/__tests__/helpers/native-restart-workspace.ts b/server/src/__tests__/helpers/native-restart-workspace.ts new file mode 100644 index 00000000000..bb3d0acfe32 --- /dev/null +++ b/server/src/__tests__/helpers/native-restart-workspace.ts @@ -0,0 +1,7 @@ +import { mkdtemp } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; + +export function createNativeRestartRuntimeRoot(): Promise { + return mkdtemp(resolve(tmpdir(), "native-restart-runtime-")); +} diff --git a/server/src/__tests__/helpers/recovery-deadline-bootstrap.ts b/server/src/__tests__/helpers/recovery-deadline-bootstrap.ts new file mode 100644 index 00000000000..ed2a9dc782b --- /dev/null +++ b/server/src/__tests__/helpers/recovery-deadline-bootstrap.ts @@ -0,0 +1,12 @@ +import type { AdapterExecutionResult } from "../../adapters/types.js"; + +export const controlledCompanyBootstrapMessage = "Controlled review bootstrap unavailable"; +export const controlledCompanyBootstrapScript = `console.error(${JSON.stringify(controlledCompanyBootstrapMessage)});process.exit(75)`; + +export function controlledCompanyBootstrapResult(config: Record, result: AdapterExecutionResult): AdapterExecutionResult { + const args = config.args; + if (config.command !== process.execPath || !Array.isArray(args) || args.length !== 2 || args[0] !== "-e" || + args[1] !== controlledCompanyBootstrapScript || result.exitCode !== 75 || result.signal !== null || result.timedOut || + result.resultJson?.stdout !== "" || result.resultJson?.stderr !== controlledCompanyBootstrapMessage + "\n") return result; + return { ...result, executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }; +} diff --git a/server/src/__tests__/helpers/recovery-deadline-ordinary-bootstrap.ts b/server/src/__tests__/helpers/recovery-deadline-ordinary-bootstrap.ts new file mode 100644 index 00000000000..0966060127e --- /dev/null +++ b/server/src/__tests__/helpers/recovery-deadline-ordinary-bootstrap.ts @@ -0,0 +1,19 @@ +import { classifyCodexTerminalSessionFailure } from "@paperclipai/adapter-codex-local/server"; +import type { AdapterExecutionResult } from "../../adapters/types.js"; + +export const controlledOrdinaryBootstrapFailure = { + category: "service", title: "Selected model is at capacity", details: "Controlled bootstrap refusal before any domain action.", +} as const; +export const controlledOrdinaryBootstrapProtocol = JSON.stringify(controlledOrdinaryBootstrapFailure); +export const controlledOrdinaryBootstrapScript = `console.error(${JSON.stringify(controlledOrdinaryBootstrapProtocol)});process.exit(75)`; + +export function controlledOrdinaryBootstrapResult(config: Record, result: AdapterExecutionResult): AdapterExecutionResult { + const args = config.args; + if (config.command !== process.execPath || !Array.isArray(args) || args.length !== 2 || args[0] !== "-e" || + args[1] !== controlledOrdinaryBootstrapScript || result.exitCode !== 75 || result.signal !== null || result.timedOut || + result.resultJson?.stdout !== "" || result.resultJson?.stderr !== controlledOrdinaryBootstrapProtocol + "\n") return result; + const failure = JSON.parse(result.resultJson.stderr) as typeof controlledOrdinaryBootstrapFailure; + const classified = classifyCodexTerminalSessionFailure(failure, new Date()); + if (classified?.errorFamily !== "transient_upstream") return result; + return { ...result, ...classified, executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }; +} diff --git a/server/src/__tests__/helpers/recovery-preparation-failure.ts b/server/src/__tests__/helpers/recovery-preparation-failure.ts new file mode 100644 index 00000000000..6e680f7fa1a --- /dev/null +++ b/server/src/__tests__/helpers/recovery-preparation-failure.ts @@ -0,0 +1,85 @@ +import type { heartbeatService } from "../../services/heartbeat.js"; +import type { DeadlineReconciliationSnapshot } from "../../services/deadline-reconciliation.js"; +import { readIdleRetrySchedule } from "../../services/heartbeat/idle-retry-schedule.js"; +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { agents, environments, heartbeatRuns, plugins, type Db } from "@paperclipai/db"; +import { expect, vi } from "vitest"; +import { environmentRuntimeService } from "../../services/environment-runtime.js"; +import type { PluginWorkerManager } from "../../services/plugin-worker-manager.js"; + +function barrier() { + let resolve!: () => void; + const promise = new Promise(done => { resolve = done; }); + return { promise, resolve }; +} + +export async function preparationFailureFixture(db: Db, agentId: string) { + const pluginId = randomUUID(), environmentId = randomUUID(), key = `recovery-preparation-${pluginId}`; + const call = vi.fn(); + const manager = { isRunning: () => false, call } as unknown as PluginWorkerManager; + const runtime = environmentRuntimeService(db, { pluginWorkerManager: manager, pluginWorkerReadyTimeoutMs: 0 }); + await db.insert(plugins).values({ id: pluginId, pluginKey: key, packageName: `@fixture/${key}`, version: "1.0.0", + apiVersion: 1, categories: ["automation"], status: "ready", installOrder: 1, + manifestJson: { id: key, apiVersion: 1, version: "1.0.0", displayName: "Controlled unavailable preparation worker", + description: "Pre-dispatch worker refusal fixture", author: "Paperclip", categories: ["automation"], + capabilities: ["environment.drivers.register"], entrypoints: { worker: "dist/worker.js" }, + environmentDrivers: [{ driverKey: key, kind: "sandbox_provider", displayName: "Controlled unavailable worker", + configSchema: { type: "object" } }] } }); + await db.insert(environments).values({ id: environmentId, name: key, driver: "sandbox", status: "active", + config: { provider: key, image: "fixture:unavailable", reuseLease: false } }); + await db.update(agents).set({ defaultEnvironmentId: environmentId }).where(eq(agents.id, agentId)); + const preparing = barrier(), releasePreparation = barrier(), cleaning = barrier(), releaseCleanup = barrier(); + const acquire = runtime.acquireRunLease.bind(runtime), release = runtime.releaseRunLeases.bind(runtime); + let first = true, firstCleanup = true, activeRunId: string | undefined; + const phases = { preparing: false, cleaning: false }; + const acquireSpy = vi.spyOn(runtime, "acquireRunLease").mockImplementation(async input => { + if (first && input.agentId === agentId && input.heartbeatRunId) { first = false; activeRunId = input.heartbeatRunId; phases.preparing = true; preparing.resolve(); await releasePreparation.promise; } + return acquire(input); + }); + const releaseSpy = vi.spyOn(runtime, "releaseRunLeases").mockImplementation(async (...args) => { + if (firstCleanup && activeRunId && args[0] === activeRunId) { firstCleanup = false; phases.cleaning = true; cleaning.resolve(); await releaseCleanup.promise; } + return release(...args); + }); + return { runtime, pluginId, environmentId, call, phases, preparing, releasePreparation, cleaning, releaseCleanup, + async useLocalEnvironment() { await db.update(agents).set({ defaultEnvironmentId: null }).where(eq(agents.id, agentId)); }, + unblock() { releasePreparation.resolve(); releaseCleanup.resolve(); }, + async dispose() { + releasePreparation.resolve(); releaseCleanup.resolve(); acquireSpy.mockRestore(); releaseSpy.mockRestore(); + await db.update(agents).set({ defaultEnvironmentId: null }).where(eq(agents.id, agentId)); + await db.delete(environments).where(eq(environments.id, environmentId)); + await db.delete(plugins).where(eq(plugins.id, pluginId)); + }, + }; +} + +export async function drivePreparationFailure(input: { + db: Db; heartbeat: ReturnType; control: Awaited>; + entered: unknown; persisted: unknown; companyId: string; agentId: string; issueId: string; + run: (id: string) => Promise; owner: () => DeadlineReconciliationSnapshot; +}) { + const { db, heartbeat, control, entered, persisted, companyId, agentId, issueId, run, owner } = input; + let wakeError: unknown; + const waking = heartbeat.wakeup(agentId, { source: "on_demand", reason: "manual", requestedByActorType: "system", + contextSnapshot: { issueId: issueId, taskId: issueId } }); + void waking.catch(error => { wakeError = error; }); + await vi.waitFor(() => { if (wakeError) throw wakeError; expect(control.phases.preparing).toBe(true); }, { timeout: 15_000 }); + const [claimed] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "running"))); + expect(claimed).toMatchObject({ runtimeMode: "legacy", executionStage: "preparing", processPid: null, + processGroupId: null, processStartedAt: null, legacyProcessSettlement: null }); + expect(claimed.controllerBootId).toEqual(expect.any(String)); expect(claimed.controllerLeaseExpiresAt).toBeInstanceOf(Date); + expect(entered).not.toHaveBeenCalled(); expect(persisted).not.toHaveBeenCalled(); + expect(heartbeat.getTaskDrainStatus().activeRuns).toBeGreaterThan(0); + control.releasePreparation.resolve(); + await vi.waitFor(() => { if (wakeError) throw wakeError; expect(control.phases.cleaning).toBe(true); }, { timeout: 15_000 }); + const failed = await run(claimed.id); + expect(failed).toMatchObject({ status: "failed", errorCode: "plugin_worker_unavailable", executionStage: "preparing", + processPid: null, processGroupId: null, processStartedAt: null, legacyProcessSettlement: null }); + expect(control.call).not.toHaveBeenCalled(); expect(entered).not.toHaveBeenCalled(); expect(persisted).not.toHaveBeenCalled(); + expect(failed.controllerLeaseExpiresAt).toBeInstanceOf(Date); + expect((await readIdleRetrySchedule(db, Date.now(), owner)).blocked).toBe(true); + control.releaseCleanup.resolve(); + await heartbeat.drainActiveRunExecutions(); + const admitted = await waking; expect(admitted?.id).toBe(claimed.id); + return run(claimed.id); +} diff --git a/server/src/__tests__/recovery-deadline-acceptance.test.ts b/server/src/__tests__/recovery-deadline-acceptance.test.ts index 9a6a6e8d763..741b0f13e51 100644 --- a/server/src/__tests__/recovery-deadline-acceptance.test.ts +++ b/server/src/__tests__/recovery-deadline-acceptance.test.ts @@ -1,3 +1,11 @@ +import { preparationFailureFixture, drivePreparationFailure } from "./helpers/recovery-preparation-failure.js"; +import { controlledOrdinaryBootstrapProtocol, controlledOrdinaryBootstrapResult, controlledOrdinaryBootstrapScript } from "./helpers/recovery-deadline-ordinary-bootstrap.js"; +import { controlledCompanyBootstrapResult, controlledCompanyBootstrapScript } from "./helpers/recovery-deadline-bootstrap.js"; +import * as legacySettlement from "../services/heartbeat/legacy-process-settlement.js"; +import express, { type Request } from "express"; +import request from "supertest"; +import { issueRoutes } from "../routes/issues.js"; +import { errorHandler } from "../middleware/index.js"; import { registerBudgetEnforcementWork } from "../services/budget-enforcement-worker.js"; import { hasPendingBudgetEvaluation, queueBudgetEvaluation } from "../services/budget-evaluation-work.js"; import { PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared"; @@ -6,12 +14,12 @@ import { registerDeadlineWork } from "../services/deadline-work.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { taskWatchdogService } from "../services/task-watchdogs.js"; import { processAdapter } from "../adapters/process/index.js"; -import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.js"; +import { getServerAdapter, registerServerAdapter, unregisterServerAdapter } from "../adapters/index.js"; import { issueService } from "../services/issues.js"; import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { agents, agentWakeupRequests, applyPendingMigrations, authUsers, companies, companyMemberships, createDb, heartbeatRunEvents, heartbeatRuns, issues, type Db } from "@paperclipai/db"; +import { agents, agentWakeupRequests, applyPendingMigrations, authUsers, companies, companyMemberships, createDb, environmentLeases, issueExecutionDecisions, heartbeatRunEvents, heartbeatRuns, issues, type Db } from "@paperclipai/db"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { heartbeatService } from "../services/heartbeat.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -49,10 +57,13 @@ describe.skipIf(!fixtureUrl)("recovery deadline producer, proof and resumed exec await applyPendingMigrations(adminUrl!); db = createDb(fixtureUrl!, { maxConnections: 4, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); registerServerAdapter({ ...processAdapter, type: "recovery_deadline_process", execute: async context => { - const result = await processAdapter.execute(context); + const result = controlledOrdinaryBootstrapResult(context.config, + controlledCompanyBootstrapResult(context.config, await processAdapter.execute(context))); if (result.exitCode === 0 && !result.timedOut && !result.errorCode) { const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, context.runId)); - if (!current?.issueId) throw new Error("Controlled agent requires its admitted task"); + if (!current) throw new Error("Controlled agent requires its admitted run"); + if (current.scopeKind === "company" && current.issueId === null) return result; + if (!current.issueId) throw new Error("Controlled agent requires its admitted task"); const tasks = issueService(db); await tasks.addComment(current.issueId, "The controlled local process produced recovery-deadline-result; task completed.", { agentId: context.agent.id, runId: context.runId }); @@ -62,6 +73,22 @@ describe.skipIf(!fixtureUrl)("recovery deadline producer, proof and resumed exec return result; } }); }, 90_000); + beforeAll(() => { + registerServerAdapter({ ...processAdapter, type: "recovery_review_process", execute: async context => { + const result = await processAdapter.execute(context); + if (result.exitCode === 75) return { ...result, + executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }; + if (result.exitCode === 0 && !result.timedOut && !result.errorCode) { + const current = await run(context.runId); + if (!current.issueId) throw new Error("Review agent requires its admitted issue"); + const response = await patchIssue(current.issueId, { type: "agent", agentId: context.agent.id, + companyId: current.companyId, runId: context.runId }, + { status: "done", comment: "Reviewed the controlled local process result; approved." }); + expect(response.status, JSON.stringify(response.body)).toBe(200); + } + return result; + } }); + }); beforeEach(async () => { await db.execute(sql`TRUNCATE companies CASCADE`); heartbeat = heartbeatService(db); @@ -84,7 +111,7 @@ describe.skipIf(!fixtureUrl)("recovery deadline producer, proof and resumed exec vi.useRealTimers(); expect(errors).toEqual([]); }); - afterAll(async () => { unregisterServerAdapter("recovery_deadline_process"); await db?.$client.end({ timeout: 1 }); }); + afterAll(async () => { unregisterServerAdapter("recovery_deadline_process"); unregisterServerAdapter("recovery_review_process"); await db?.$client.end({ timeout: 1 }); }); async function start(options: { budgets?: boolean } = {}) { coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, @@ -104,7 +131,7 @@ describe.skipIf(!fixtureUrl)("recovery deadline producer, proof and resumed exec expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5_000 }); } - async function fixture(options: { retainedSourceLock?: boolean; quotaResetAt?: Date } = {}) { + async function fixture(options: { retainedSourceLock?: boolean; quotaResetAt?: Date; companyScope?: boolean } = {}) { const companyId = randomUUID(); await db.insert(authUsers).values({ id: "recovery-test-user", name: "Recovery test owner", email: "recovery-owner@example.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); @@ -119,22 +146,104 @@ describe.skipIf(!fixtureUrl)("recovery deadline producer, proof and resumed exec { companyId, principalId: "recovery-test-user", principalType: "user", status: "active", membershipRole: "owner" }, { companyId, principalId: agent.id, principalType: "agent", status: "active", membershipRole: "member" }, ]); - const [issue] = await db.insert(issues).values({ companyId, title: "Retry producer acceptance", status: "in_progress", assigneeAgentId: agent.id }).returning(); - const [source] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, scopeKind: "issue", issueId: issue.id, + const [issue] = await db.insert(issues).values({ companyId, title: "Retry producer acceptance", status: options.companyScope ? "done" : "in_progress", assigneeAgentId: agent.id }).returning(); + const [source] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, scopeKind: options.companyScope ? "company" : "issue", issueId: options.companyScope ? null : issue.id, status: "failed", responsibleUserId: "recovery-test-user", errorCode: options.quotaResetAt ? "provider_quota" : "codex_transient_upstream", finishedAt: new Date(), resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, ...(options.quotaResetAt ? { errorFamily: "provider_quota", retryNotBefore: options.quotaResetAt.toISOString(), providerQuotaRetryNotBefore: options.quotaResetAt.toISOString(), - } : {}) }, contextSnapshot: { issueId: issue.id }, + } : {}) }, contextSnapshot: options.companyScope ? {} : { issueId: issue.id }, }).returning(); - if (options.retainedSourceLock !== false) + if (!options.companyScope && options.retainedSourceLock !== false) await db.update(issues).set({ executionRunId: source.id, checkoutRunId: source.id }).where(eq(issues.id, issue.id)); return { companyId, agent, issue, source }; } async function schedule(sourceId: string, delayMs = 600_000) { const result = await heartbeat.scheduleBoundedRetry(sourceId, { delayMs, random: () => 0 }); - if (result.outcome !== "scheduled") throw new Error(`Retry producer failed: ${result.outcome}`); + if (result.outcome !== "scheduled") throw new Error(`Retry producer failed: ${JSON.stringify(result)}`); return result; } + async function patchIssue(issueId: string, actor: Request["actor"], body: Record) { + const app = express(); app.use(express.json()); + app.use((req, _res, next) => { req.actor = actor; next(); }); + app.use("/api", issueRoutes(db, {} as Parameters[1])); app.use(errorHandler); + return request(app).patch(`/api/issues/${issueId}`).send(body); + } + async function reviewFailure(quota: boolean) { + const f = await fixture({ retainedSourceLock: false }); + const reset = new Date(Math.ceil((Date.now() + 40_000) / 60_000) * 60_000); + const resetClock = `${reset.getUTCHours()}:${String(reset.getUTCMinutes()).padStart(2, "0")}`; + const message = quota ? `You've hit your usage limit. Try again at ${resetClock} (UTC).` : "Controlled review bootstrap unavailable"; + const [reviewer] = await db.insert(agents).values({ companyId: f.companyId, name: "Controlled reviewer", role: "qa", + status: "paused", adapterType: "recovery_review_process", + adapterConfig: { command: process.execPath, args: ["-e", `console.error(${JSON.stringify(message)});process.exit(75)`], cwd: process.cwd() }, + runtimeConfig: f.agent.runtimeConfig, + }).returning(); + await db.insert(companyMemberships).values({ companyId: f.companyId, principalId: reviewer.id, + principalType: "agent", status: "active", membershipRole: "member" }); + const stageId = randomUUID(); + const response = await patchIssue(f.issue.id, { type: "board", userId: "recovery-test-user", + companyIds: [f.companyId], isInstanceAdmin: true, source: "local_implicit" }, + { status: "in_review", comment: "Implementation ready for controlled review.", executionPolicy: { + stages: [{ id: stageId, type: "review", participants: [{ type: "agent", agentId: reviewer.id }] }], + } }); + expect(response.status, JSON.stringify(response.body)).toBe(200); + const [pending] = await db.select().from(issues).where(eq(issues.id, f.issue.id)); + expect(pending).toMatchObject({ status: "in_review", executionState: { status: "pending", currentStageId: stageId, + currentParticipant: { type: "agent", agentId: reviewer.id } } }); + await db.update(agents).set({ status: "idle" }).where(eq(agents.id, reviewer.id)); + await start(); await settled(); + const source = await heartbeat.wakeup(reviewer.id, { source: "on_demand", reason: "execution_review_participant_recovery", + requestedByActorType: "system", contextSnapshot: { issueId: f.issue.id, currentStageId: stageId } }); + if (!source) throw new Error("Current reviewer was not admitted"); + await vi.waitFor(async () => expect((await run(source.id)).status).toBe("failed"), { timeout: 15_000 }); + await heartbeat.drainActiveRunExecutions(); + const saved = await run(source.id); + expect(saved.stderrExcerpt).toContain(message); + expect(saved.resultJson?.executionRecovery).toMatchObject({ kind: "bootstrap", providerWorkStarted: false }); + await db.update(agents).set({ status: "idle", adapterConfig: { command: process.execPath, + args: ["-e", "console.log('recovery-review-result')"], cwd: process.cwd() } }).where(eq(agents.id, reviewer.id)); + return { ...f, reviewer, stageId, reviewSource: saved, reset }; + } + async function actualCompanyFailure(expectReceipt = true, companyScope = true) { + const f = await fixture(companyScope ? { companyScope: true } : { companyScope: false, retainedSourceLock: false }); + await db.update(agents).set({ adapterConfig: { command: process.execPath, + args: ["-e", companyScope ? controlledCompanyBootstrapScript : controlledOrdinaryBootstrapScript], + cwd: process.cwd() } }).where(eq(agents.id, f.agent.id)); + await start(); await settled(); + const admitted = await heartbeat.wakeup(f.agent.id, { source: "on_demand", reason: "manual", + requestedByActorType: "system", contextSnapshot: companyScope ? {} : { issueId: f.issue.id, taskId: f.issue.id } }); + if (!admitted) throw new Error("Company failure source was not admitted"); + await vi.waitFor(async () => expect((await run(admitted.id)).status).toBe("failed"), { timeout: 15_000 }); + await heartbeat.drainActiveRunExecutions(); + const failed = await run(admitted.id); + expect(failed).toMatchObject({ scopeKind: companyScope ? "company" : "issue", issueId: companyScope ? null : f.issue.id, + status: "failed", errorCode: companyScope ? "adapter_failed" : "codex_transient_upstream" }); + expect(failed.stderrExcerpt).toContain(companyScope ? "Controlled review bootstrap unavailable" : controlledOrdinaryBootstrapProtocol); + expect(failed.processPid).toBeGreaterThan(1); expect(failed.processGroupId).toBeGreaterThan(1); + expect(failed.processStartedAt).toBeInstanceOf(Date); expect(failed.controllerBootId).toEqual(expect.any(String)); + expect(failed.controllerLeaseExpiresAt).toBeNull(); + if (expectReceipt) expect(legacySettlement.legacySourceProcessSettled(failed)).toBe(true); + else expect(failed.legacyProcessSettlement).toBeNull(); + await db.update(agents).set({ adapterConfig: { command: process.execPath, + args: ["-e", "console.log('recovery-deadline-result')"], cwd: process.cwd() } }).where(eq(agents.id, f.agent.id)); + return { ...f, actualSource: failed }; + } + async function assertReviewResult(f: Awaited>, retryId: string) { + await vi.waitFor(async () => expect((await run(retryId)).status).toBe("succeeded"), { timeout: 45_000 }); + await heartbeat.drainActiveRunExecutions(); + expect((await run(retryId)).stdoutExcerpt).toContain("recovery-review-result"); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issue.id)); + expect(task.status).toBe("done"); + const decisions = await db.select().from(issueExecutionDecisions).where(eq(issueExecutionDecisions.issueId, f.issue.id)); + expect(decisions).toHaveLength(1); + expect(decisions[0]).toMatchObject({ stageId: f.stageId, actorAgentId: f.reviewer.id, createdByRunId: retryId, outcome: "approved" }); + const starts = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, retryId), + eq(heartbeatRunEvents.message, "run started"))); + expect(starts).toHaveLength(1); + const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.reviewSource.id)); + expect(successors.map(row => row.id)).toEqual([retryId]); + expect(await heartbeat.promoteDueScheduledRetries()).toEqual({ promoted: 0, runIds: [] }); + } const run = (id: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]); async function proof() { const at = Date.now(); @@ -203,6 +312,314 @@ describe.skipIf(!fixtureUrl)("recovery deadline producer, proof and resumed exec expect(await heartbeat.promoteDueScheduledRetries()).toEqual({ promoted: 0, runIds: [] }); }, 30_000); + it.each([false, true])("cold-reconstructs a real company retry with represented coalescence=%s exactly once", async coalesce => { + const f = await fixture({ companyScope: true }); await start(); await settled(); + const scheduled = await schedule(f.source.id, 30_000); await settled(); + expect(scheduled.dueAt.getTime() - scheduled.run.createdAt.getTime()).toBeGreaterThanOrEqual(29_000); + expect(scheduled.run).toMatchObject({ scopeKind: "company", issueId: null, retryOfRunId: f.source.id }); + if (coalesce) { + const accepted = await heartbeat.wakeup(f.agent.id, { source: "on_demand", reason: "company_retry_followup", + contextSnapshot: { companyRetryInput: "retained-input" }, requestedByActorType: "system" }); + expect(accepted?.id).toBe(scheduled.run.id); + const receipts = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.runId, scheduled.run.id), eq(agentWakeupRequests.status, "coalesced"))); + expect(receipts).toHaveLength(1); const coalescedId = receipts[0].id; + expect(receipts[0].finishedAt).not.toBeNull(); + expect((await run(scheduled.run.id)).contextSnapshot?.companyRetryInput).toBe("retained-input"); + const payload = receipts[0].payload; + const foreign = await fixture({ companyScope: true }); + for (const patch of [{ companyId: foreign.companyId }, { agentId: foreign.agent.id }, { runId: null }, + { finishedAt: null }, { payload: { ...payload, unrepresentedInput: "lost" } }]) { + await db.update(agentWakeupRequests).set(patch).where(eq(agentWakeupRequests.id, coalescedId)); + expect((await proof()).backgroundWork).not.toBe("none"); + await db.update(agentWakeupRequests).set({ payload, finishedAt: receipts[0].finishedAt, companyId: f.companyId, agentId: f.agent.id, runId: scheduled.run.id }).where(eq(agentWakeupRequests.id, coalescedId)); + } + } + const report = await proof(); + expect(report.backgroundWork, JSON.stringify(report.diagnostics)).toBe("none"); + expect(report.scheduledWake?.nextWakeAt).toBe(scheduled.dueAt.toISOString()); + await worker!.stop(); await coordinator!.stop(); + expect((await proof()).backgroundWork).not.toBe("none"); + heartbeat = heartbeatService(db); await start(); await settled(); + expect(coordinator!.nextWakeAt()).toBe(scheduled.dueAt.getTime()); + await vi.waitFor(async () => expect((await run(scheduled.run.id)).status).toBe("succeeded"), { timeout: 40_000 }); + await heartbeat.drainActiveRunExecutions(); + const saved = await run(scheduled.run.id); + expect(saved.stdoutExcerpt).toContain("recovery-deadline-result"); + if (coalesce) expect(saved.contextSnapshot?.companyRetryInput).toBe("retained-input"); + const all = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId)); + expect(all.map(row => row.id).sort()).toEqual([f.source.id, scheduled.run.id].sort()); + const starts = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, scheduled.run.id), + eq(heartbeatRunEvents.message, "run started"))); + expect(starts).toHaveLength(1); + expect(await heartbeat.promoteDueScheduledRetries()).toEqual({ promoted: 0, runIds: [] }); + }, 60_000); + + it("cold-reconstructs an actual failed company process history without erasing its identity", async () => { + const f = await actualCompanyFailure(); + const scheduled = await schedule(f.actualSource.id, 30_000); await settled(); + const report = await proof(); expect(report.backgroundWork, JSON.stringify(report.diagnostics)).toBe("none"); + expect(report.scheduledWake?.nextWakeAt).toBe(scheduled.dueAt.toISOString()); + const receipt = f.actualSource.legacyProcessSettlement; + for (const patch of [{ legacyProcessSettlement: null }, { legacyProcessSettlement: {} }, + { legacyProcessSettlement: { ...receipt, processPid: "malformed" } }, + { legacyProcessSettlement: { ...receipt, companyId: randomUUID() } }, + { processPid: f.actualSource.processPid! + 1 }, { controllerBootId: randomUUID() }, + { controllerLeaseExpiresAt: new Date(0) }, { controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }, { processStartedAt: null }, + { resultJson: { ...f.actualSource.resultJson, cancellation: {} } }]) { + const rollback = new Error("changed historical company settlement rollback"); + await expect(db.transaction(async tx => { + await tx.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, f.actualSource.id)); + expect((await readIdleRetrySchedule(tx, Date.now(), worker!.retries.reconciliationSnapshot)).blocked).toBe(true); + if ("controllerBootId" in patch || "processPid" in patch || "controllerLeaseExpiresAt" in patch || + "processStartedAt" in patch || "resultJson" in patch) { + expect(await legacySettlement.recordLegacyProcessSettlement(tx as unknown as Db, f.actualSource, () => true)).toBe(false); + } + throw rollback; + })).rejects.toThrow(rollback.message); + } + const pendingCleanupRollback = new Error("pending source cleanup rollback"); + await expect(db.transaction(async tx => { + await tx.update(heartbeatRuns).set({ legacyProcessSettlement: null }).where(eq(heartbeatRuns.id, f.actualSource.id)); + await tx.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.actualSource.id, + provider: "local", status: "pending_cleanup", cleanupStatus: "failed" }); + expect(await legacySettlement.recordLegacyProcessSettlement(tx as unknown as Db, f.actualSource, () => true)).toBe(false); + const [held] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.actualSource.id)); + expect(held.legacyProcessSettlement).toBeNull(); + expect((await readIdleRetrySchedule(tx, Date.now(), worker!.retries.reconciliationSnapshot)).blocked).toBe(true); + throw pendingCleanupRollback; + })).rejects.toThrow(pendingCleanupRollback.message); + await worker!.stop(); await coordinator!.stop(); + expect((await proof()).backgroundWork).not.toBe("none"); + heartbeat = heartbeatService(db); await start(); await settled(); + expect(coordinator!.nextWakeAt()).toBe(scheduled.dueAt.getTime()); + await vi.waitFor(async () => expect((await run(scheduled.run.id)).status).toBe("succeeded"), { timeout: 45_000 }); + await heartbeat.drainActiveRunExecutions(); + expect((await run(scheduled.run.id)).stdoutExcerpt).toContain("recovery-deadline-result"); + expect((await run(f.actualSource.id)).processPid).toBe(f.actualSource.processPid); + const starts = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.runId, scheduled.run.id), eq(heartbeatRunEvents.message, "run started"))); + expect(starts).toHaveLength(1); + const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.actualSource.id)); + expect(successors).toHaveLength(1); + expect(await heartbeat.promoteDueScheduledRetries()).toEqual({ promoted: 0, runIds: [] }); + }, 60_000); + + it("cold-reconstructs an actual failed ordinary issue process with exact source settlement", async () => { + const jitter = vi.spyOn(Math, "random").mockReturnValue(0.5); + let f: Awaited>; + try { f = await actualCompanyFailure(true, false); } + finally { jitter.mockRestore(); } + const produced = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.actualSource.id)); + expect(produced, JSON.stringify(produced.map(row => ({ id: row.id, status: row.status, dueAt: row.scheduledRetryAt, + reason: row.scheduledRetryReason })))).toHaveLength(1); + expect(produced[0]).toMatchObject({ status: "scheduled_retry", scopeKind: "issue", issueId: f.issue.id, + agentId: f.agent.id, companyId: f.companyId, scheduledRetryReason: "transient_failure", scheduledRetryAttempt: 1, + startedAt: null, finishedAt: null }); + expect(produced[0].scheduledRetryAt).toBeInstanceOf(Date); + const scheduled = { run: produced[0], dueAt: produced[0].scheduledRetryAt! }; + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, scheduled.run.wakeupRequestId!)); + expect(wake).toMatchObject({ runId: scheduled.run.id, companyId: f.companyId, agentId: f.agent.id, status: "queued" }); + const producerEvents = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.actualSource.id), + sql`${heartbeatRunEvents.message} LIKE 'Scheduled bounded retry %'`)); + expect(producerEvents).toHaveLength(1); + expect(producerEvents[0].payload).toMatchObject({ retryRunId: scheduled.run.id, retryReason: "transient_failure", + scheduledRetryAt: scheduled.dueAt.toISOString(), baseDelayMs: 30_000, delayMs: 30_000 }); + expect(scheduled.dueAt.getTime()).toBeGreaterThan(Date.now()); + await settled(); + const observedReceipt = f.actualSource.legacyProcessSettlement; + expect(observedReceipt).not.toBeNull(); + expect(f.actualSource).toMatchObject({ scopeKind: "issue", issueId: f.issue.id, nativeIssueId: null, nativeSessionId: null }); + expect((await readIdleRetrySchedule(db, Date.now(), worker!.retries.reconciliationSnapshot)).blocked).toBe(false); + const report = await proof(); expect(report.backgroundWork, JSON.stringify(report.diagnostics)).toBe("none"); + expect(report.scheduledWake?.nextWakeAt).toBe(scheduled.dueAt.toISOString()); + for (const patch of [{ legacyProcessSettlement: null }, { legacyProcessSettlement: {} }, + { processPid: f.actualSource.processPid! + 1 }, { processStartedAt: new Date(f.actualSource.processStartedAt!.getTime() + 1000) }, + { controllerLeaseExpiresAt: new Date(0) }, { nativeIssueId: f.issue.id }, + { resultJson: { ...f.actualSource.resultJson, cancellation: {} } }]) { + const rollback = new Error("ordinary issue source regression rollback"); + await expect(db.transaction(async tx => { + await tx.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, f.actualSource.id)); + expect((await readIdleRetrySchedule(tx, Date.now(), worker!.retries.reconciliationSnapshot)).blocked).toBe(true); + throw rollback; + })).rejects.toThrow(rollback.message); + } + expect((await run(f.actualSource.id)).legacyProcessSettlement).toEqual(observedReceipt); + await settled(); + await worker!.stop(); await coordinator!.stop(); + expect((await proof()).backgroundWork).not.toBe("none"); + heartbeat = heartbeatService(db); await start(); await settled(); + expect(coordinator!.nextWakeAt()).toBe(scheduled.dueAt.getTime()); + await vi.waitFor(async () => expect((await run(scheduled.run.id)).status).toBe("succeeded"), { timeout: 45_000 }); + await heartbeat.drainActiveRunExecutions(); + expect((await run(scheduled.run.id)).stdoutExcerpt).toContain("recovery-deadline-result"); + expect((await run(f.actualSource.id)).processPid).toBe(f.actualSource.processPid); + expect((await run(f.actualSource.id)).legacyProcessSettlement).toEqual(observedReceipt); + const starts = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.runId, scheduled.run.id), eq(heartbeatRunEvents.message, "run started"))); + expect(starts).toHaveLength(1); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.actualSource.id))).map(row => row.id)) + .toEqual([scheduled.run.id]); + expect(await heartbeat.promoteDueScheduledRetries()).toEqual({ promoted: 0, runIds: [] }); + }, 60_000); + + it("keeps failed optional settlement persistence blocked while normal execution cleanup continues", async () => { + const failedWrite = vi.spyOn(legacySettlement, "recordLegacyProcessSettlement") + .mockRejectedValue(new Error("controlled optional settlement write failure")); + try { + const f = await actualCompanyFailure(false); + expect(failedWrite).toHaveBeenCalledTimes(1); + const scheduled = await schedule(f.actualSource.id, 30_000); await settled(); + expect((await proof()).backgroundWork).not.toBe("none"); + expect((await run(f.actualSource.id)).status).toBe("failed"); + await heartbeat.cancelRun(scheduled.run.id, "controlled receipt failure cleanup", { suppressImmediateRecovery: true }); + await heartbeat.drainActiveRunExecutions(); + expect((await run(scheduled.run.id)).status).toBe("cancelled"); + } finally { failedWrite.mockRestore(); } + }, 30_000); + + async function claimedPreparationFailure(optionalWriteFailure = false) { + const f = await fixture({ retainedSourceLock: false }); + const control = await preparationFailureFixture(db, f.agent.id); + const entered = vi.spyOn(getServerAdapter("recovery_deadline_process"), "execute"); + const persisted = vi.spyOn(legacySettlement, "recordLegacyPreparationSettlement"); + if (optionalWriteFailure) persisted.mockRejectedValueOnce(new Error("controlled preparation receipt persistence failure")); + heartbeat = heartbeatService(db, { environmentRuntime: control.runtime }); + try { await start(); await settled(); } + catch (error) { control.unblock(); entered.mockRestore(); persisted.mockRestore(); await control.dispose(); throw error; } + return { ...f, control, entered, persisted }; + } + + async function disposePreparation(f: Awaited>) { + f.control.unblock(); + try { await worker?.stop(); await coordinator?.stop(); await heartbeat.drainActiveRunExecutions(); } + finally { f.entered.mockRestore(); f.persisted.mockRestore(); await f.control.dispose(); } + } + + it("certifies actual never-dispatched preparation only after owned cleanup and cold-reconstructs its saved retry", async () => { + const f = await claimedPreparationFailure(); + const jitter = vi.spyOn(Math, "random").mockReturnValue(0.5); + try { + const failed = await drivePreparationFailure({ db, heartbeat, control: f.control, entered: f.entered, persisted: f.persisted, + companyId: f.companyId, agentId: f.agent.id, issueId: f.issue.id, run, owner: worker!.retries.reconciliationSnapshot }); jitter.mockRestore(); await settled(); + expect(failed.controllerLeaseExpiresAt).toBeNull(); expect(failed.processPid).toBeNull(); + expect(f.persisted).toHaveBeenCalledTimes(1); expect(legacySettlement.legacySourceProcessSettled(failed)).toBe(true); + expect(failed.legacyProcessSettlement).toMatchObject({ schema: "paperclip.legacy-preparation-settlement.v1", + targetKind: "not-dispatched", adapterDispatched: false, runId: failed.id, controllerBootId: failed.controllerBootId }); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, failed.id)); + expect(retry).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "transient_failure", issueId: f.issue.id, + startedAt: null, finishedAt: null }); expect(retry.scheduledRetryAt).toBeInstanceOf(Date); + const dueAt = retry.scheduledRetryAt!; + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, retry.wakeupRequestId!)); + expect(wake).toMatchObject({ runId: retry.id, status: "queued", companyId: f.companyId, agentId: f.agent.id }); + const scheduled = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, failed.id), + sql`${heartbeatRunEvents.message} LIKE 'Scheduled bounded retry %'`)); + expect(scheduled).toHaveLength(1); expect(scheduled[0].payload).toMatchObject({ delayMs: 30_000, baseDelayMs: 30_000, + retryRunId: retry.id, scheduledRetryAt: dueAt.toISOString() }); + expect((await readIdleRetrySchedule(db, Date.now(), worker!.retries.reconciliationSnapshot)).blocked).toBe(false); + expect((await proof()).backgroundWork).not.toBe("none"); + for (const patch of [{ controllerBootId: randomUUID() }, { controllerLeaseExpiresAt: new Date(0) }, + { startedAt: new Date(0) }, { executionStage: "dispatching" }, { costAccountingPending: true }, + { nativeIssueId: f.issue.id }, { processPid: 42 }, { processStartedAt: new Date() }, + { resultJson: { ...failed.resultJson, cancellation: {} } }]) { + const rollback = new Error("preparation current SQL fence rollback"); + await expect(db.transaction(async tx => { + await tx.update(heartbeatRuns).set({ ...patch, legacyProcessSettlement: null }).where(eq(heartbeatRuns.id, failed.id)); + expect(await legacySettlement.recordLegacyPreparationSettlement(tx as unknown as Db, failed, () => true)).toBe(false); + const [current] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, failed.id)); + expect(current.legacyProcessSettlement).toBeNull(); + throw rollback; + })).rejects.toThrow(rollback.message); + } + const cleanupRollback = new Error("preparation pending cleanup SQL rollback"); + await expect(db.transaction(async tx => { + await tx.update(heartbeatRuns).set({ legacyProcessSettlement: null }).where(eq(heartbeatRuns.id, failed.id)); + await tx.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: failed.id, + provider: "local", status: "pending_cleanup", cleanupStatus: "failed" }); + expect(await legacySettlement.recordLegacyPreparationSettlement(tx as unknown as Db, failed, () => true)).toBe(false); + const [current] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, failed.id)); + expect(current.legacyProcessSettlement).toBeNull(); throw cleanupRollback; + })).rejects.toThrow(cleanupRollback.message); + expect((await run(failed.id)).legacyProcessSettlement).toEqual(failed.legacyProcessSettlement); + await f.control.useLocalEnvironment(); + await worker!.stop(); await coordinator!.stop(); + expect((await readIdleRetrySchedule(db, Date.now(), worker!.retries.reconciliationSnapshot)).blocked).toBe(true); + heartbeat = heartbeatService(db, { environmentRuntime: f.control.runtime }); await start(); await settled(); + expect(coordinator!.nextWakeAt()).toBe(dueAt.getTime()); + await new Promise(resolve => setTimeout(resolve, Math.max(0, dueAt.getTime() - Date.now()))); + await finish(retry.id); + expect(f.entered).toHaveBeenCalledTimes(1); + expect((await run(failed.id)).processPid).toBeNull(); + expect((await run(failed.id)).legacyProcessSettlement).toEqual(failed.legacyProcessSettlement); + expect(await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, retry.id), + eq(heartbeatRunEvents.message, "run started")))).toHaveLength(1); + expect(await heartbeat.promoteDueScheduledRetries()).toEqual({ promoted: 0, runIds: [] }); + } finally { jitter.mockRestore(); await disposePreparation(f); } + }, 60_000); + + it("leaves a failed optional preparation receipt blocked while normal cleanup and retry ownership complete", async () => { + const f = await claimedPreparationFailure(true); const jitter = vi.spyOn(Math, "random").mockReturnValue(0.5); + try { + const failed = await drivePreparationFailure({ db, heartbeat, control: f.control, entered: f.entered, persisted: f.persisted, + companyId: f.companyId, agentId: f.agent.id, issueId: f.issue.id, run, owner: worker!.retries.reconciliationSnapshot }); jitter.mockRestore(); await settled(); + expect(f.persisted).toHaveBeenCalledTimes(1); expect(failed.legacyProcessSettlement).toBeNull(); + expect(failed.controllerLeaseExpiresAt).toBeNull(); expect(f.entered).not.toHaveBeenCalled(); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, failed.id)); + expect(retry).toMatchObject({ status: "scheduled_retry" }); expect(retry.scheduledRetryAt).toBeInstanceOf(Date); + expect((await readIdleRetrySchedule(db, Date.now(), worker!.retries.reconciliationSnapshot)).blocked).toBe(true); + await heartbeat.cancelRun(retry.id, "controlled failed preparation proof cleanup", { suppressImmediateRecovery: true }); + expect((await run(retry.id)).status).toBe("cancelled"); + } finally { jitter.mockRestore(); await disposePreparation(f); } + }, 60_000); + + it("never certifies an adapter that was entered without publishing a process identity", async () => { + const f = await fixture({ retainedSourceLock: false }); + const execution = vi.spyOn(getServerAdapter("recovery_deadline_process"), "execute").mockResolvedValueOnce({ + exitCode: 75, signal: null, timedOut: false, errorCode: "codex_transient_upstream", errorFamily: "transient_upstream", + executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }); + const jitter = vi.spyOn(Math, "random").mockReturnValue(0.5); + try { + await start(); await settled(); + const admitted = await heartbeat.wakeup(f.agent.id, { source: "on_demand", reason: "manual", requestedByActorType: "system", + contextSnapshot: { issueId: f.issue.id, taskId: f.issue.id } }); + if (!admitted) throw new Error("Adapter-entry negative was not admitted"); + await heartbeat.drainActiveRunExecutions(); await settled(); + const failed = await run(admitted.id); + expect(execution).toHaveBeenCalledTimes(1); expect(failed).toMatchObject({ status: "failed", processPid: null, + processGroupId: null, processStartedAt: null, legacyProcessSettlement: null }); + expect(failed.controllerBootId).toEqual(expect.any(String)); expect(failed.controllerLeaseExpiresAt).toBeNull(); + expect(legacySettlement.legacySourceProcessSettled(failed)).toBe(false); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, failed.id)); + expect(retry).toMatchObject({ status: "scheduled_retry" }); + expect((await readIdleRetrySchedule(db, Date.now(), worker!.retries.reconciliationSnapshot)).blocked).toBe(true); + await heartbeat.cancelRun(retry.id, "controlled entered-adapter cleanup", { suppressImmediateRecovery: true }); + } finally { jitter.mockRestore(); execution.mockRestore(); } + }, 60_000); + + it("keeps company cancellation and first producer during drain fenced", async () => { + const f = await fixture({ companyScope: true }); await start(); await settled(); + const old = await owners.retries!(db, Date.now(), () => {}); + draining = true; + const scheduled = await schedule(f.source.id, 30_000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5_000 }); + expect(old.current()).toBe(false); + expect((await proof()).backgroundWork).not.toBe("none"); + draining = false; worker!.wake(); await settled(); + expect((await proof()).backgroundWork).toBe("none"); + for (const principalId of [f.agent.id, "recovery-test-user"]) { + await db.update(companyMemberships).set({ status: "inactive" }).where(and( + eq(companyMemberships.companyId, f.companyId), eq(companyMemberships.principalId, principalId))); + expect((await proof()).backgroundWork).not.toBe("none"); + await db.update(companyMemberships).set({ status: "active" }).where(and( + eq(companyMemberships.companyId, f.companyId), eq(companyMemberships.principalId, principalId))); + } + await heartbeat.cancelRun(scheduled.run.id, "company operator stop", { suppressImmediateRecovery: true }); + await worker!.stop(); await coordinator!.stop(); heartbeat = heartbeatService(db); + await start(); await settled(); + expect(await heartbeat.promoteDueScheduledRetries(new Date(scheduled.dueAt.getTime() + 1))).toEqual({ promoted: 0, runIds: [] }); + expect(await run(scheduled.run.id)).toMatchObject({ status: "cancelled", startedAt: null }); + }, 20_000); + it("cancellation before the deadline invalidates its alarm and prevents resumed execution", async () => { const f = await fixture(); await start(); await settled(); const scheduled = await schedule(f.source.id); await settled(); @@ -235,6 +652,96 @@ describe.skipIf(!fixtureUrl)("recovery deadline producer, proof and resumed exec await heartbeat.cancelRun(scheduled.run.id, "fixture cleanup", { suppressImmediateRecovery: true }); }, 20_000); + it("cold-reconstructs the current reviewer retry and persists one ordinary review decision", async () => { + const f = await reviewFailure(false); + const scheduled = await heartbeat.scheduleBoundedRetry(f.reviewSource.id, { delayMs: 30_000, random: () => 0, + retryReason: "execution_review_participant_recovery", wakeReason: "execution_review_participant_recovery" }); + if (scheduled.outcome !== "scheduled") throw new Error(`Review retry not scheduled: ${scheduled.outcome}`); + await settled(); + const report = await proof(); + expect(report.backgroundWork, JSON.stringify(report.diagnostics)).toBe("none"); + expect(report.scheduledWake?.nextWakeAt).toBe(scheduled.dueAt.toISOString()); + const [pending] = await db.select().from(issues).where(eq(issues.id, f.issue.id)); + for (const state of [{ ...pending.executionState, currentStageId: randomUUID() }, + { ...pending.executionState, currentParticipant: { type: "user", userId: "recovery-test-user" } }]) { + const rollback = new Error("review retry changed participant rollback"); + await expect(db.transaction(async tx => { + await tx.update(issues).set({ executionState: state }).where(eq(issues.id, f.issue.id)); + expect((await readIdleRetrySchedule(tx, Date.now(), worker!.retries.reconciliationSnapshot)).blocked).toBe(true); + throw rollback; + })).rejects.toThrow(rollback.message); + } + await worker!.stop(); await coordinator!.stop(); heartbeat = heartbeatService(db); + expect((await proof()).backgroundWork).not.toBe("none"); + await start(); await settled(); + expect(coordinator!.nextWakeAt()).toBe(scheduled.dueAt.getTime()); + await assertReviewResult(f, scheduled.run.id); + }, 60_000); + + it("carries actual reviewer quota failure through monitor cold reconstruction, due claim and useful review", async () => { + const f = await reviewFailure(true); + const monitor = () => registerDeadlineWork(coordinator!, DELIVERY_QUEUES.issueMonitors, { + retryMs: 1000, canRun: () => !draining, + run: signal => heartbeat.tickIssueMonitors(new Date(), () => !signal.aborted && !draining), + next: () => heartbeat.nextIssueMonitorAt(), + }); + monitorWorker = monitor(); await monitorWorker.ready; + owners.quota = bindIdleDeadlineOwner(monitorWorker.reconciliationSnapshot, readIdleQuotaSchedule, + snapshot => ({ issueWaits: snapshot.issueWaits })); + const recovery = await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId, issueId: f.issue.id }); + expect(recovery.providerQuotaMonitored).toBe(1); + await vi.waitFor(() => expect(monitorWorker!.reconciliationSnapshot().reconciled).toBe(true)); await settled(); + const [saved] = await db.select().from(issues).where(eq(issues.id, f.issue.id)); + expect(saved.monitorNextCheckAt).toEqual(f.reset); + expect(saved.executionPolicy?.monitor).toMatchObject({ externalRef: f.reviewSource.id, + serviceName: PROVIDER_QUOTA_MONITOR_SERVICE_NAME }); + const report = await proof(); + expect(report.backgroundWork, JSON.stringify(report.diagnostics)).toBe("none"); + expect(report.scheduledWake?.nextWakeAt).toBe(f.reset.toISOString()); + for (const state of [{ ...saved.executionState, currentStageId: randomUUID() }, + { ...saved.executionState, currentParticipant: { type: "user", userId: "recovery-test-user" } }]) { + const rollback = new Error("review quota changed authority rollback"); + await expect(db.transaction(async tx => { + await tx.update(issues).set({ executionState: state }).where(eq(issues.id, f.issue.id)); + expect((await readIdleQuotaSchedule(tx, Date.now(), monitorWorker!.reconciliationSnapshot)).blocked).toBe(true); + throw rollback; + })).rejects.toThrow(rollback.message); + } + const paused = new Error("review quota revoked current authority rollback"); + await expect(db.transaction(async tx => { + await tx.update(agents).set({ status: "paused" }).where(eq(agents.id, f.reviewer.id)); + expect((await readIdleQuotaSchedule(tx, Date.now(), monitorWorker!.reconciliationSnapshot)).blocked).toBe(true); + throw paused; + })).rejects.toThrow(paused.message); + const rollback = new Error("native review quota source rollback"); + await expect(db.transaction(async tx => { + await tx.update(heartbeatRuns).set({ runtimeMode: "native" }).where(eq(heartbeatRuns.id, f.reviewSource.id)); + expect((await readIdleQuotaSchedule(tx, Date.now(), monitorWorker!.reconciliationSnapshot)).blocked).toBe(true); + throw rollback; + })).rejects.toThrow(rollback.message); + await monitorWorker.stop(); await worker!.stop(); await coordinator!.stop(); + heartbeat = heartbeatService(db); await start(); monitorWorker = monitor(); await monitorWorker.ready; + owners.quota = bindIdleDeadlineOwner(monitorWorker.reconciliationSnapshot, readIdleQuotaSchedule, + snapshot => ({ issueWaits: snapshot.issueWaits })); + expect(coordinator!.nextWakeAt()).toBe(f.reset.getTime()); + let retryId = ""; + await vi.waitFor(async () => { + const rows = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.reviewSource.id)); + expect(rows).toHaveLength(1); retryId = rows[0].id; + expect(rows[0].scheduledRetryReason).toBe("execution_review_participant_recovery"); + }, { timeout: Math.max(0, f.reset.getTime() - Date.now()) + 10_000 }); + await settled(); + const retry = await run(retryId); + expect(retry.agentId).toBe(f.reviewer.id); + expect(retry.scheduledRetryAt!.getTime()).toBeGreaterThan(f.reset.getTime()); + const retryDelay = retry.scheduledRetryAt!.getTime() - retry.createdAt.getTime(); + expect(retryDelay).toBeGreaterThanOrEqual(23_900); + expect(retryDelay).toBeLessThanOrEqual(36_100); + const retryProof = await proof(); + expect(retryProof.backgroundWork, JSON.stringify(retryProof.diagnostics)).toBe("none"); + await assertReviewResult(f, retryId); + }, 170_000); + it("the quota recovery producer saves an exact future monitor inspected by its registered owner", async () => { const reset = new Date(Date.now() + 600_000); const f = await fixture({ retainedSourceLock: false, quotaResetAt: reset }); diff --git a/server/src/__tests__/recovery-deadline-bootstrap.test.ts b/server/src/__tests__/recovery-deadline-bootstrap.test.ts new file mode 100644 index 00000000000..14b6c418444 --- /dev/null +++ b/server/src/__tests__/recovery-deadline-bootstrap.test.ts @@ -0,0 +1,36 @@ +import { describe, expect, it } from "vitest"; +import type { AdapterExecutionResult } from "../adapters/types.js"; +import { legacyExecutionNeedsReconciliation } from "../services/legacy-execution-recovery.js"; +import { controlledCompanyBootstrapMessage, controlledCompanyBootstrapResult, controlledCompanyBootstrapScript } from "./helpers/recovery-deadline-bootstrap.js"; +const config = { command: process.execPath, args: ["-e", controlledCompanyBootstrapScript] }; +const raw: AdapterExecutionResult = { exitCode: 75, signal: null, timedOut: false, errorMessage: "Process exited with code 75", + resultJson: { stdout: "", stderr: controlledCompanyBootstrapMessage + "\n" } }; +const needsReconciliation = (result: AdapterExecutionResult) => legacyExecutionNeedsReconciliation({ runtimeMode: "legacy", + status: "failed", errorCode: "adapter_failed", resultJson: { ...result.resultJson, + ...(result.executionRecovery ? { executionRecovery: result.executionRecovery } : {}) } }); +describe("controlled company bootstrap fixture contract", () => { + it("distinguishes stopped-process evidence from replay-safe pre-task failure", () => { + expect(needsReconciliation(raw)).toBe(true); + const result = controlledCompanyBootstrapResult(config, raw); + expect(result.executionRecovery).toEqual({ kind: "bootstrap", providerWorkStarted: false }); + expect(needsReconciliation(result)).toBe(false); + expect(result.resultJson).toBe(raw.resultJson); + expect(result.exitCode).toBe(75); + }); + it("does not classify another command, appended action, extra argument, or unknown program", () => { + for (const candidate of [{ ...config, command: "/unowned/node" }, { ...config, args: ["-e", controlledCompanyBootstrapScript + ";doWork()"] }, + { ...config, args: [...config.args, "extra"] }, { ...config, args: ["-e", "unknownWork()"] }]) { + expect(controlledCompanyBootstrapResult(candidate, raw)).toBe(raw); + expect(needsReconciliation(controlledCompanyBootstrapResult(candidate, raw))).toBe(true); + } + }); + it("preserves unknown output, timeout and signal failures without replay evidence", () => { + for (const patch of [{ timedOut: true }, { signal: "SIGTERM" }, { exitCode: 1 }, + { resultJson: { stdout: "task action happened", stderr: controlledCompanyBootstrapMessage + "\n" } }, + { resultJson: { stdout: "", stderr: "Unknown provider action outcome\n" } }]) { + const result = { ...raw, ...patch } as AdapterExecutionResult; + expect(controlledCompanyBootstrapResult(config, result)).toBe(result); + expect(needsReconciliation(controlledCompanyBootstrapResult(config, result))).toBe(true); + } + }); +}); diff --git a/server/src/__tests__/recovery-deadline-ordinary-bootstrap.test.ts b/server/src/__tests__/recovery-deadline-ordinary-bootstrap.test.ts new file mode 100644 index 00000000000..28540a8fec7 --- /dev/null +++ b/server/src/__tests__/recovery-deadline-ordinary-bootstrap.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, it } from "vitest"; +import type { AdapterExecutionResult } from "../adapters/types.js"; +import { classifyCodexTerminalSessionFailure } from "@paperclipai/adapter-codex-local/server"; +import { legacyExecutionNeedsReconciliation } from "../services/legacy-execution-recovery.js"; +import { readTransientRecoveryContractFromRun } from "../services/heartbeat/retries.js"; +import { controlledOrdinaryBootstrapFailure, controlledOrdinaryBootstrapProtocol, controlledOrdinaryBootstrapResult, + controlledOrdinaryBootstrapScript } from "./helpers/recovery-deadline-ordinary-bootstrap.js"; +const config = { command: process.execPath, args: ["-e", controlledOrdinaryBootstrapScript] }; +const raw: AdapterExecutionResult = { exitCode: 75, signal: null, timedOut: false, errorMessage: "Process exited with code 75", + resultJson: { stdout: "", stderr: controlledOrdinaryBootstrapProtocol + "\n" } }; +const persisted = (result: AdapterExecutionResult) => ({ runtimeMode: "legacy" as const, status: "failed" as const, + errorCode: result.errorCode ?? "adapter_failed", resultJson: { ...result.resultJson, + ...(result.errorFamily ? { errorFamily: result.errorFamily } : {}), + ...(result.executionRecovery ? { executionRecovery: result.executionRecovery } : {}) } }); +describe("controlled ordinary retry bootstrap contract", () => { + it("uses the production typed service classifier and replay-safe no-action contract", () => { + const result = controlledOrdinaryBootstrapResult(config, raw); + const classified = classifyCodexTerminalSessionFailure(controlledOrdinaryBootstrapFailure, new Date()); + expect(classified).not.toBeNull(); + if (!classified) throw new Error("Controlled service-capacity failure must have a production retry classification"); + expect(result).toMatchObject(classified); + expect(result.executionRecovery).toEqual({ kind: "bootstrap", providerWorkStarted: false }); + expect(readTransientRecoveryContractFromRun(persisted(result))).toEqual({ errorFamily: "transient_upstream", retryNotBefore: null }); + expect(legacyExecutionNeedsReconciliation(persisted(result))).toBe(false); + expect(result.resultJson).toBe(raw.resultJson); + }); + it("rejects another executable, modified program and additional arguments", () => { + for (const candidate of [{ ...config, command: "/unowned/node" }, { ...config, args: ["-e", controlledOrdinaryBootstrapScript + ";doWork()"] }, + { ...config, args: [...config.args, "extra"] }]) expect(controlledOrdinaryBootstrapResult(candidate, raw)).toBe(raw); + }); + it("leaves unknown outputs, malformed protocol, timeouts, signals and other exits unreconciled", () => { + for (const patch of [{ timedOut: true }, { signal: "SIGTERM" }, { exitCode: 1 }, + { resultJson: { stdout: "domain action", stderr: controlledOrdinaryBootstrapProtocol + "\n" } }, + { resultJson: { stdout: "", stderr: "{malformed}" } }, + { resultJson: { stdout: "", stderr: JSON.stringify({ category: "service", title: "Unknown outcome" }) + "\n" } }]) { + const result = { ...raw, ...patch } as AdapterExecutionResult; + expect(controlledOrdinaryBootstrapResult(config, result)).toBe(result); + expect(readTransientRecoveryContractFromRun(persisted(result))).toBeNull(); + expect(legacyExecutionNeedsReconciliation(persisted(result))).toBe(true); + } + }); +}); diff --git a/server/src/__tests__/run-secret-redaction.test.ts b/server/src/__tests__/run-secret-redaction.test.ts index 5b52a3a0efa..a6f9a093e7e 100644 --- a/server/src/__tests__/run-secret-redaction.test.ts +++ b/server/src/__tests__/run-secret-redaction.test.ts @@ -7,6 +7,11 @@ import { createRunSecretRedactionRegistry, redactRegisteredSecretValues } from " const secret = "q2a-exact-secret-value"; describe("registered run secret redaction", () => { + it("omits host-only legacy settlement receipts from full and nested run responses", () => { + const proof = { schema: "paperclip.legacy-process-settlement.v1", processPid: 42 }; + expect(redactRegisteredSecretValues({ id: "run", legacyProcessSettlement: proof, + nested: [{ legacyProcessSettlement: proof, status: "failed" }] }, [])).toEqual({ id: "run", nested: [{ status: "failed" }] }); + }); it("strips private credential fields and denies names unless the caller supplies an authorized projection", () => { const input = { resultJson: { configurationIncomplete: { credentialAccess: { connectionName: "private name", grantId: "private grant", secret: "private value" }, diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index ea2c8379175..49877b828a5 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -560,7 +560,10 @@ vi.mock("../services/index.js", () => ({ })), })); -vi.mock("../services/chat-completion-delivery.js", () => ({ chatCompletionDeliveryService: () => ({ hasPending: vi.fn(async () => false), sweepPending: completionSweepMock }) })); +vi.mock("../services/chat-completion-delivery.js", () => ({ + CHAT_COMPLETION_WAKE_REASON: "chat_task_completed", + chatCompletionDeliveryService: () => ({ hasPending: vi.fn(async () => false), sweepPending: completionSweepMock }), +})); // Only startup registration belongs here; real handoff persistence, authority, // and provider settlement are covered by the database-backed workflow suites. diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 1d9d40bb4be..c7bbd976b47 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -27,6 +27,7 @@ import { promptInjectionPreventionService } from "./prompt-injection-prevention. import { createHeartbeatAgentStatus } from "./heartbeat/agent-status.js"; import { heartbeatExternalProvider, nativeRuntimeHasInstructionFilesystem } from "./heartbeat/external-provider.js"; import { pendingCaseHandoffForRun, recordCaseHandoffExecutionReleased, recordLegacyCaseHandoffCleanupIntent, recordLegacyCaseHandoffCleanupVerified } from "./heartbeat/case-handoff-lifecycle.js"; +import { recordLegacyPreparationSettlement, recordLegacyProcessSettlement } from "./heartbeat/legacy-process-settlement.js"; import { hasStoppedLocalProcessTree, legacyAdapterNeedsLocalProcessSettlement } from "./heartbeat/process-settlement.js"; import { dotRunnerBroker } from "./dot-runner-broker.js"; import { withRunExecutionTiming, measureRunPhase, measureWorkspaceSyncPhase, isWorkspaceSyncDetailPhase } from "./execution-preparation-timing.js"; @@ -1885,6 +1886,7 @@ export function heartbeatService( let legacyAdapterEntered = false; let legacyLocalAdapterEntered = false; + let legacyPreparationFailedBeforeDispatch = false; let persistUsageCaptureFailure: (() => Promise) | undefined; let run = await getRun(runId); if (!run) return; @@ -3800,14 +3802,18 @@ export function heartbeatService( ); } else { await preserveNativeWorkspaceOnFailure(); - if (!nativeOwnershipHeld && !nativeStartupCleanupScheduled && !nativeStartupCleanupError) await runCompletion.failRunSetup({ - run, runId, outerErr, signal: executionControl.controller.signal, - providerDispatchStarted: legacyAdapterEntered || nativeDispatchStarted, - identityRedactor, readFailureReportSecrets, - }); + if (!nativeOwnershipHeld && !nativeStartupCleanupScheduled && !nativeStartupCleanupError) { + await runCompletion.failRunSetup({ + run, runId, outerErr, signal: executionControl.controller.signal, + providerDispatchStarted: legacyAdapterEntered || nativeDispatchStarted, + identityRedactor, readFailureReportSecrets, + }); + legacyPreparationFailedBeforeDispatch = !legacyAdapterEntered && !nativeDispatchStarted; + } } } finally { let caseHandoffCleanupSettled = false; + let failedProcessCancellationSeen = failedProcessRunCancellations.has(run.id); await instructionWorkingCopy?.nativeInstructionReservation?.release().catch(error => logger.warn({ runId: run.id, err: error }, "Managed warm session preparation cleanup failed")); if (managedAiRuntime) await managedAiRuntime.cleanup().catch(() => logger.warn({ runId: run.id }, "AI connection refresh or cleanup failed")); let latestRun = await getRun(run.id).catch(() => null); @@ -4050,6 +4056,7 @@ export function heartbeatService( // A failed owned Stop remains visible until this exact executor settles, // including a graceful exit result arriving after the cancellation error. // It is never retained beyond the active execution's cleanup. + failedProcessCancellationSeen ||= failedProcessRunCancellations.has(run.id); failedProcessRunCancellations.delete(run.id); executionControl.finish(); if (adapterExecutionControls.get(run.id) === executionControl) { @@ -4061,6 +4068,22 @@ export function heartbeatService( // that a local legacy adapter's closed-stdio descendants stopped. if (caseHandoffCleanupSettled && latestRun?.runtimeMode === "legacy" && legacyLocalAdapterEntered) { caseHandoffCleanupSettled = hasStoppedLocalProcessTree(latestRun); + const canRecordSettlement = () => caseHandoffCleanupSettled && + !nativeSessionResumeScheduled && !nativeWorkspaceFinalizeScheduled && !nativeOwnershipHeld && + !nativeStartupCleanupScheduled && !nativeStartupCleanupError && !failedProcessCancellationSeen && + !failedProcessRunCancellations.has(run.id) && !processRunCancellationSettlements.has(run.id) && + !adapterExecutionControls.has(run.id) && !activeRunExecutions.has(run.id); + if (canRecordSettlement()) await recordLegacyProcessSettlement(db, latestRun, canRecordSettlement) + .catch(error => logger.warn({ runId: run.id, error }, "Could not record optional legacy process settlement")); + } + if (caseHandoffCleanupSettled && latestRun?.runtimeMode === "legacy" && legacyPreparationFailedBeforeDispatch) { + const canRecordPreparation = () => caseHandoffCleanupSettled && legacyPreparationFailedBeforeDispatch && + !legacyAdapterEntered && !nativeDispatchStarted && !nativeSessionResumeScheduled && !nativeWorkspaceFinalizeScheduled && + !nativeOwnershipHeld && !nativeStartupCleanupScheduled && !nativeStartupCleanupError && !failedProcessCancellationSeen && + !failedProcessRunCancellations.has(run.id) && !processRunCancellationSettlements.has(run.id) && + !adapterExecutionControls.has(run.id) && !activeRunExecutions.has(run.id) && !runningProcesses.has(run.id); + if (canRecordPreparation()) await recordLegacyPreparationSettlement(db, latestRun, canRecordPreparation) + .catch(error => logger.warn({ runId: run.id, error }, "Could not record optional legacy preparation settlement")); } if (caseHandoffCleanupSettled && latestRun && !nativeSessionResumeScheduled && !nativeWorkspaceFinalizeScheduled && !nativeOwnershipHeld && !nativeStartupCleanupScheduled && !nativeStartupCleanupError && isHeartbeatRunTerminalStatus(latestRun.status)) { diff --git a/server/src/services/heartbeat/idle-recovery-schedule.test.ts b/server/src/services/heartbeat/idle-recovery-schedule.test.ts index 2bda6ae6c6a..b15327b4327 100644 --- a/server/src/services/heartbeat/idle-recovery-schedule.test.ts +++ b/server/src/services/heartbeat/idle-recovery-schedule.test.ts @@ -10,14 +10,21 @@ import { bindIdleDeadlineOwner } from "../idle-deadline-owner.js"; const now = Date.parse("2099-01-01T00:00:00Z"); const owner = () => ({ revision: "registered:1", reconciled: true }); -function tx(rows: unknown[], onRead: () => void = () => {}) { +function tx(rows: unknown[], onRead: () => void = () => {}, receipts: unknown[] = []) { + let reads = 0; const query = { from: () => query, leftJoin: () => query, where: () => query, orderBy: () => query, - limit: async () => { onRead(); return rows; } }; + limit: async () => { onRead(); return reads++ === 0 ? rows : receipts; } }; return { select: () => query } as unknown as Pick; } function retry() { - return { run: { id: "retry", companyId: "company", agentId: "agent", wakeupRequestId: "wake", status: "scheduled_retry", + return { run: { id: "retry", scopeKind: "issue", issueId: "issue", companyId: "company", agentId: "agent", wakeupRequestId: "wake", status: "scheduled_retry", + runtimeMode: "legacy", nativeIssueId: null, nativeSessionId: null, processPid: null, processGroupId: null, + processStartedAt: null, controllerBootId: null, controllerLeaseExpiresAt: null, hasProcessSettlement: false, scheduledRetryAt: new Date(now + 600_000), startedAt: null, finishedAt: null, costAccountingPending: false, resultJson: null }, + source: { id: "source", companyId: "company", agentId: "agent", scopeKind: "issue", issueId: "issue", + nativeIssueId: null, nativeSessionId: null, runtimeMode: "legacy", status: "failed", settled: true, + hasCancellation: false, processPid: null, processGroupId: null, processStartedAt: null, + controllerBootId: null, controllerLeaseExpiresAt: null, legacyProcessSettlement: null, finishedAt: new Date(now - 1000) }, wake: { id: "wake", runId: "retry", companyId: "company", agentId: "agent", status: "queued", finishedAt: null }, agent: { id: "agent", companyId: "company", status: "idle" }, company: { status: "active" }, issue: { id: "issue", companyId: "company", assigneeAgentId: "agent", assigneeUserId: null, status: "in_progress", @@ -47,6 +54,140 @@ describe("owned recovery deadline inspection", () => { expect((await readIdleRetrySchedule(tx([{ ...retry(), run: { ...row.run, ...patch } }]), now, owner)).blocked).toBe(true); } }); + it("requires an exact settled predecessor for an ordinary issue retry", async () => { + const row = retry(); + expect((await readIdleRetrySchedule(tx([row]), now, owner)).blocked).toBe(false); + const sources = [ + ["missing", null], + ["foreign company", { ...row.source, companyId: "foreign" }], + ["foreign agent", { ...row.source, agentId: "foreign" }], + ["foreign issue", { ...row.source, issueId: "foreign" }], + ["company scope", { ...row.source, scopeKind: "company" }], + ["unknown scope", { ...row.source, scopeKind: "unknown" }], + ["cancelled", { ...row.source, hasCancellation: true }], + ["unsettled", { ...row.source, settled: false }], + ["running", { ...row.source, status: "running" }], + ["native runtime", { ...row.source, runtimeMode: "native" }], + ["native issue", { ...row.source, nativeIssueId: "issue" }], + ["native session", { ...row.source, nativeSessionId: "session" }], + ["retained PID without receipt", { ...row.source, processPid: 42 }], + ["retained birth without receipt", { ...row.source, processStartedAt: new Date(now - 2000) }], + ]; + for (const [label, source] of sources) { + expect((await readIdleRetrySchedule(tx([{ ...row, source }]), now, owner)).blocked, String(label)).toBe(true); + } + }); + it("binds retained process settlement to every source identity field for an ordinary issue retry", async () => { + const row = retry(); + const identity = { ...row.source, processPid: 42, processGroupId: 42, + processStartedAt: new Date(now - 2000), controllerBootId: "boot" }; + const receipt = { schema: "paperclip.legacy-process-settlement.v1", runId: identity.id, + companyId: identity.companyId, agentId: identity.agentId, runtimeMode: "legacy", targetKind: "local", + controllerBootId: identity.controllerBootId, processPid: identity.processPid, processGroupId: identity.processGroupId, + processStartedAt: identity.processStartedAt.toISOString(), finishedAt: identity.finishedAt.toISOString(), status: identity.status }; + const retained = { ...row, source: { ...identity, legacyProcessSettlement: receipt } }; + expect((await readIdleRetrySchedule(tx([retained]), now, owner)).blocked).toBe(false); + for (const patch of [{ schema: "unknown" }, { runId: "foreign" }, { companyId: "foreign" }, + { agentId: "foreign" }, { runtimeMode: "native" }, { targetKind: "remote" }, { controllerBootId: "foreign" }, + { processPid: 43 }, { processGroupId: 43 }, { processStartedAt: new Date(now - 3000).toISOString() }, + { finishedAt: new Date(now - 2000).toISOString() }, { status: "interrupted" }, { extra: true }]) { + expect((await readIdleRetrySchedule(tx([{ ...retained, source: { ...retained.source, + legacyProcessSettlement: { ...receipt, ...patch } } }]), now, owner)).blocked, JSON.stringify(patch)).toBe(true); + } + for (const proof of [null, {}, [], { ...receipt, processPid: "42" }]) { + expect((await readIdleRetrySchedule(tx([{ ...retained, source: { ...retained.source, + legacyProcessSettlement: proof } }]), now, owner)).blocked, JSON.stringify(proof)).toBe(true); + } + for (const patch of [{ processPid: 43 }, { processGroupId: 43 }, { controllerBootId: "claimed" }, + { processStartedAt: new Date(now - 3000) }, { finishedAt: null }, { nativeIssueId: "issue" }, + { nativeSessionId: "session" }, { controllerLeaseExpiresAt: new Date(0) }, + { controllerLeaseExpiresAt: new Date(now + 1000) }]) { + expect((await readIdleRetrySchedule(tx([{ ...retained, source: { ...retained.source, ...patch } }]), now, owner)).blocked, + JSON.stringify(patch)).toBe(true); + } + }); + it("rejects runtime ownership on the future target of an ordinary issue retry", async () => { + const row = retry(); + for (const patch of [{ runtimeMode: "unknown" }, { runtimeMode: null }, { nativeIssueId: "issue" }, + { nativeSessionId: "session" }, { processPid: 42 }, { processPid: 0 }, { processGroupId: 42 }, + { processStartedAt: new Date(now - 2000) }, { controllerBootId: "boot" }, { controllerBootId: "" }, + { controllerLeaseExpiresAt: new Date(0) }, { controllerLeaseExpiresAt: new Date(now + 1000) }, + { hasProcessSettlement: true }]) { + expect((await readIdleRetrySchedule(tx([{ ...row, run: { ...row.run, ...patch } }]), now, owner)).blocked, + JSON.stringify(patch)).toBe(true); + } + }); + it("covers explicit company retries without granting incidental task context authority", async () => { + const row = retry(); + const company = { ...row, issue: null, run: { ...row.run, scopeKind: "company", issueId: null, + runtimeMode: "legacy", scopedContinuation: false, responsibleUserId: "owner", responsibleUserActive: true, agentMemberActive: true }, + source: { id: "source", companyId: "company", agentId: "agent", scopeKind: "company", issueId: null, + nativeIssueId: null, runtimeMode: "legacy", status: "failed", settled: true, hasCancellation: false, + processPid: null, processGroupId: null, processStartedAt: null, controllerBootId: null, controllerLeaseExpiresAt: null, legacyProcessSettlement: null } }; + expect(await readIdleRetrySchedule(tx([company]), now, owner)).toMatchObject({ blocked: false, + runIds: ["retry"], wakeIds: ["wake"], issueWaits: [] }); + for (const patch of [{ scopeKind: "unknown" }, { issueId: "issue" }, { nativeIssueId: "native" }, + { responsibleUserActive: false }, { agentMemberActive: false }, { processPid: 42 }, { scopedContinuation: true }]) { + expect((await readIdleRetrySchedule(tx([{ ...company, run: { ...company.run, ...patch } }]), now, owner)).blocked).toBe(true); + } + for (const patch of [{ companyId: "foreign" }, { agentId: "foreign" }, { settled: false }, + { scopedContinuation: true }, { processPid: 42 }, { scopeKind: "issue" }, { runtimeMode: "native" }, { controllerBootId: "live" }]) { + expect((await readIdleRetrySchedule(tx([{ ...company, source: { ...company.source, ...patch } }]), now, owner)).blocked).toBe(true); + } + }); + it("covers only finished exact represented coalesced receipts for the inspected retry", async () => { + const receipt = { id: "coalesced", runId: "retry", companyId: "company", agentId: "agent", + status: "coalesced", finishedAt: new Date(now - 1000), represented: true }; + expect(await readIdleRetrySchedule(tx([retry()], () => {}, [receipt]), now, owner)).toMatchObject({ + blocked: false, wakeIds: ["wake", "coalesced"] }); + for (const patch of [{ runId: "orphan" }, { companyId: "foreign" }, { agentId: "foreign" }, + { unsupportedContinuation: true }, { represented: false }, { represented: null }, { finishedAt: null }]) { + expect((await readIdleRetrySchedule(tx([retry()], () => {}, [{ ...receipt, ...patch }]), now, owner)).blocked).toBe(true); + } + }); + it("binds a review retry to the exact current stage, participant and safe failed predecessor", async () => { + const row = retry(); + const reviewerAgentId = "11111111-1111-4111-8111-111111111111"; + const stageId = "22222222-2222-4222-8222-222222222222"; + const review = { ...row, run: { ...row.run, agentId: reviewerAgentId, runtimeMode: "legacy", + responsibleUserActive: true, agentMemberActive: true }, wake: { ...row.wake, agentId: reviewerAgentId }, + source: { companyId: "company", agentId: reviewerAgentId, scopeKind: "issue", issueId: "issue", + contextIssueId: "issue", contextStageId: stageId, runtimeMode: "legacy", status: "failed", settled: true, safePreparationFailure: true, + processPid: null, processGroupId: null, processStartedAt: null, controllerBootId: null, controllerLeaseExpiresAt: null, legacyProcessSettlement: null }, + issue: { ...row.issue, status: "in_review", executionState: { status: "pending", currentStageId: stageId, + currentStageIndex: 0, currentStageType: "review", currentParticipant: { type: "agent", agentId: reviewerAgentId, userId: null }, + returnAssignee: null, completedStageIds: [], lastDecisionId: null, lastDecisionOutcome: null }, + executionPolicy: { stages: [{ id: stageId, type: "review", participants: [{ type: "agent", agentId: reviewerAgentId, userId: null }] }] } } }; + expect(await readIdleRetrySchedule(tx([review]), now, owner)).toMatchObject({ blocked: false, + issueWaits: [{ kind: "current_review_retry", issueId: "issue", runId: "retry", reviewerAgentId, stageId }] }); + for (const patch of [{ contextStageId: null }, { contextStageId: "changed" }, { companyId: "foreign" }, + { hasCancellation: true }, { scopeKind: "company" }, { contextIssueId: "foreign" }, { safePreparationFailure: false }, { runtimeMode: "native" }]) { + expect((await readIdleRetrySchedule(tx([{ ...review, source: { ...review.source, ...patch } }]), now, owner)).blocked).toBe(true); + } + for (const patch of [{ currentStageIndex: 1 }, { status: "idle" }, { currentStageId: "changed" }, + { currentStageId: null }, { currentStageId: undefined }, { currentStageId: "" }, + { currentParticipant: { type: "user", userId: "human" } }]) { + expect((await readIdleRetrySchedule(tx([{ ...review, issue: { ...review.issue, + executionState: { ...review.issue.executionState, ...patch } } }]), now, owner)).blocked).toBe(true); + } + expect((await readIdleRetrySchedule(tx([{ ...review, issue: { ...review.issue, + executionPolicy: { stages: [{ type: "review", participants: [{ type: "agent", agentId: reviewerAgentId, userId: null }] }] } } }]), now, owner)).blocked).toBe(true); + const common = { id: "issue", status: "in_review", company_status: "active", scoped: true, + assignee_agent_id: "implementor", assignee_user_id: null, agent_status: "idle", unsafe_dependencies: false, + checkout_run_id: null, execution_run_id: null, conversation_agent_id: null, monitor_wake_requested_at: null, + monitor_next_check_at: null, execution_policy: review.issue.executionPolicy, execution_state: review.issue.executionState, + unblock_descriptor: null, dependency_ready: false, response_driven_wait: false, passive_conversation: false }; + const wait = { kind: "current_review_retry" as const, issueId: "issue", runId: "retry", reviewerAgentId, stageId }; + const read = (rows: unknown[], waits = [wait]) => readIdleIssueSchedule({ execute: async () => rows } as unknown as Pick, "owner", now, waits); + expect((await read([common])).blocked).toBe(false); + expect((await read([common], [])).blocked).toBe(true); + for (const patch of [{ unsafe_dependencies: true }, { scoped: false }, { execution_run_id: "foreign" }, + { checkout_run_id: "foreign" }, { company_status: "paused" }, { execution_policy: null }, + { execution_state: { ...review.issue.executionState, currentStageId: "changed" } }]) { + expect((await read([{ ...common, ...patch }])).blocked).toBe(true); + } + expect((await read([common, { ...common, id: "unknown-sibling", status: "unknown" }])).blocked).toBe(true); + }); it("uses the current agent reviewer rather than the source assignee", async () => { const row = quota(); const reviewer = "11111111-1111-4111-8111-111111111111"; diff --git a/server/src/services/heartbeat/idle-retry-schedule.ts b/server/src/services/heartbeat/idle-retry-schedule.ts index a85880b2e9f..7b002ef3dd2 100644 --- a/server/src/services/heartbeat/idle-retry-schedule.ts +++ b/server/src/services/heartbeat/idle-retry-schedule.ts @@ -1,6 +1,10 @@ +import { legacySourceProcessSettled } from "./legacy-process-settlement.js"; import { parseIssueExecutionState } from "../issue-execution-policy.js"; import { agents, agentWakeupRequests, companies, heartbeatRuns, issues, type Db } from "@paperclipai/db"; -import { eq, sql } from "drizzle-orm"; +import { and, eq, sql, type SQLWrapper } from "drizzle-orm"; +import { issueExecutionPolicySchema } from "@paperclipai/shared"; +import { alias } from "drizzle-orm/pg-core"; +import { DEFERRED_WAKE_CONTEXT_KEY } from "./queue.js"; import { finishedIdleRun } from "../idle-work-proof.js"; import type { DeadlineReconciliationSnapshot } from "../deadline-reconciliation.js"; import { idleDeadlineRowLimit, isFutureIdleDeadline, type IdleDeadlineSnapshot, type IdleIssueWait } from "../idle-deadline-types.js"; @@ -17,13 +21,39 @@ export async function readIdleRetrySchedule(tx: Pick, now: number, const state: unknown[] = [before]; const blocked = (): IdleRetrySchedule => ({ blocked: true, state, deadlines: [], runIds: [], wakeIds: [], issueWaits: [] }); if (!before.reconciled) return blocked(); + const source = alias(heartbeatRuns, "retry_source"); const rows = await tx.select({ run: { - id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + id: heartbeatRuns.id, scopeKind: heartbeatRuns.scopeKind, issueId: heartbeatRuns.issueId, + nativeIssueId: heartbeatRuns.nativeIssueId, nativeSessionId: heartbeatRuns.nativeSessionId, + scopedContinuation: scopedContinuation(heartbeatRuns.contextSnapshot), responsibleUserId: heartbeatRuns.responsibleUserId, + responsibleUserActive: sql`EXISTS (SELECT 1 FROM company_memberships m WHERE m.company_id = ${heartbeatRuns.companyId} + AND m.principal_type = 'user' AND m.principal_id = ${heartbeatRuns.responsibleUserId} AND m.status = 'active')`, + companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + agentMemberActive: sql`EXISTS (SELECT 1 FROM company_memberships m WHERE m.company_id = ${heartbeatRuns.companyId} + AND m.principal_type = 'agent' AND m.principal_id = ${heartbeatRuns.agentId}::text AND m.status = 'active')`, + processPid: heartbeatRuns.processPid, processGroupId: heartbeatRuns.processGroupId, + processStartedAt: heartbeatRuns.processStartedAt, + hasProcessSettlement: sql`${heartbeatRuns.legacyProcessSettlement} IS NOT NULL`, + controllerBootId: heartbeatRuns.controllerBootId, controllerLeaseExpiresAt: heartbeatRuns.controllerLeaseExpiresAt, wakeupRequestId: heartbeatRuns.wakeupRequestId, status: heartbeatRuns.status, scheduledRetryAt: heartbeatRuns.scheduledRetryAt, runtimeMode: heartbeatRuns.runtimeMode, startedAt: heartbeatRuns.startedAt, finishedAt: heartbeatRuns.finishedAt, costAccountingPending: heartbeatRuns.costAccountingPending, resultJson: sql>`jsonb_build_object('startupCancellation', ${heartbeatRuns.resultJson}->'startupCancellation', 'cancellation', ${heartbeatRuns.resultJson}->'cancellation')`, revision: sql`${heartbeatRuns}.xmin::text`, + }, source: { id: source.id, companyId: source.companyId, agentId: source.agentId, + scopedContinuation: scopedContinuation(source.contextSnapshot), + contextIssueId: sql`${source.contextSnapshot}->>'issueId'`, + contextStageId: sql`${source.contextSnapshot}->>'currentStageId'`, + scopeKind: source.scopeKind, issueId: source.issueId, nativeIssueId: source.nativeIssueId, nativeSessionId: source.nativeSessionId, + hasCancellation: sql`COALESCE(${source.resultJson} ?| ARRAY['startupCancellation','cancellation','executionCancellation','nativeCancellation'], false)`, + safePreparationFailure: sql`${source.resultJson}#>>'{executionRecovery,kind}' = 'bootstrap' + AND ${source.resultJson}#>'{executionRecovery,providerWorkStarted}' = 'false'::jsonb`, + status: source.status, runtimeMode: source.runtimeMode, processPid: source.processPid, processGroupId: source.processGroupId, + processStartedAt: source.processStartedAt, startedAt: source.startedAt, executionStage: source.executionStage, + finishedAt: source.finishedAt, legacyProcessSettlement: source.legacyProcessSettlement, + controllerBootId: source.controllerBootId, controllerLeaseExpiresAt: source.controllerLeaseExpiresAt, + settled: sql`${sql.raw(finishedIdleRun("retry_source"))}`, + revision: sql`${source}.xmin::text`, }, wake: { id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, agentId: agentWakeupRequests.agentId, runId: agentWakeupRequests.runId, status: agentWakeupRequests.status, finishedAt: agentWakeupRequests.finishedAt, revision: sql`${agentWakeupRequests}.xmin::text`, @@ -33,10 +63,11 @@ export async function readIdleRetrySchedule(tx: Pick, now: number, status: issues.status, hiddenAt: issues.hiddenAt, executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId, conversationAgentId: issues.conversationAgentId, monitorNextCheckAt: issues.monitorNextCheckAt, monitorWakeRequestedAt: issues.monitorWakeRequestedAt, executionState: issues.executionState, - executionPolicy: sql>`jsonb_build_object('monitor', ${issues.executionPolicy}->'monitor')`, + executionPolicy: sql>`jsonb_build_object('monitor', ${issues.executionPolicy}->'monitor', 'stages', coalesce(${issues.executionPolicy}->'stages', '[]'::jsonb))`, revision: sql`${issues}.xmin::text`, } }) .from(heartbeatRuns) + .leftJoin(source, eq(source.id, heartbeatRuns.retryOfRunId)) .leftJoin(agentWakeupRequests, eq(agentWakeupRequests.id, heartbeatRuns.wakeupRequestId)) .leftJoin(agents, eq(agents.id, heartbeatRuns.agentId)) .leftJoin(companies, eq(companies.id, heartbeatRuns.companyId)) @@ -46,28 +77,101 @@ export async function readIdleRetrySchedule(tx: Pick, now: number, state.push(rows); if (rows.length > idleDeadlineRowLimit) return blocked(); const deadlines: number[] = [], runIds: string[] = [], wakeIds: string[] = [], issueWaits: IdleIssueWait[] = []; - for (const { run, wake, agent, company, issue } of rows) { + for (const { run, source, wake, agent, company, issue } of rows) { const at = run.scheduledRetryAt?.getTime() ?? NaN; const execution = parseIssueExecutionState(issue?.executionState); if (!isFutureIdleDeadline(at, now) || run.startedAt || run.finishedAt || run.costAccountingPending - || run.runtimeMode === "native" || run.resultJson?.startupCancellation || run.resultJson?.cancellation + || run.runtimeMode !== "legacy" || run.nativeIssueId !== null || run.nativeSessionId !== null + || run.processPid !== null || run.processGroupId !== null || run.processStartedAt !== null + || run.controllerBootId !== null || run.controllerLeaseExpiresAt !== null || run.hasProcessSettlement !== false + || run.resultJson?.startupCancellation || run.resultJson?.cancellation || !agent || agent.companyId !== run.companyId || !["idle", "active", "error"].includes(agent.status) || !company || company.status !== "active" || !wake || wake.companyId !== run.companyId || wake.agentId !== run.agentId || wake.runId !== run.id || wake.id !== run.wakeupRequestId || wake.status !== "queued" || wake.finishedAt + ) return blocked(); + if (run.scopeKind === "company") { + if (run.issueId !== null || issue || run.nativeIssueId || run.nativeSessionId || run.runtimeMode !== "legacy" + || !run.responsibleUserId || !run.responsibleUserActive || !run.agentMemberActive + || run.processPid || run.processGroupId || run.controllerBootId || run.controllerLeaseExpiresAt + || !source || source.companyId !== run.companyId || source.agentId !== run.agentId + || source.scopeKind !== "company" || source.issueId !== null || source.nativeIssueId || source.nativeSessionId + || source.runtimeMode !== "legacy" || !source.settled || source.hasCancellation + || !legacySourceProcessSettled(source) + || run.scopedContinuation || source.scopedContinuation) return blocked(); + } else if (issue?.status === "in_review") { + const policy = issueExecutionPolicySchema.safeParse(issue.executionPolicy); + const currentStageId = execution?.currentStageId; + const stage = policy.success ? policy.data.stages.find(stage => stage.id === currentStageId) : undefined; + if (run.scopeKind !== "issue" || run.runtimeMode !== "legacy" || run.nativeIssueId || run.nativeSessionId + || run.processPid || run.processGroupId || run.controllerBootId || run.controllerLeaseExpiresAt + || run.issueId !== issue.id || issue.companyId !== run.companyId + || issue.assigneeUserId || issue.hiddenAt || issue.conversationAgentId + || issue.monitorNextCheckAt || issue.monitorWakeRequestedAt || issue.executionPolicy?.monitor + || !run.responsibleUserActive || !run.agentMemberActive + || !execution || typeof currentStageId !== "string" || currentStageId.length === 0 + || execution.status !== "pending" || execution.currentStageType !== "review" + || execution.currentParticipant?.type !== "agent" || execution.currentParticipant.agentId !== run.agentId + || !stage || stage.type !== "review" || !policy.success + || execution.currentStageIndex !== policy.data.stages.findIndex(candidate => candidate.id === stage.id) + || !stage.participants.some(participant => participant.type === "agent" && participant.agentId === run.agentId) + || !source || source.companyId !== run.companyId || source.agentId !== run.agentId + || source.scopeKind !== "issue" || source.issueId !== issue.id || source.contextIssueId !== issue.id + || source.contextStageId !== stage.id || source.nativeIssueId || source.nativeSessionId + || source.runtimeMode !== "legacy" || !["failed", "timed_out", "interrupted"].includes(source.status ?? "") + || !source.settled || source.hasCancellation || source.safePreparationFailure !== true + || !legacySourceProcessSettled(source) + || (issue.executionRunId !== null && issue.executionRunId !== run.id) + || (issue.checkoutRunId !== null && issue.checkoutRunId !== run.id)) return blocked(); + issueWaits.push({ kind: "current_review_retry", issueId: issue.id, runId: run.id, + reviewerAgentId: run.agentId, stageId: currentStageId }); + } else if (run.scopeKind !== "issue" || run.issueId !== issue?.id + || !source || source.companyId !== run.companyId || source.agentId !== run.agentId + || source.scopeKind !== "issue" || source.issueId !== issue?.id + || source.nativeIssueId !== null || source.nativeSessionId !== null || source.runtimeMode !== "legacy" + || source.settled !== true || source.hasCancellation !== false || !legacySourceProcessSettled(source) || !issue || issue.companyId !== run.companyId || issue.assigneeAgentId !== run.agentId || issue.assigneeUserId || !["todo", "in_progress"].includes(issue.status) || issue.hiddenAt || issue.conversationAgentId || issue.monitorNextCheckAt || issue.monitorWakeRequestedAt || issue.executionPolicy?.monitor || (issue.executionState !== null && (!execution || execution.status !== "idle")) || (issue.executionRunId !== null && issue.executionRunId !== run.id) || (issue.checkoutRunId !== null && issue.checkoutRunId !== run.id)) return blocked(); - deadlines.push(at); runIds.push(run.id); wakeIds.push(wake.id); issueWaits.push({ kind: "scheduled_retry", issueId: issue.id, runId: run.id }); + deadlines.push(at); runIds.push(run.id); wakeIds.push(wake.id); + if (issue && issue.status !== "in_review") issueWaits.push({ kind: "scheduled_retry", issueId: issue.id, runId: run.id }); + } + if (runIds.length) { + const receipts = await tx.select({ id: agentWakeupRequests.id, runId: agentWakeupRequests.runId, + companyId: agentWakeupRequests.companyId, agentId: agentWakeupRequests.agentId, + status: agentWakeupRequests.status, finishedAt: agentWakeupRequests.finishedAt, + unsupportedContinuation: sql`COALESCE((${agentWakeupRequests.payload}->${DEFERRED_WAKE_CONTEXT_KEY}) + ?| ARRAY['interactionId','continuationPolicy','chatThreadId','conversationId','chatCompletionDeliveryIds'], false)`, + represented: sql`jsonb_typeof(${agentWakeupRequests.payload}->${DEFERRED_WAKE_CONTEXT_KEY}) = 'object' + AND (${agentWakeupRequests.payload}->${DEFERRED_WAKE_CONTEXT_KEY}) <@ ${heartbeatRuns.contextSnapshot} + AND (${agentWakeupRequests.payload} - ${DEFERRED_WAKE_CONTEXT_KEY}) <@ ${heartbeatRuns.contextSnapshot}`, + revision: sql`${agentWakeupRequests}.xmin::text`, + }).from(agentWakeupRequests).leftJoin(heartbeatRuns, eq(heartbeatRuns.id, agentWakeupRequests.runId)) + .where(and(eq(heartbeatRuns.status, "scheduled_retry"), eq(agentWakeupRequests.status, "coalesced"))) + .orderBy(agentWakeupRequests.id).limit(idleDeadlineRowLimit + 1); + state.push(receipts); + if (receipts.length > idleDeadlineRowLimit) return blocked(); + const admitted = new Map(rows.map(row => [row.run.id, row.run])); + for (const receipt of receipts) { + const run = receipt.runId ? admitted.get(receipt.runId) : undefined; + if (!run || receipt.companyId !== run.companyId || receipt.agentId !== run.agentId + || !receipt.finishedAt || receipt.unsupportedContinuation || receipt.represented !== true) return blocked(); + wakeIds.push(receipt.id); + } } const after = owner(); if (!after.reconciled || before.revision !== after.revision) return blocked(); return { blocked: false, state, deadlines, runIds, wakeIds, issueWaits }; } +function scopedContinuation(context: SQLWrapper) { + return sql`COALESCE(${context} ?| ARRAY['issueId','taskId','taskKey','nativeIssueId', + 'conversationId','chatThreadId','interactionId','continuationPolicy','chatCompletionDeliveryIds'], false)`; +} + export function idleRetryWorkChecks(snapshot: Pick) { const exclude = (column: string, ids: string[]) => ids.length ? sql` AND ${sql.identifier(column)} NOT IN (${sql.join(ids.map(id => sql`${id}::uuid`), sql`, `)})` : sql``; diff --git a/server/src/services/heartbeat/legacy-process-settlement.test.ts b/server/src/services/heartbeat/legacy-process-settlement.test.ts new file mode 100644 index 00000000000..9c2d9ff334b --- /dev/null +++ b/server/src/services/heartbeat/legacy-process-settlement.test.ts @@ -0,0 +1,100 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Db } from "@paperclipai/db"; +import { legacyControllerBootId, legacyControllerClaim } from "../legacy-controller-lease.js"; +import { legacySourceProcessSettled, recordLegacyProcessSettlement, type LegacyProcessSettlementSource } from "./legacy-process-settlement.js"; +const { pendingCleanup } = vi.hoisted(() => ({ pendingCleanup: vi.fn() })); +vi.mock("../execution-blocker.js", () => ({ hasPendingRunCleanup: pendingCleanup })); +const startedAt = new Date("2099-01-01T00:00:00Z"), finishedAt = new Date("2099-01-01T00:00:01Z"); +function source() { + return { id: "run", companyId: "company", agentId: "agent", runtimeMode: "legacy", status: "failed", + controllerBootId: legacyControllerBootId, controllerLeaseExpiresAt: null, processPid: 42, processGroupId: 42, + processStartedAt: startedAt, finishedAt, legacyProcessSettlement: { schema: "paperclip.legacy-process-settlement.v1", + runId: "run", companyId: "company", agentId: "agent", runtimeMode: "legacy", targetKind: "local", + controllerBootId: legacyControllerBootId, processPid: 42, processGroupId: 42, + processStartedAt: startedAt.toISOString(), finishedAt: finishedAt.toISOString(), status: "failed" } } satisfies LegacyProcessSettlementSource; +} +function database(row: LegacyProcessSettlementSource | null, failure?: Error) { + const update = vi.fn(() => ({ set: vi.fn(() => ({ where: vi.fn(() => ({ returning: async () => { + if (failure) throw failure; return [{ id: row!.id }]; + } })) })) })); + const query = { from: () => query, where: () => query, for: async () => row ? [row] : [] }; + const tx = { select: () => query, update }; + const db = { transaction: async (run: (tx: unknown) => unknown) => run(tx) } as unknown as Db; + return { db, update }; +} +describe("host-owned legacy process settlement", () => { + beforeEach(() => { pendingCleanup.mockReset(); pendingCleanup.mockResolvedValue(false); }); + it("binds every receipt field to the exact retained local process identity", () => { + const run = source(); expect(legacySourceProcessSettled(run)).toBe(true); + for (const [key, value] of Object.entries(run.legacyProcessSettlement!)) { + expect(legacySourceProcessSettled({ ...run, legacyProcessSettlement: { ...run.legacyProcessSettlement, [key]: typeof value === "number" ? value + 1 : "foreign" } })).toBe(false); + } + for (const value of [null, [], {}, true, "closed", { ...run.legacyProcessSettlement, processPid: "42" }, + { ...run.legacyProcessSettlement, extra: true }, { ...run.legacyProcessSettlement, processPid: {} }]) + expect(legacySourceProcessSettled({ ...run, legacyProcessSettlement: value })).toBe(false); + for (const patch of [{ controllerLeaseExpiresAt: new Date(0) }, { controllerLeaseExpiresAt: new Date("2199-01-01") }, + { controllerBootId: "new-claim" }, { processPid: 43 }, { processGroupId: 43 }, { processStartedAt: new Date(0) }, + { finishedAt: new Date(0) }, { processPid: null }, { processGroupId: null }, { processStartedAt: null }, + { runtimeMode: "native" }, { status: "cancelled" }, { status: "running" }]) + expect(legacySourceProcessSettled({ ...run, ...patch })).toBe(false); + }); + it("keeps the never-launched branch closed and rejects stale or partial evidence", () => { + const run = { ...source(), processPid: null, processGroupId: null, processStartedAt: null, + controllerBootId: null, legacyProcessSettlement: null }; + expect(legacySourceProcessSettled(run)).toBe(true); + expect(legacySourceProcessSettled({ ...run, controllerBootId: legacyControllerBootId })).toBe(false); + expect(legacySourceProcessSettled({ ...run, legacyProcessSettlement: source().legacyProcessSettlement })).toBe(false); + }); + it("persists only after the current owner fence and cleanup obligations settle", async () => { + const run = source(), saved = database(run); + expect(await recordLegacyProcessSettlement(saved.db, run, () => true)).toBe(true); + expect(saved.update).toHaveBeenCalledTimes(1); + pendingCleanup.mockResolvedValue(true); const pending = database(run); + expect(await recordLegacyProcessSettlement(pending.db, run, () => true)).toBe(false); + expect(pending.update).not.toHaveBeenCalled(); + pendingCleanup.mockResolvedValue(false); + const held = database(run); let checks = 0; + expect(await recordLegacyProcessSettlement(held.db, run, () => ++checks === 1)).toBe(false); + expect(held.update).not.toHaveBeenCalled(); + expect(await recordLegacyProcessSettlement(database(null).db, run, () => true)).toBe(false); + }); + it("clears a prior receipt on the normal legacy claim", () => { + expect(legacyControllerClaim("legacy")).toMatchObject({ legacyProcessSettlement: null }); + }); + it("does not turn failed optional proof persistence into a certificate", async () => { + const run = { ...source(), legacyProcessSettlement: null }, failed = database(run, new Error("receipt write failed")); + await expect(recordLegacyProcessSettlement(failed.db, run, () => true)).rejects.toThrow("receipt write failed"); + expect(legacySourceProcessSettled(run)).toBe(false); + expect(await recordLegacyProcessSettlement(failed.db, run, () => false)).toBe(false); + }); +}); + +describe("claimed preparation host settlement", () => { + function preparing() { + return { ...source(), processPid: null, processGroupId: null, processStartedAt: null, + startedAt, executionStage: "preparing", legacyProcessSettlement: null as unknown }; + } + function proof() { + const run = preparing(); + return { schema: "paperclip.legacy-preparation-settlement.v1", runId: run.id, companyId: run.companyId, agentId: run.agentId, + runtimeMode: "legacy", targetKind: "not-dispatched", adapterDispatched: false, controllerBootId: run.controllerBootId, + processPid: null, processGroupId: null, processStartedAt: null, executionStage: "preparing", + startedAt: startedAt.toISOString(), finishedAt: finishedAt.toISOString(), status: "failed" }; + } + it("requires exact positive host proof for a claimed source with no process metadata", () => { + const run = preparing(); expect(legacySourceProcessSettled(run)).toBe(false); + expect(legacySourceProcessSettled({ ...run, legacyProcessSettlement: proof() })).toBe(true); + for (const value of [null, {}, [], "bootstrap", source().legacyProcessSettlement, { ...proof(), extra: true }]) + expect(legacySourceProcessSettled({ ...run, legacyProcessSettlement: value })).toBe(false); + }); + it("binds no-dispatch proof to every claimed generation and refuses current dispatch or process ownership", () => { + const run = { ...preparing(), legacyProcessSettlement: proof() }; + for (const [key, value] of Object.entries(proof())) + expect(legacySourceProcessSettled({ ...run, legacyProcessSettlement: { ...proof(), [key]: typeof value === "boolean" ? true : "foreign" } })).toBe(false); + for (const patch of [{ controllerBootId: "new-claim" }, { controllerLeaseExpiresAt: new Date(0) }, + { startedAt: new Date(0) }, { finishedAt: new Date(0) }, { executionStage: "dispatching" }, + { processPid: 42 }, { processGroupId: 42 }, { processStartedAt: startedAt }, { runtimeMode: "native" }, + { status: "running" }, { status: "cancelled" }, { status: "interrupted" }]) + expect(legacySourceProcessSettled({ ...run, ...patch })).toBe(false); + }); +}); diff --git a/server/src/services/heartbeat/legacy-process-settlement.ts b/server/src/services/heartbeat/legacy-process-settlement.ts new file mode 100644 index 00000000000..0c1e155f13c --- /dev/null +++ b/server/src/services/heartbeat/legacy-process-settlement.ts @@ -0,0 +1,124 @@ +import { and, eq, inArray, isNull, sql } from "drizzle-orm"; +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { hasPendingRunCleanup } from "../execution-blocker.js"; +import { legacyControllerBootId } from "../legacy-controller-lease.js"; + +const schema = "paperclip.legacy-process-settlement.v1"; +const statuses = ["failed", "timed_out", "interrupted"]; +export type LegacyProcessSettlementSource = Pick & { legacyProcessSettlement: unknown; + startedAt?: Date | null; executionStage?: string | null }; + +function identityIsComplete(run: LegacyProcessSettlementSource) { + return run.runtimeMode === "legacy" && statuses.includes(run.status) && + typeof run.controllerBootId === "string" && run.controllerBootId.length > 0 && + run.controllerLeaseExpiresAt === null && Number.isSafeInteger(run.processPid) && (run.processPid ?? 0) > 1 && + Number.isSafeInteger(run.processGroupId) && (run.processGroupId ?? 0) > 1 && + run.processStartedAt instanceof Date && Number.isFinite(run.processStartedAt.getTime()) && + run.finishedAt instanceof Date && Number.isFinite(run.finishedAt.getTime()); +} + +function receipt(run: LegacyProcessSettlementSource) { + return { schema, runId: run.id, companyId: run.companyId, agentId: run.agentId, + runtimeMode: "legacy", targetKind: "local", controllerBootId: run.controllerBootId, + processPid: run.processPid, processGroupId: run.processGroupId, + processStartedAt: run.processStartedAt!.toISOString(), finishedAt: run.finishedAt!.toISOString(), status: run.status }; +} + +function preparationIdentityIsComplete(run: LegacyProcessSettlementSource) { + return run.runtimeMode === "legacy" && run.status === "failed" && run.executionStage === "preparing" && + typeof run.controllerBootId === "string" && run.controllerBootId.length > 0 && run.controllerLeaseExpiresAt === null && + run.processPid === null && run.processGroupId === null && run.processStartedAt === null && + run.startedAt instanceof Date && Number.isFinite(run.startedAt.getTime()) && + run.finishedAt instanceof Date && Number.isFinite(run.finishedAt.getTime()) && run.startedAt.getTime() <= run.finishedAt.getTime(); +} + +function preparationReceipt(run: LegacyProcessSettlementSource) { + return { schema: "paperclip.legacy-preparation-settlement.v1", runId: run.id, companyId: run.companyId, agentId: run.agentId, + runtimeMode: "legacy", targetKind: "not-dispatched", adapterDispatched: false, controllerBootId: run.controllerBootId, + processPid: null, processGroupId: null, processStartedAt: null, executionStage: "preparing", + startedAt: run.startedAt!.toISOString(), finishedAt: run.finishedAt!.toISOString(), status: run.status }; +} + +function matchesReceipt(value: unknown, expected: Record): boolean { + if (!value || typeof value !== "object" || Array.isArray(value)) return false; + const proof = value as Record; + return Object.keys(proof).length === Object.keys(expected).length && Object.entries(expected).every(([key, value]) => proof[key] === value); +} + +export function legacySourceProcessSettled(run: LegacyProcessSettlementSource): boolean { + if (run.runtimeMode !== "legacy" || !statuses.includes(run.status) || run.controllerLeaseExpiresAt !== null) return false; + if (run.processPid === null && run.processGroupId === null && run.processStartedAt === null && run.controllerBootId === null) + return run.legacyProcessSettlement === null; + if (preparationIdentityIsComplete(run)) return matchesReceipt(run.legacyProcessSettlement, preparationReceipt(run)); + if (!identityIsComplete(run)) return false; + const value: unknown = run.legacyProcessSettlement; + if (!value || typeof value !== "object" || Array.isArray(value)) return false; + const expected = receipt(run); + const proof = value as Record; + return Object.keys(proof).length === Object.keys(expected).length && + Object.entries(expected).every(([key, value]) => proof[key] === value); +} + +export async function recordLegacyProcessSettlement(db: Db, observed: LegacyProcessSettlementSource, + canCommit: () => boolean): Promise { + if (!canCommit() || observed.controllerBootId !== legacyControllerBootId || !identityIsComplete(observed)) return false; + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const [current] = await tx.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, + controllerBootId: heartbeatRuns.controllerBootId, controllerLeaseExpiresAt: heartbeatRuns.controllerLeaseExpiresAt, + processPid: heartbeatRuns.processPid, processGroupId: heartbeatRuns.processGroupId, + processStartedAt: heartbeatRuns.processStartedAt, finishedAt: heartbeatRuns.finishedAt, + legacyProcessSettlement: heartbeatRuns.legacyProcessSettlement, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, observed.id), eq(heartbeatRuns.companyId, observed.companyId), + eq(heartbeatRuns.agentId, observed.agentId), eq(heartbeatRuns.runtimeMode, "legacy"), + eq(heartbeatRuns.controllerBootId, legacyControllerBootId), isNull(heartbeatRuns.controllerLeaseExpiresAt), + inArray(heartbeatRuns.status, statuses), eq(heartbeatRuns.status, observed.status), + sql`(${heartbeatRuns.scheduledRetryAt} IS NULL OR ${heartbeatRuns.scheduledRetryAt} <= ${heartbeatRuns.finishedAt})`, + eq(heartbeatRuns.finishedAt, observed.finishedAt!), + eq(heartbeatRuns.processPid, observed.processPid!), eq(heartbeatRuns.processGroupId, observed.processGroupId!), + eq(heartbeatRuns.processStartedAt, observed.processStartedAt!), isNull(heartbeatRuns.nativeIssueId), + isNull(heartbeatRuns.nativeSessionId), isNull(heartbeatRuns.executionControlDeadlineAt), + eq(heartbeatRuns.costAccountingPending, false), + sql`NOT COALESCE(${heartbeatRuns.resultJson} ?| ARRAY['startupCancellation','cancellation','executionCancellation','nativeCancellation'], false)`, + )).for("update"); + if (!current || !canCommit() || await hasPendingRunCleanup(tx, current.companyId, current.id) || !canCommit()) return false; + const updated = await tx.update(heartbeatRuns).set({ legacyProcessSettlement: receipt(current) }) + .where(and(eq(heartbeatRuns.id, current.id), eq(heartbeatRuns.companyId, current.companyId))) + .returning({ id: heartbeatRuns.id }); + return updated.length === 1; + }); +} + +export async function recordLegacyPreparationSettlement(db: Db, observed: LegacyProcessSettlementSource, + canCommit: () => boolean): Promise { + if (!canCommit() || observed.controllerBootId !== legacyControllerBootId || !preparationIdentityIsComplete(observed)) return false; + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const [current] = await tx.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, + controllerBootId: heartbeatRuns.controllerBootId, controllerLeaseExpiresAt: heartbeatRuns.controllerLeaseExpiresAt, + processPid: heartbeatRuns.processPid, processGroupId: heartbeatRuns.processGroupId, + processStartedAt: heartbeatRuns.processStartedAt, startedAt: heartbeatRuns.startedAt, finishedAt: heartbeatRuns.finishedAt, + executionStage: heartbeatRuns.executionStage, legacyProcessSettlement: heartbeatRuns.legacyProcessSettlement, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, observed.id), eq(heartbeatRuns.companyId, observed.companyId), + eq(heartbeatRuns.agentId, observed.agentId), eq(heartbeatRuns.runtimeMode, "legacy"), + eq(heartbeatRuns.controllerBootId, legacyControllerBootId), isNull(heartbeatRuns.controllerLeaseExpiresAt), + eq(heartbeatRuns.status, "failed"), eq(heartbeatRuns.executionStage, "preparing"), + eq(heartbeatRuns.startedAt, observed.startedAt!), eq(heartbeatRuns.finishedAt, observed.finishedAt!), + isNull(heartbeatRuns.processPid), isNull(heartbeatRuns.processGroupId), isNull(heartbeatRuns.processStartedAt), + isNull(heartbeatRuns.legacyProcessSettlement), isNull(heartbeatRuns.nativeIssueId), isNull(heartbeatRuns.nativeSessionId), + isNull(heartbeatRuns.executionControlDeadlineAt), eq(heartbeatRuns.costAccountingPending, false), + sql`(${heartbeatRuns.scheduledRetryAt} IS NULL OR ${heartbeatRuns.scheduledRetryAt} <= ${heartbeatRuns.finishedAt})`, + sql`NOT COALESCE(${heartbeatRuns.resultJson} ?| ARRAY['startupCancellation','cancellation','executionCancellation','nativeCancellation'], false)`, + )).for("update"); + if (!current || !preparationIdentityIsComplete(current) || !canCommit() || + await hasPendingRunCleanup(tx, current.companyId, current.id) || !canCommit()) return false; + const updated = await tx.update(heartbeatRuns).set({ legacyProcessSettlement: preparationReceipt(current) }) + .where(and(eq(heartbeatRuns.id, current.id), eq(heartbeatRuns.companyId, current.companyId))) + .returning({ id: heartbeatRuns.id }); + return updated.length === 1; + }); +} diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 28fefdb965b..2ef0c635191 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -589,6 +589,7 @@ export async function persistHeartbeatRunProcessMetadata( const run = await tx .update(heartbeatRuns) .set({ + legacyProcessSettlement: null, processPid: meta.pid, processGroupId: meta.processGroupId, processStartedAt: Number.isNaN(startedAt.getTime()) diff --git a/server/src/services/heartbeat/runtime-selection.ts b/server/src/services/heartbeat/runtime-selection.ts index a4025cbefe7..2cd2ab9bfd2 100644 --- a/server/src/services/heartbeat/runtime-selection.ts +++ b/server/src/services/heartbeat/runtime-selection.ts @@ -977,6 +977,7 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele .update(heartbeatRuns) .set({ runtimeMode: "native", + legacyProcessSettlement: null, controllerBootId: null, controllerLeaseExpiresAt: null, executionStage: null, diff --git a/server/src/services/idle-deadline-types.ts b/server/src/services/idle-deadline-types.ts index 88a0c8ad051..ee2e9c3848c 100644 --- a/server/src/services/idle-deadline-types.ts +++ b/server/src/services/idle-deadline-types.ts @@ -33,6 +33,7 @@ export type IdleDeadlineOwners = Partial, ownerId: st const waitsByIssue = new Map(waits.map(wait => [wait.issueId, wait])); for (const row of rows) { const wait = waitsByIssue.get(row.id); - const retry = wait?.kind === "scheduled_retry" ? wait : undefined; + const reviewRetry = wait?.kind === "current_review_retry" ? wait : undefined; + const retry = wait?.kind === "scheduled_retry" ? wait : reviewRetry; const quota = wait?.kind === "provider_quota" ? wait : undefined; const policy = record(row.execution_policy); const state = record(row.execution_state); + let currentReviewRetry = false; + if (reviewRetry) { + const execution = parseIssueExecutionState(row.execution_state); + const parsedPolicy = issueExecutionPolicySchema.safeParse(row.execution_policy); + const stage = parsedPolicy.success ? parsedPolicy.data.stages.find(stage => stage.id === reviewRetry.stageId) : undefined; + currentReviewRetry = row.status === "in_review" && !row.assignee_user_id + && execution?.status === "pending" && execution.currentStageType === "review" + && execution.currentStageId === reviewRetry.stageId + && execution.currentParticipant?.type === "agent" && execution.currentParticipant.agentId === reviewRetry.reviewerAgentId + && stage?.type === "review" && stage.participants.some(participant => + participant.type === "agent" && participant.agentId === reviewRetry.reviewerAgentId) + && parsedPolicy.success && execution.currentStageIndex === parsedPolicy.data.stages.findIndex(stage => stage.id === reviewRetry.stageId); + if (!currentReviewRetry) return { blocked: true }; + } if (!statuses.has(row.status) || row.company_status !== "active" || row.scoped !== true || (row.assignee_agent_id && !agentStatuses.has(row.agent_status ?? "")) || row.unsafe_dependencies !== false @@ -74,7 +90,7 @@ export async function readIdleIssueSchedule(tx: Pick, ownerId: st || (row.execution_run_id && row.execution_run_id !== retry?.runId) || (row.conversation_agent_id && row.passive_conversation !== true) || row.monitor_wake_requested_at || (row.execution_state !== null && state?.status !== "idle" - && !(quota?.reviewer && row.status === "in_review" && state?.status === "pending")) + && !currentReviewRetry && !(quota?.reviewer && row.status === "in_review" && state?.status === "pending")) || (row.execution_policy !== null && !policy)) return { blocked: true }; if (row.status === "blocked" && row.dependency_ready) return { blocked: true }; if (row.monitor_next_check_at !== null) { @@ -94,7 +110,7 @@ export async function readIdleIssueSchedule(tx: Pick, ownerId: st } if (policy?.monitor != null) return { blocked: true }; const unblock = record(row.unblock_descriptor); - const retryWait = retry && ["todo", "in_progress"].includes(row.status) && Boolean(row.assignee_agent_id) && !row.assignee_user_id; + const retryWait = currentReviewRetry || wait?.kind === "scheduled_retry" && ["todo", "in_progress"].includes(row.status) && Boolean(row.assignee_agent_id) && !row.assignee_user_id; const passive = retryWait || row.passive_conversation === true || row.status === "backlog" || (row.status === "todo" && !row.assignee_agent_id) || (row.status === "todo" && row.assignee_agent_id && row.agent_status === "idle" && row.untouched_onboarding === true) diff --git a/server/src/services/legacy-controller-lease.ts b/server/src/services/legacy-controller-lease.ts index 41a84d7603f..a6a281db059 100644 --- a/server/src/services/legacy-controller-lease.ts +++ b/server/src/services/legacy-controller-lease.ts @@ -30,6 +30,7 @@ export class LegacyControllerLeaseLostError extends Error { export function legacyControllerClaim(runtimeMode: string) { if (runtimeMode === "native") return {}; return { + legacyProcessSettlement: null, controllerBootId: legacyControllerBootId, controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`, executionStage: "preparing", diff --git a/server/src/services/native-runtime/native-idle-sleep-continuation.integration.test.ts b/server/src/services/native-runtime/native-idle-sleep-continuation.integration.test.ts index 1e64b0bc626..2b0662ec512 100644 --- a/server/src/services/native-runtime/native-idle-sleep-continuation.integration.test.ts +++ b/server/src/services/native-runtime/native-idle-sleep-continuation.integration.test.ts @@ -1,3 +1,5 @@ +import { once } from "node:events"; +import { createServer } from "node:http"; import { randomUUID } from "node:crypto"; import { mkdir, readFile, writeFile } from "node:fs/promises"; import { join } from "node:path"; @@ -173,9 +175,13 @@ describeRecoveryPostgres("native idle sleep cold continuation", () => { app.use((req, _res, next) => { req.actor = { type: "board", userId, source: "local_implicit" }; next(); }); app.use("/api", issueRoutes(db, {} as Parameters[1])); app.use(errorHandler); + const commentServer = createServer(app); heartbeat.startTaskDrain({ ttlMs: 60_000 }); try { - await request(app).post(`/api/issues/${issueId}/comments`).send({ body: "Continue the saved work and record the continuation.", reopen: true, resume: true }).expect(201); + commentServer.listen(0, "127.0.0.1"); + await once(commentServer, "listening"); + expect(commentServer.address()).toMatchObject({ address: "127.0.0.1" }); + await request(commentServer).post(`/api/issues/${issueId}/comments`).send({ body: "Continue the saved work and record the continuation.", reopen: true, resume: true }).expect(201); await vi.waitFor(async () => { const wakes = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "issue_reopened_via_comment"))); expect(wakes).toHaveLength(1); @@ -183,7 +189,12 @@ describeRecoveryPostgres("native idle sleep cold continuation", () => { expect(heartbeat.getTaskDrainStatus().pendingWakes).toBe(0); }); expect(turns).toHaveLength(1); - } finally { heartbeat.stopTaskDrain(); } + } finally { + heartbeat.stopTaskDrain(); + if (commentServer.listening) await new Promise((resolve, reject) => { + commentServer.close(error => error ? reject(error) : resolve()); + }); + } await heartbeat.resumeQueuedRuns(); await heartbeat.drainActiveRunExecutions(); await heartbeat.reconcileCostAccounting(); diff --git a/server/src/services/run-secret-redaction.ts b/server/src/services/run-secret-redaction.ts index 597d12a9bd9..306b529f39b 100644 --- a/server/src/services/run-secret-redaction.ts +++ b/server/src/services/run-secret-redaction.ts @@ -52,7 +52,7 @@ export function redactRegisteredSecretValues(input: T, values: string[], cred if (!record) return input; return Object.fromEntries( Object.entries(record) - .filter(([key]) => key !== REGISTRY_KEY && key !== "nativeControllerRecovery" && key !== "nativeTaskSessionPublication") + .filter(([key]) => key !== REGISTRY_KEY && key !== "nativeControllerRecovery" && key !== "nativeTaskSessionPublication" && key !== "legacyProcessSettlement") .map(([key, value]) => [key, key === "credentialAccess" ? (credentialConnectionName ? { connectionName: redactText(credentialConnectionName, values) } : {}) : redactRegisteredSecretValues(value, values, credentialConnectionName)]), diff --git a/tests/e2e/chat-adapters-ui-providers.spec.ts b/tests/e2e/chat-adapters-ui-providers.spec.ts index bd3f627f2bc..d03bf4c9486 100644 --- a/tests/e2e/chat-adapters-ui-providers.spec.ts +++ b/tests/e2e/chat-adapters-ui-providers.spec.ts @@ -386,6 +386,7 @@ test.describe.serial("native chat adapter UI", () => { await expect(page).toHaveURL(rootUrl); await retryConsentWindow.getByRole("link", { name: "Approve installation" }).click(); await expect.poll(() => callbackRequests).toBe(2); + await expect(retryConsentWindow).toHaveURL(rootUrl); await retryConsentWindow.close(); await page.bringToFront(); await expect(page.getByRole("heading", { name: "Send a message to your agent", exact: true })).toBeVisible();