From 96bc021edb6158cfeac6242a418c8775f7f02a15 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 00:56:04 -0600 Subject: [PATCH] fix(server): certify settled legacy processes for future retries (#19377) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Bounded retries must preserve current authority, budgets and deadlines. > - The tenant can wait for supported future work after the source process and its cleanup stop. > - Retained process identity alone cannot prove that cleanup finished. > - This change records host-owned settlement and checks its exact identity before a retry can cover its wait. > - Uncertain process state, changed authority, live claims and cleanup debt keep the tenant awake. ## Linked Issues or Issue Description Refs #19001. Related dependencies: #19019 and #18988. Related #19340 covers explicit Native task Retry; this change covers future Legacy process retries. Current integrated source: `1645970a9322d1213ba25fc7b52366c285682977`, normally merged with actual master `131b566ca4fa866c4348739c75cfe48c805b3398`. The prerequisites above are merged. This PR targets `master`. **What happened?** A future issue, company or review retry can retain its failed source process identity. The deadline reader could not prove that host cleanup finished. Supported waits stayed blocked. Ordinary issue retries also needed an exact source-settlement check, and represented coalesced requests needed exact input coverage. **Expected behavior** A supported future retry covers only its existing wait, under current company and actor authority, after exact host settlement. Missing, malformed, stale, cancelled, Native or unfinished source state stays blocked. No retry gains new execution authority. **Steps to reproduce** Fail an ordinary process or fail environment preparation before adapter entry. Let normal cleanup schedule the existing retry. Inspect its settlement and deadline, stop and reconstruct the owner, and admit one useful successor at the saved deadline. Hold cleanup or change authority to verify that the reader refuses coverage. ## What Changed - Add nullable internal `legacyProcessSettlement` state through normally generated migration `0439_cultured_impossible_man`. No backfill or adapter result creates the record. - Write exact host settlement after stopped-process and cleanup checks. Compare the current run and reject live claims or cleanup debt. Optional persistence failure preserves normal cleanup and keeps sleep coverage blocked. - Add a separate preparation receipt for a claimed Legacy run that failed before any adapter or Native dispatch. Require the same boot, run times, preparing stage, released claim and completed cleanup. A null PID alone proves nothing. - Clear settlement on a new claim, process metadata or Native selection. Omit the internal record from public serialized data. - Check exact source lineage and current authority for ordinary issue, company and taskless retries. Cover completed coalesced requests only when the admitted retry represents their exact inputs. - Add the closed `current_review_retry` wait with current reviewer, stage and existing issue authority checks. Require a nonempty current stage ID. - Add actual producer, cold reconstruction, SQL negative, redaction and optional-persistence controls. - Repair CI fixture expectations for the additive column, startup mock, budget owner and popup return URL. Bind the controlled Native comment server explicitly to loopback. - Replace temporary Slack clock logging with an exact persisted-deadline assertion. Wait until that real deadline before the existing single default poll. Keep both forced polls, business assertions and timing limits. ## Verification - On final integrated source `1645970a`, normal Shared/SDK prerequisites, Runner TypeScript and the entire normal App layered typecheck pass. Tracked source and lockfile remain unchanged. - All 26 whole-file settlement, recovery-reader and bootstrap controls pass. Seven separate real PostgreSQL producer/negative cases pass at `cf16c6a`; each has one selected passing case and 18 filtered registrations. Filtered registrations are not acceptance. - Master then changed the preparation source-lock scope. Its three intersecting cases were rerun on final `1645970a`: an actual failed company process, an actual failed ordinary issue process, and held preparation cleanup. Each preserves its source identity and original deadline and admits one useful successor. Issue and preparation cases exercise normal automatic retry; the company case uses the normal explicit bounded-retry producer. - The unchanged 26 controls and four other PostgreSQL cases retain their `cf16c6a` proof after a scoped source comparison. Optional settlement/preparation persistence failures keep certification blocked while normal cleanup completes. An entered adapter without process identity stays uncertified. This is scoped carry, not a claim that every earlier case ran again on the final head. - All recorded test owners and controlled groups closed. Final read-only SQL audits returned no test sessions. One earlier postcase resource barrier observed a transient unattributed database backend; that database and original evidence remain preserved. Its later disappearance is recovery, not an explained cause. No assertion, deadline or test body was relaxed. - Migration0439 was generated normally after master history. All executable SQL and journal history are preserved; the five retained snapshots are0435–0439. Every Legacy-owned patch and unrelated actual-master blob/mode is preserved. - Published exact head `1645970a9322d1213ba25fc7b52366c285682977` passes all current required checks: 68 successful contexts and eight expected skips after exhaustive pagination and superseded-run accounting. The fresh complete Greptile review scores this exact head 5/5. All four inline review threads are resolved. The latest Ponytail turn stopped without a verdict; one normal retry was requested, with no active turn or new finding observed. No old-head score is used. The original old-base policy failure, its dependent aggregate failure, and the corrected description-gate failure remain recorded. No full local repository suite is claimed; remote typecheck, test and build gates pass. To repeat the real database coverage, set both `PAPERCLIP_RECOVERY_DEADLINES_TEST_DATABASE_URL` and `PAPERCLIP_RECOVERY_DEADLINES_TEST_ADMIN_DATABASE_URL` to the same fresh dedicated test database and run the selected `recovery-deadline-acceptance.test.ts` cases with one worker. An unset URL, skipped case or filtered registration is not acceptance. The fixtures use controlled local processes and real PostgreSQL. Reconstruction creates new service owners inside the same test process. They do not prove deployed Cloud alarms, physical host/database suspension, new-host recovery or Native capacity settlement. The original30-second retry remains below the Cloud270-second minimum wake lead. The merge projection also includes actual master `c3180398faca815fed7a62674bd4b0c4e88884ae`, whose common completion path now sends deferred/recovery notifications. Its unpublished normal merge tree is `16b028d57150f8caa576e913f38343a365b542dd`. Normal Shared/SDK, Runner TypeScript and the full App graph pass on that combined source. One unchanged ordinary-issue acceptance case was rerun there: an actual failed process retains the original automatic retry deadline, admits one useful successor, and passes the settlement, authority and no-duplicate assertions. It passed in 45.055 s; 18 filtered registrations are not acceptance. All known owners and groups closed, and the independent final database audit found zero remaining test sessions. The successful database was removed normally. Recorded worker socket peak is unavailable for this run; parent-process zero values are not used as worker measurements. All 29 Legacy-owned paths and every unrelated master blob and mode are preserved. Two outside-diff review observations concern existing waiting-chat compaction and cross-host Native result-lease recovery. Their files are byte-identical to current master. This focused Legacy change does not fix or qualify those separate paths. ## Risks Settlement proves exact local closure. It does not cover live provider work, Native ownership, cancellation, costs, claims, cleanup debt or changed authority. Independent checks remain. Ship normal migration0439 with the reader and writer. Preserve executable history and the newest five snapshots0435–0439. Generate the MCP child's next migration normally after this parent. The prerequisites are merged; the final remote checks must use this integrated exact head. This proof uses controlled local processes and external PostgreSQL. It does not establish deployed Cloud alarm acceptance, physical host or database suspension, cold host recovery, or R5 Native capacity settlement. The30-second retry is below the Cloud minimum wake lead. This focused repair follows the roadmap's bounded recovery and Cloud deployment work. It adds no new provider policy. ## Model Used OpenAI `gpt-6.1-sol` assisted with readers, fixtures, source review and verification through Codex. A coordinating GPT-6-family agent implemented and reviewed shared lifecycle wiring and integration. Tools included Git, native TypeScript checks and controlled tests. Exact served revisions and context limits were not exposed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used and its available capability details - [x] I have checked ROADMAP.md and related work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have linked existing public work and described the issue - [x] I have not included internal issue IDs or private links - [x] My branch name describes the change - [x] Required focused checks pass at their explicitly recorded source/carry scopes above - [x] I have added or updated meaningful tests - [x] I have recorded the internal contract and migration requirements - [x] I have documented risks and evidence limits - [x] All required published exact-head Paperclip CI gates are green - [x] Published exact-head Greptile is 5/5 with no open findings in this focused diff; separate unchanged master observations are identified above - [x] I will address reviewer findings before requesting merge --------- Co-authored-by: Paperclip Co-authored-by: Codex --- packages/db/src/client.test.ts | 7 +- .../0439_cultured_impossible_man.sql | 1 + ...{0434_snapshot.json => 0439_snapshot.json} | 320 ++++++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/heartbeat_runs.ts | 1 + .../helpers/native-restart-fixture.ts | 6 +- .../helpers/native-restart-workspace.ts | 7 + .../helpers/recovery-deadline-bootstrap.ts | 12 + .../recovery-deadline-ordinary-bootstrap.ts | 19 + .../helpers/recovery-preparation-failure.ts | 85 +++ .../recovery-deadline-acceptance.test.ts | 529 +++++++++++++++++- .../recovery-deadline-bootstrap.test.ts | 36 ++ ...covery-deadline-ordinary-bootstrap.test.ts | 42 ++ .../__tests__/run-secret-redaction.test.ts | 5 + .../server-startup-feedback-export.test.ts | 5 +- server/src/services/heartbeat.ts | 33 +- .../heartbeat/idle-recovery-schedule.test.ts | 147 ++++- .../services/heartbeat/idle-retry-schedule.ts | 116 +++- .../legacy-process-settlement.test.ts | 100 ++++ .../heartbeat/legacy-process-settlement.ts | 124 ++++ .../src/services/heartbeat/run-lifecycle.ts | 1 + .../services/heartbeat/runtime-selection.ts | 1 + server/src/services/idle-deadline-types.ts | 1 + server/src/services/idle-issue-schedule.ts | 24 +- .../src/services/legacy-controller-lease.ts | 1 + ...dle-sleep-continuation.integration.test.ts | 15 +- server/src/services/run-secret-redaction.ts | 2 +- tests/e2e/chat-adapters-ui-providers.spec.ts | 1 + 28 files changed, 1608 insertions(+), 40 deletions(-) create mode 100644 packages/db/src/migrations/0439_cultured_impossible_man.sql rename packages/db/src/migrations/meta/{0434_snapshot.json => 0439_snapshot.json} (99%) create mode 100644 server/src/__tests__/helpers/native-restart-workspace.ts create mode 100644 server/src/__tests__/helpers/recovery-deadline-bootstrap.ts create mode 100644 server/src/__tests__/helpers/recovery-deadline-ordinary-bootstrap.ts create mode 100644 server/src/__tests__/helpers/recovery-preparation-failure.ts create mode 100644 server/src/__tests__/recovery-deadline-bootstrap.test.ts create mode 100644 server/src/__tests__/recovery-deadline-ordinary-bootstrap.test.ts create mode 100644 server/src/services/heartbeat/legacy-process-settlement.test.ts create mode 100644 server/src/services/heartbeat/legacy-process-settlement.ts 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();