From f73ace9f1ae93a6d7975a0807640fa41f899df56 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 23 Oct 2026 14:06:17 -0700 Subject: [PATCH] Clear stale native recovery state so closed work expires (#17359) 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 > - Native runs keep a finalization row, and tasks can keep recovery actions while work is stuck > - Git cache expiration and native harness backup expiration read these rows to decide if closed work is still live > - Commit paths never cleared `native_run_finalizations.recovery_state`, so committed rows kept `resuming_session` forever > - Recovery actions on done or cancelled tasks also stayed active when the close did not go through the issue routes > - So workspaces and backups of tasks that closed weeks ago stayed live and used disk space > - This pull request clears the stale state at commit, backfills old rows, resolves stale recovery actions on closed tasks, and lets the two expiration jobs ignore settled state on closed trees > - The benefit is that closed work expires on schedule, while live leases, unsettled phases, unverified owners, and in-flight system recovery still keep it ## Linked Issues or Issue Description Refs #17248 **What happened?** On one instance, Git cache expiration and native harness backup expiration kept the workspaces and backups of tasks that were done for more than 7 days. Three kinds of stale state caused this: 1. 51 `native_run_finalizations` rows had `phase = 'committed'` and `recovery_state = 'resuming_session'`. The commit writes (status decision commit, attention replay, result persistence, and the finalizer claim) never cleared `recovery_state`. No reader uses `recovery_state` on a row that has a result, except the expiration predicates. 2. Rows with `phase = 'terminal_failure'` and `recovery_state = 'blocked'` (or NULL) are permanent states. Nothing clears them, and continuations are refused for closed tasks. 3. `issue_recovery_actions` with status `active` or `escalated` stayed open after the source task became done or cancelled. Only the issue routes cancel them at close. The periodic sweep `reconcileActiveRecoveryActions` skipped every action unless `wakePolicy.type` was a bounded owner policy or the owner was the board. Agent-owned `resume_native_run` actions (which use `wakePolicy.kind`) and system-owned actions were never resolved. **Expected behavior** When a task tree is closed for the configured time, its Git cache and native backup expire. Only live state keeps them: a held finalization lease, an unsettled finalization phase, an unverified native workspace owner, or system recovery that is scheduled, running, or awaiting evidence. **Steps to reproduce** 1. Run a native task. Let the runner restart so that restart recovery sets `recovery_state = 'resuming_session'` on the finalization row. 2. Let the resumed run commit its result and close the task as done. 3. Wait past the expiration window (or run the sweep with a later clock). 4. See that the finalization row is still `committed` + `resuming_session`, and that Git cache expiration and native backup expiration skip the closed workspace. **Paperclip version or commit** `master` at 837add0caa0 **Deployment mode** Self-hosted server with native runtime enabled **Agent adapter(s) involved** Not adapter-specific (core bug) ## What Changed - Clear `recovery_state` and `recovery_request_id` in every write that persists or commits a native result: `recordCoordinatorDecision` in the status decision committer, the attention replay in `routePersistedNativeResultAttention`, `completeRun` in both `NativeRunCoordinatorStore` and `PaperclipControlPlanePort`, and `claimCoordinator` in the finalizer. - Add data migration `0376_clear_committed_native_recovery_state.sql` with its unchanged schema snapshot (the oldest snapshot is pruned, as the generator does). It clears `recovery_state` and `recovery_request_id` on rows that are `committed` with no lease owner. `recovery_history` keeps the audit trail. - In `reconcileActiveRecoveryActions`, resolve active and escalated actions on done or cancelled tasks (`source_terminal`) before the wake policy and owner filter. The existing owner-unverified guard still runs first. Actions with system recovery in `scheduled`, `running`, or `awaiting_evidence` are skipped. - In `isGitWorkspaceExpirationEligible` (closed-tree mode only), a recovery action blocks only if its cause is `native_workspace_finalization_owner_unverified` or its system recovery is in flight. A finalization blocks only if it has a lease owner or its phase is not `committed` or `terminal_failure`. Idle mode keeps the old strict checks. - In native harness backup expiration, a settled finalization (`committed` or `terminal_failure`, no lease owner) no longer blocks once the session's tasks are closed for the configured days. ## Verification - `pnpm exec vitest run` in `server/` on the touched and related files: `git-workspace-expiration`, `git-workspace-expiration-eligibility`, `native-harness-backup-expiration`, `native-finalization-migration`, `native-interaction-bridge`, `native-run-finalizer-telemetry`, `paperclip-control-plane-port`, `issue-recovery-actions`, `native-status-arbiter-corpus`: 9 files, 321 tests passed. After the tests moved to `native-finalization-recovery-state.test.ts`, that file, `git-workspace-expiration` and `native-run-finalizer-telemetry` passed again (95 tests). - `pnpm exec vitest run src/migration-snapshot-drift.test.ts` in `packages/db`: passed. - `pnpm exec vitest run src/__tests__/heartbeat-process-recovery.test.ts`: 408 tests passed. - `pnpm run typecheck` in `server/`: passed. - `pnpm run check:migrations` in `packages/db`: passed. - New tests: - Result persistence (port and coordinator store), finalizer claim and status decision commit (new focused file `native-finalization-recovery-state.test.ts`), and attention replay each clear `recovery_state` and `recovery_request_id`. - The migration clears committed, unleased rows. It keeps leased committed rows, `observed` rows, and `terminal_failure` rows unchanged, and keeps `recovery_history`. - The sweep resolves an agent-owned `resume_native_run` action and an escalated system action on a done task, and keeps an action with scheduled system recovery. The existing test "retains the exact native owner hold without changing a terminal task's disposition" still covers the owner-unverified hold. - Git expiration eligibility: a closed tree expires with `committed` + `resuming_session`, `terminal_failure` + `blocked`, an active agent recovery action, or an escalated system action. It stays with a leased finalization, an `observed` finalization, an owner-unverified hold, in-flight system recovery, an open tree, and in idle mode. - A full Git expiration of a closed tree with stale native state, then a reopen to the recorded commit. The existing "preserves scheduled runs, unresolved recovery, and live previews" test now uses an `observed` and a leased finalization as the unresolved recovery. - Native backup expiration expires `committed` + `resuming_session` and `terminal_failure` + `blocked`, and keeps an `observed` or leased finalization. - Mutation checks. I reverted each fix on its own, saw the matching test fail, and restored it: - Status decision commit, finalizer claim, port `completeRun`, store `completeRun`, attention replay: each new clearing test failed. - Migration (`WHERE false`): the backfill test failed. - Sweep order (wake policy filter moved back in front) and the in-flight system recovery guard: the sweep test failed. - Git eligibility: the closed-tree action relaxation, the finalization relaxation, the owner-unverified guard, and idle strictness each made the matching eligibility tests fail. - Backup expiration (`recovery_state IS NOT NULL` restored): both new expiry cases failed. - Not run: the full server suite and the e2e suites. CI runs them. ## Risks - The migration is a data-only `UPDATE` on `native_run_finalizations`. It touches only committed rows with no lease owner and a non-null `recovery_state`. The table is not on the known-large list, and the migration safety check passes. `recovery_history` is not changed. - Clearing `recovery_state` at commit and claim: the readers of this column need `phase = 'observed'` (continuation contract), `result_id IS NULL` (recovery service, failed-run retry), or running runs (heartbeat recovery). A row that has a result does not match any of them. The restart recovery summary counts will drop because they no longer count committed rows. - The sweep now resolves more actions on closed tasks. A done or cancelled task cannot continue, so these actions cannot act. Unverified owner holds and in-flight system recovery are kept. - Expiration now removes more closed workspaces and backups. Restore points and Git cache restoration still allow a reopen to rebuild the workspace. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), 1M context, through Claude Code with tool use (shell, file edits, tests). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [ ] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- ..._clear_committed_native_recovery_state.sql | 3 + ...{0371_snapshot.json => 0376_snapshot.json} | 260 +++++++++++++++++- packages/db/src/migrations/meta/_journal.json | 7 + ...t-workspace-expiration-eligibility.test.ts | 60 +++- .../git-workspace-expiration.test.ts | 15 +- .../heartbeat-process-recovery.test.ts | 27 ++ .../native-finalization-migration.test.ts | 44 +++ .../native-harness-backup-expiration.test.ts | 15 +- .../native-interaction-bridge.test.ts | 41 +++ .../src/services/git-workspace-expiration.ts | 10 +- ...native-finalization-recovery-state.test.ts | 126 +++++++++ .../native-harness-backup-expiration.ts | 1 - .../native-interaction-bridge.ts | 2 + .../native-run-coordinator-store.ts | 2 + .../native-runtime/native-run-finalizer.ts | 2 + .../paperclip-control-plane-port.test.ts | 30 ++ .../paperclip-control-plane-port.ts | 2 + .../status-decision-committer.ts | 2 + server/src/services/recovery/service.ts | 27 +- 19 files changed, 656 insertions(+), 20 deletions(-) create mode 100644 packages/db/src/migrations/0376_clear_committed_native_recovery_state.sql rename packages/db/src/migrations/meta/{0371_snapshot.json => 0376_snapshot.json} (99%) create mode 100644 server/src/services/native-runtime/native-finalization-recovery-state.test.ts diff --git a/packages/db/src/migrations/0376_clear_committed_native_recovery_state.sql b/packages/db/src/migrations/0376_clear_committed_native_recovery_state.sql new file mode 100644 index 00000000000..750ed227a75 --- /dev/null +++ b/packages/db/src/migrations/0376_clear_committed_native_recovery_state.sql @@ -0,0 +1,3 @@ +UPDATE "native_run_finalizations" +SET "recovery_state" = NULL, "recovery_request_id" = NULL, "updated_at" = now() +WHERE "phase" = 'committed' AND "lease_owner" IS NULL AND "recovery_state" IS NOT NULL; diff --git a/packages/db/src/migrations/meta/0371_snapshot.json b/packages/db/src/migrations/meta/0376_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0371_snapshot.json rename to packages/db/src/migrations/meta/0376_snapshot.json index 61c248b374c..1015edaf69f 100644 --- a/packages/db/src/migrations/meta/0371_snapshot.json +++ b/packages/db/src/migrations/meta/0376_snapshot.json @@ -1,6 +1,6 @@ { - "id": "7caf727e-0913-40b3-b01c-842e9447666f", - "prevId": "a7c9681c-fb33-47c3-8139-3d88d7bbfe19", + "id": "7832df38-fa1b-4687-bd06-eccbc29bd28c", + "prevId": "3e17203b-ec95-41e9-8363-685982450199", "version": "7", "dialect": "postgresql", "tables": { @@ -28473,6 +28473,250 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.execution_workspace_restore_points": { + "name": "execution_workspace_restore_points", + "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 + }, + "execution_workspace_id": { + "name": "execution_workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "source_issue_id": { + "name": "source_issue_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "generation": { + "name": "generation", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "checkout_name": { + "name": "checkout_name", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "''" + }, + "object_key": { + "name": "object_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "sha256": { + "name": "sha256", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "byte_size": { + "name": "byte_size", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "activated_at": { + "name": "activated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "tree_closed_at": { + "name": "tree_closed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "delete_after": { + "name": "delete_after", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "review_generation": { + "name": "review_generation", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "deleted_at": { + "name": "deleted_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "last_error": { + "name": "last_error", + "type": "text", + "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": { + "execution_workspace_restore_points_object_key_uniq": { + "name": "execution_workspace_restore_points_object_key_uniq", + "columns": [ + { + "expression": "object_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + }, + "execution_workspace_restore_points_live_status_uniq": { + "name": "execution_workspace_restore_points_live_status_uniq", + "columns": [ + { + "expression": "execution_workspace_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "generation", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "checkout_name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"execution_workspace_restore_points\".\"status\" in ('pending', 'active', 'expiring')", + "concurrently": false, + "method": "btree", + "with": {} + }, + "execution_workspace_restore_points_status_delete_after_idx": { + "name": "execution_workspace_restore_points_status_delete_after_idx", + "columns": [ + { + "expression": "status", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "delete_after", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "execution_workspace_restore_points_company_idx": { + "name": "execution_workspace_restore_points_company_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "execution_workspace_restore_points_source_issue_idx": { + "name": "execution_workspace_restore_points_source_issue_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "source_issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "execution_workspace_restore_points_active_source_idx": { + "name": "execution_workspace_restore_points_active_source_idx", + "columns": [ + { + "expression": "source_issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"execution_workspace_restore_points\".\"status\" = 'active'", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.execution_workspace_runtime_leases": { "name": "execution_workspace_runtime_leases", "schema": "", @@ -38627,6 +38871,18 @@ "primaryKey": false, "notNull": false }, + "dismissed_at": { + "name": "dismissed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "dismissed_by_user_id": { + "name": "dismissed_by_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "created_at": { "name": "created_at", "type": "timestamp with time zone", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 801be2f3d07..2d3b7e1d5b0 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2619,6 +2619,13 @@ "when": 1792775097887, "tag": "0375_short_mercury", "breakpoints": true + }, + { + "idx": 376, + "version": "7", + "when": 1792783857131, + "tag": "0376_clear_committed_native_recovery_state", + "breakpoints": true } ] } \ No newline at end of file diff --git a/server/src/__tests__/git-workspace-expiration-eligibility.test.ts b/server/src/__tests__/git-workspace-expiration-eligibility.test.ts index a577f4eee4e..95a98b017a6 100644 --- a/server/src/__tests__/git-workspace-expiration-eligibility.test.ts +++ b/server/src/__tests__/git-workspace-expiration-eligibility.test.ts @@ -1,8 +1,9 @@ import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; -import { agents, companies, createDb, executionWorkspaces, heartbeatRuns, issues, projects, projectWorkspaces } from "@paperclipai/db"; +import { agents, companies, createDb, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, projects, projectWorkspaces } from "@paperclipai/db"; import { isGitWorkspaceExpirationEligible } from "../services/git-workspace-expiration.js"; +import { createSystemRecovery } from "../services/system-recovery-state.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; type Plan = { "Relation Name"?: string; "Actual Loops"?: number; Plans?: Plan[] }; @@ -91,4 +92,61 @@ describeDatabase("Git expiration eligibility", () => { await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: f.old }).where(eq(heartbeatRuns.id, runId)); expect(await f.eligible()).toBe(true); }); + + type StaleState = { finalization?: Partial; action?: Partial }; + async function seedStaleState(f: Awaited>, state: StaleState) { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, status: "succeeded", + finishedAt: f.old, nativeIssueId: f.issueId }); + if (state.finalization) { + await db.insert(nativeRunFinalizations).values({ runId, companyId: f.companyId, issueId: f.issueId, phase: "committed", ...state.finalization }); + } + if (state.action) { + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog", + status: "active", ownerType: "agent", ownerAgentId: f.agentId, cause: "native_finalization_error", fingerprint: randomUUID(), + evidence: { runId }, nextAction: "Resume the native run.", wakePolicy: { kind: "resume_native_run", runId }, ...state.action }); + } + } + + it.each<[string, StaleState]>([ + ["committed resuming_session", { finalization: { recoveryState: "resuming_session", recoveryRequestId: "restart" } }], + ["terminal_failure blocked", { finalization: { phase: "terminal_failure", recoveryState: "blocked" } }], + ["active agent recovery", { action: {} }], + ["escalated system recovery", { action: { status: "escalated", ownerType: "system", ownerAgentId: null } }], + ])("expires a closed tree despite stale %s", async (_label, state) => { + const f = await fixture(); + await seedStaleState(f, state); + expect(await f.eligible()).toBe(true); + }); + + it.each<[string, StaleState]>([ + ["a leased finalization", { finalization: { leaseOwner: "native-finalizer:live", recoveryState: "resuming_session" } }], + ["an observed finalization", { finalization: { phase: "observed", recoveryState: "resuming_session" } }], + ["an unverified owner hold", { action: { ownerType: "board", ownerAgentId: null, cause: "native_workspace_finalization_owner_unverified", wakePolicy: null } }], + ["in-flight system recovery", { action: { ownerType: "system", ownerAgentId: null, evidence: { systemRecovery: createSystemRecovery(randomUUID()) } } }], + ])("keeps a closed tree with %s", async (_label, state) => { + const f = await fixture(); + await seedStaleState(f, state); + expect(await f.eligible()).toBe(false); + }); + + it("keeps an open tree despite settled native state", async () => { + const f = await fixture(); + await seedStaleState(f, { finalization: { recoveryState: "resuming_session" } }); + await db.update(issues).set({ status: "in_progress", completedAt: null }).where(eq(issues.id, f.issueId)); + expect(await f.eligible()).toBe(false); + }); + + it.each<[string, StaleState]>([ + ["committed resuming_session", { finalization: { recoveryState: "resuming_session" } }], + ["terminal_failure blocked", { finalization: { phase: "terminal_failure", recoveryState: "blocked" } }], + ["active agent recovery", { action: {} }], + ])("keeps idle-mode expiration strict for %s", async (_label, state) => { + const f = await fixture(); + const idleCutoff = new Date(Date.now() + 60_000); + const idleEligible = () => db.transaction(tx => isGitWorkspaceExpirationEligible(tx, f.workspace, f.now, idleCutoff)); + expect(await idleEligible()).toBe(true); + await seedStaleState(f, state); + expect(await idleEligible()).toBe(false); + }); }); diff --git a/server/src/__tests__/git-workspace-expiration.test.ts b/server/src/__tests__/git-workspace-expiration.test.ts index 9e470c3de59..b023805c47a 100644 --- a/server/src/__tests__/git-workspace-expiration.test.ts +++ b/server/src/__tests__/git-workspace-expiration.test.ts @@ -8,7 +8,7 @@ import { pathToFileURL } from "node:url"; import { createRequire } from "node:module"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; -import { companies, projects, projectWorkspaces, issues, executionWorkspaces, agents, heartbeatRuns, nativeRunFinalizations, workspaceRuntimeServices, executionWorkspaceRestorePoints, createDb } from "@paperclipai/db"; +import { companies, projects, projectWorkspaces, issues, executionWorkspaces, agents, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, workspaceRuntimeServices, executionWorkspaceRestorePoints, createDb } from "@paperclipai/db"; import { markGitRestorePointExpired } from "../services/git-restore-point-storage.js"; import { GIT_RESTORE_POINT_EXPIRED_WARNING } from "../services/execution-workspaces.js"; import { getStorageService } from "../storage/index.js"; @@ -922,8 +922,9 @@ describeEmbeddedPostgres("reconstructable Git expiration", () => { await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId, nativeIssueId: f.issueId, status: "scheduled_retry" }); expect((await f.expire()).expired).toBe(false); await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); - await db.insert(nativeRunFinalizations).values({ runId, companyId: f.companyId, issueId: f.issueId, phase: "terminal_failure", recoveryState: "blocked" }); + await db.insert(nativeRunFinalizations).values({ runId, companyId: f.companyId, issueId: f.issueId, phase: "observed", recoveryState: "resuming_session" }); expect((await f.expire()).expired).toBe(false); + await db.update(nativeRunFinalizations).set({ phase: "committed", leaseOwner: "native-finalizer:live" }).where(eq(nativeRunFinalizations.runId, runId)); expect((await f.expire()).expired).toBe(false); await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); await db.insert(workspaceRuntimeServices).values({ id: randomUUID(), companyId: f.companyId, executionWorkspaceId: f.workspaceId, scopeType: "execution_workspace", serviceName: "Preview", status: "running", lifecycle: "persistent", provider: "local" }); @@ -931,6 +932,16 @@ describeEmbeddedPostgres("reconstructable Git expiration", () => { expect(await exists(f.cwd)).toBe(true); }); + it("expires a closed tree whose native recovery state was left behind", async () => { + const f = await fixture(), agentId = randomUUID(), runIds = [randomUUID(), randomUUID()]; + await db.insert(agents).values({ id: agentId, companyId: f.companyId, name: "Fixture", role: "engineer", adapterType: "process" }); + await db.insert(heartbeatRuns).values(runIds.map(id => ({ id, companyId: f.companyId, agentId, nativeIssueId: f.issueId, status: "succeeded", finishedAt: new Date(Date.now() - 48 * 3_600_000) }))); + await db.insert(nativeRunFinalizations).values([["committed", "resuming_session"], ["terminal_failure", "blocked"]].map(([phase, recoveryState], i) => ({ runId: runIds[i]!, companyId: f.companyId, issueId: f.issueId, phase, recoveryState }))); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog", ownerAgentId: agentId, + cause: "native_finalization_error", fingerprint: randomUUID(), nextAction: "Resume the native run.", wakePolicy: { kind: "resume_native_run", runId: runIds[0] } }); + expect(await f.expire()).toMatchObject({ expired: true }); expect(await exists(f.cwd)).toBe(false); + }); + it("skips deletion when authenticated upstream verification fails", async () => { const f = await fixture(); await fs.rename(f.remote, `${f.remote}.offline`); diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index d0d98e6a52c..3af86f73e77 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1,5 +1,6 @@ import { enableRunFailureReporting } from "./helpers/run-failure-reporting.js"; import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; +import { createSystemRecovery } from "../services/system-recovery-state.js"; import * as conversationContinuation from "../services/conversation-continuation.js"; import { classifyEnvironmentCapabilities, environmentRuntimeService } from "../services/environment-runtime.js"; import * as workspaceRuntime from "../services/workspace-runtime.js"; @@ -2643,6 +2644,32 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { await heartbeat.drainActiveRunExecutions(); }); + it("resolves stale recovery on terminal tasks unless system recovery is still in flight", async () => { + const f = await seedRunFixture({ runStatus: "failed" }); + const seedTerminalAction = async (action: Partial) => { + const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Closed recovery source", + status: "done", completedAt: new Date(), assigneeAgentId: f.agentId }).returning(); + const [row] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: issue.id, + kind: "active_run_watchdog", status: "active", ownerType: "agent", ownerAgentId: f.agentId, + cause: "native_finalization_error", fingerprint: randomUUID(), evidence: { runId: f.runId }, + nextAction: "Resume the native run.", ...action }).returning(); + return row; + }; + const agentNative = await seedTerminalAction({ wakePolicy: { kind: "resume_native_run", runId: f.runId } }); + const systemEscalated = await seedTerminalAction({ status: "escalated", ownerType: "system", ownerAgentId: null }); + const systemInFlight = await seedTerminalAction({ ownerType: "system", ownerAgentId: null, + evidence: { runId: f.runId, systemRecovery: createSystemRecovery(f.runId) } }); + + await heartbeatService(db).reconcileStrandedAssignedIssues(); + + for (const action of [agentNative, systemEscalated]) { + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)))[0]) + .toMatchObject({ status: "resolved", outcome: "restored", resolutionNote: "source_terminal" }); + } + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, systemInFlight.id)))[0]) + .toMatchObject({ status: "active", resolvedAt: null }); + }); + it.each(["settled", "rejected", "late_callback", "provider_transport_failed"] as const)( "joins startup and reap retained cleanup, keeps recovery live, and drains its %s operation", async (outcome) => { diff --git a/server/src/__tests__/native-finalization-migration.test.ts b/server/src/__tests__/native-finalization-migration.test.ts index 6ba1a2cbd24..050be89dfa5 100644 --- a/server/src/__tests__/native-finalization-migration.test.ts +++ b/server/src/__tests__/native-finalization-migration.test.ts @@ -3,10 +3,14 @@ import { readFile } from "node:fs/promises"; import { describe, expect, it } from "vitest"; import { eq, sql } from "drizzle-orm"; import { + agents, applyPendingMigrations, + companies, createDb, heartbeatRunEvents, heartbeatRuns, + issues, + nativeRunFinalizations, } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -129,4 +133,44 @@ describe("P6-18 / MIG-01..04 native finalization migration", () => { await temporary.cleanup(); } }, 60_000); + + it("clears recovery state left on committed, unleased finalizations", async () => { + const temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-recovery-backfill-"); + const migration = await readFile( + new URL("../../../packages/db/src/migrations/0376_clear_committed_native_recovery_state.sql", import.meta.url), + "utf8", + ); + const db = createDb(temporary.connectionString); + try { + const [company] = await db.insert(companies).values({ name: "Backfill", issuePrefix: "BKF" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company!.id, name: "Backfill agent" }).returning(); + const [issue] = await db.insert(issues).values({ companyId: company!.id, title: "Backfill", status: "done" }).returning(); + const history = [{ disposition: "resume_dead_runner" }]; + const fixtures = [ + { key: "committed", phase: "committed", leaseOwner: null, recoveryState: "resuming_session" }, + { key: "leased", phase: "committed", leaseOwner: "native-finalizer:live", recoveryState: "resuming_session" }, + { key: "observed", phase: "observed", leaseOwner: null, recoveryState: "resuming_session" }, + { key: "terminal", phase: "terminal_failure", leaseOwner: null, recoveryState: "blocked" }, + ]; + const runIds = new Map(); + for (const fixture of fixtures) { + const [run] = await db.insert(heartbeatRuns).values({ + companyId: company!.id, agentId: agent!.id, status: "succeeded", runtimeMode: "native", nativeIssueId: issue!.id, + }).returning(); + runIds.set(fixture.key, run!.id); + await db.insert(nativeRunFinalizations).values({ + runId: run!.id, companyId: company!.id, issueId: issue!.id, phase: fixture.phase, leaseOwner: fixture.leaseOwner, + recoveryState: fixture.recoveryState, recoveryRequestId: "restart-request", recoveryHistory: history, + }); + } + for (const statement of migration.split("--> statement-breakpoint")) await db.execute(sql.raw(statement)); + const rows = new Map((await db.select().from(nativeRunFinalizations)).map((row) => [row.runId, row])); + expect(rows.get(runIds.get("committed")!)).toMatchObject({ recoveryState: null, recoveryRequestId: null, recoveryHistory: history }); + expect(rows.get(runIds.get("leased")!)).toMatchObject({ recoveryState: "resuming_session", recoveryRequestId: "restart-request" }); + expect(rows.get(runIds.get("observed")!)).toMatchObject({ recoveryState: "resuming_session", recoveryRequestId: "restart-request" }); + expect(rows.get(runIds.get("terminal")!)).toMatchObject({ recoveryState: "blocked", recoveryRequestId: "restart-request" }); + } finally { + await temporary.cleanup(); + } + }, 60_000); }); diff --git a/server/src/__tests__/native-harness-backup-expiration.test.ts b/server/src/__tests__/native-harness-backup-expiration.test.ts index 1efb58a6154..b6330409e72 100644 --- a/server/src/__tests__/native-harness-backup-expiration.test.ts +++ b/server/src/__tests__/native-harness-backup-expiration.test.ts @@ -66,7 +66,7 @@ describeEmbeddedPostgres("native harness backup expiration", () => { issueStatus?: string; closedDaysAgo?: number; runStatus?: string; - finalization?: { phase: string; recoveryState?: string | null }; + finalization?: { phase: string; recoveryState?: string | null; leaseOwner?: string | null }; leaseStatus?: string; backupAgeDays?: number; cleanupActive?: boolean; @@ -110,6 +110,7 @@ describeEmbeddedPostgres("native harness backup expiration", () => { issueId: issue!.id, phase: input.finalization.phase, recoveryState: input.finalization.recoveryState ?? null, + leaseOwner: input.finalization.leaseOwner ?? null, }); } const scope = createHash("sha256").update(randomUUID()).digest("hex"); @@ -172,7 +173,8 @@ describeEmbeddedPostgres("native harness backup expiration", () => { ["recently_closed", { closedDaysAgo: 2 }], ["live_run", { runStatus: "running" }], ["pending_finalization", { finalization: { phase: "settling" } }], - ["pending_finalization", { finalization: { phase: "committed", recoveryState: "resuming_session" } }], + ["pending_finalization", { finalization: { phase: "observed", recoveryState: "resuming_session" } }], + ["pending_finalization", { finalization: { phase: "committed", leaseOwner: "controller" } }], ["live_lease", { leaseStatus: "released" }], ["recent_backup", { backupAgeDays: 1 }], ["cleanup_active", { cleanupActive: true }], @@ -217,6 +219,15 @@ describeEmbeddedPostgres("native harness backup expiration", () => { expect(await exists(session.current)).toBe(false); }); + it.each([ + ["committed", "resuming_session"], + ["terminal_failure", "blocked"], + ] as const)("stale %s recovery state %s does not block expiry once the issue is closed", async (phase, recoveryState) => { + const session = await seedSession({ finalization: { phase, recoveryState } }); + expect(await service().tick()).toMatchObject({ expired: 1 }); + expect(await exists(session.current)).toBe(false); + }); + it("reports candidates without deleting anything in dry run", async () => { const session = await seedSession(); const result = await nativeHarnessBackupExpirationService(db, { diff --git a/server/src/__tests__/native-interaction-bridge.test.ts b/server/src/__tests__/native-interaction-bridge.test.ts index 576885a19f9..46433f52b90 100644 --- a/server/src/__tests__/native-interaction-bridge.test.ts +++ b/server/src/__tests__/native-interaction-bridge.test.ts @@ -12,6 +12,7 @@ import { issueThreadInteractions, issues, nativeRunFinalizations, + nativeRunResults, statusDecisionEffects, statusDecisions, workAssessments, @@ -22,6 +23,7 @@ import { materializeLegacyQuestionResponseWakeProjection, materializeNativeInteractionResponses, NativeInteractionBridgeError, + routePersistedNativeResultAttention, } from "../services/native-runtime/native-interaction-bridge.js"; import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js"; @@ -821,4 +823,43 @@ describe("P6-19 native interaction bridge", () => { .from(issueThreadInteractions) .where(eq(issueThreadInteractions.id, outsideInteractionId))).resolves.toEqual([{ summary: null }]); }, 30_000); + + it("clears stale recovery state when replaying a committed attention decision", async () => { + const localIssueId = randomUUID(); + const localRunId = randomUUID(); + const localContractId = randomUUID(); + const resultId = randomUUID(); + const assessmentId = randomUUID(); + const requestId = "attention:replayed"; + await db.insert(issues).values({ id: localIssueId, companyId, title: "Replayed native attention", + status: "in_progress", assigneeAgentId: agentId, workMode: "standard" }); + await db.insert(completionContracts).values({ id: localContractId, companyId, issueId: localIssueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "phase6-v1", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "replay-v1", objective: "Replay attention", criteria: [{ id: "objective", requirement: "Replay" }] }, + canonicalSha256: `replay-contract:${localContractId}`, createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: localRunId, companyId, agentId, status: "succeeded", runtimeMode: "native", + nativeIssueId: localIssueId, nativeSessionId: localRunId, completionContractId: localContractId, + completionContractSha256: `replay-contract:${localContractId}`, contextSnapshot: { issueId: localIssueId } }); + await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId: localIssueId, runId: localRunId, + completionContractId: localContractId, serverFingerprint: `fp-${resultId}`, schemaStatus: "accepted", + resultJson: { result: { attentionRequests: [{ id: requestId, requestedCapability: "context_lookup", summary: "Look up context" }] } }, + canonicalSha256: `sha-${resultId}` }); + await db.insert(workAssessments).values({ id: assessmentId, companyId, issueId: localIssueId, runId: localRunId, + contractId: localContractId, resultId, triggerKind: "native_attention", triggerRef: requestId, + triggerActorCompanyId: companyId, priorIssueStatus: "in_progress", priorStatusVersion: 0, + policyVersion: "phase6-v1", assessmentJson: {}, inputDigest: `digest-${assessmentId}` }); + const [decision] = await db.insert(statusDecisions).values({ companyId, issueId: localIssueId, runId: localRunId, + assessmentId, decisionVersion: 1, policyVersion: "phase6-v1", fromStatus: "in_progress", toStatus: "in_progress", + reasonCode: "attention_context", decisionJson: { effects: [] }, decisionDigest: `digest-${assessmentId}` }).returning(); + await db.insert(nativeRunFinalizations).values({ runId: localRunId, companyId, issueId: localIssueId, + phase: "assessing", resultId, recoveryState: "resuming_session", recoveryRequestId: "restart-request" }); + + const receipts = await routePersistedNativeResultAttention({ db, runId: localRunId }); + + expect(receipts).toEqual([expect.objectContaining({ requestId, decisionId: decision!.id, replayed: true })]); + await expect(db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, localRunId))).resolves.toEqual([ + expect.objectContaining({ phase: "committed", decisionId: decision!.id, recoveryState: null, recoveryRequestId: null }), + ]); + }); }); diff --git a/server/src/services/git-workspace-expiration.ts b/server/src/services/git-workspace-expiration.ts index d60583d9c2a..c8e4bf2cda4 100644 --- a/server/src/services/git-workspace-expiration.ts +++ b/server/src/services/git-workspace-expiration.ts @@ -145,6 +145,11 @@ export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Worksp OR coalesce(CASE WHEN status = 'done' THEN completed_at ELSE cancelled_at END, updated_at) > ${new Date(now.getTime() - DAY_MS).toISOString()}::timestamptz) AND NOT EXISTS (SELECT 1 FROM ${issues} WHERE company_id = ${workspace.companyId} AND execution_workspace_id = ${workspace.id} AND status NOT IN ('done', 'cancelled'))`; + const recoveryActionBlocks = idleCutoff ? sql`true` : sql`(cause = 'native_workspace_finalization_owner_unverified' + OR evidence->'systemRecovery'->>'state' IN ('scheduled', 'running', 'awaiting_evidence'))`; + const finalizationBlocks = idleCutoff + ? sql`(f.phase <> 'committed' OR f.lease_owner IS NOT NULL OR f.recovery_state IS NOT NULL)` + : sql`(f.lease_owner IS NOT NULL OR f.phase NOT IN ('committed', 'terminal_failure'))`; const rows = await tx.execute<{ allowed: boolean }>(sql`${expirationScope(workspace)} SELECT CASE WHEN EXISTS (SELECT 1 FROM ${executionWorkspaces} other WHERE other.id <> ${workspace.id} @@ -158,10 +163,11 @@ export async function isGitWorkspaceExpirationEligible(tx: Tx, workspace: Worksp AND NOT EXISTS (SELECT 1 FROM relevant_runs WHERE status NOT IN ('succeeded', 'failed', 'timed_out', 'cancelled', 'interrupted') OR finished_at IS NULL OR controller_lease_expires_at > now()) AND NOT EXISTS (SELECT 1 FROM ${issueRecoveryActions} WHERE company_id = ${workspace.companyId} - AND source_issue_id IN (SELECT id FROM tree) AND status IN ('active', 'escalated')) + AND source_issue_id IN (SELECT id FROM tree) AND status IN ('active', 'escalated') + AND ${recoveryActionBlocks}) AND NOT EXISTS (SELECT 1 FROM ${nativeRunFinalizations} f WHERE f.company_id = ${workspace.companyId} AND (f.issue_id IN (SELECT id FROM tree) OR f.run_id IN (SELECT id FROM relevant_runs)) - AND (f.phase <> 'committed' OR f.lease_owner IS NOT NULL OR f.recovery_state IS NOT NULL)) + AND ${finalizationBlocks}) AND NOT EXISTS (SELECT 1 FROM ${environmentLeases} l WHERE l.company_id = ${workspace.companyId} AND (l.execution_workspace_id = ${workspace.id} OR l.heartbeat_run_id IN (SELECT id FROM relevant_runs)) AND (l.status = 'pending_cleanup' OR l.cleanup_status = 'failed' OR l.metadata ? 'nativeWorkspaceExportResume' diff --git a/server/src/services/native-runtime/native-finalization-recovery-state.test.ts b/server/src/services/native-runtime/native-finalization-recovery-state.test.ts new file mode 100644 index 00000000000..f6844cd70c1 --- /dev/null +++ b/server/src/services/native-runtime/native-finalization-recovery-state.test.ts @@ -0,0 +1,126 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + agents, + companies, + completionContracts, + createDb, + heartbeatRuns, + issues, + nativeRunFinalizations, + nativeRunResults, + workAssessments, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "../../__tests__/helpers/embedded-postgres.js"; +import { + CONTROL_PLANE_CONFORMANCE_RESULT, + CONTROL_PLANE_CONFORMANCE_TERMINAL, +} from "../../vendor/paperclip-runner/testing.js"; +import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; +import { finalizeNativeRun } from "./native-run-finalizer.js"; +import { commitNativeStatusDecision } from "./status-decision-committer.js"; +import { NATIVE_STATUS_ARBITER_POLICY_VERSION } from "./status-arbiter.js"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres native finalization recovery state tests on this host: ${ + embeddedPostgresSupport.reason ?? "unsupported environment" + }`, + ); +} + +describeEmbeddedPostgres("native finalization recovery state", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + const companyId = randomUUID(); + const agentId = randomUUID(); + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-native-recovery-state-"); + db = createDb(tempDb.connectionString); + await db.insert(companies).values({ id: companyId, name: "Recovery state", issuePrefix: "RST" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Native worker", adapterType: "codex_local", status: "running" }); + }, 30_000); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seedNativeRun() { + const issueId = randomUUID(), contractId = randomUUID(), runId = randomUUID(), runnerInstanceId = randomUUID(); + const contractSha256 = `contract-${contractId}`; + await db.insert(issues).values({ id: issueId, companyId, title: "Native recovery state fixture", + status: "in_progress", assigneeAgentId: agentId, workMode: "standard" }); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "recovery-state-v1", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "recovery-state-v1", objective: "Finish", criteria: [{ id: "objective", requirement: "Complete the task" }] }, + canonicalSha256: contractSha256, createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", + nativeIssueId: issueId, nativeSessionId: runId, runnerInstanceId, completionContractId: contractId, + completionContractSha256: contractSha256, contextSnapshot: { issueId } }); + return { issueId, contractId, contractSha256, runId, runnerInstanceId }; + } + + async function completeResult(fixture: Awaited>) { + const port = new PaperclipControlPlanePort(db, { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, + sessionId: fixture.runId, completionContractId: fixture.contractId, completionContractSha256: fixture.contractSha256, + sourceInstanceId: fixture.runnerInstanceId, controlPlaneSourceInstanceId: "recovery-state-test" }); + await port.openRun({ identity: { runId: fixture.runId, sessionId: fixture.runId, companyId, issueId: fixture.issueId, agentId }, + backendKind: "mock", sourceInstanceId: fixture.runnerInstanceId }); + await port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, + callerResultId: `${fixture.runId}:result` }); + } + + async function readFinalization(runId: string) { + return (await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0]; + } + + it("clears stale recovery state when the finalizer claims a persisted result", async () => { + const fixture = await seedNativeRun(); + await completeResult(fixture); + await db.update(nativeRunFinalizations).set({ recoveryState: "resuming_session", recoveryRequestId: "restart-request" }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", failpoint: "status_projection" }) + .catch(() => undefined); + expect(await readFinalization(fixture.runId)).toMatchObject({ + phase: "retryable_failure", recoveryState: null, recoveryRequestId: null, + }); + }); + + it("clears stale recovery state when a status decision commits", async () => { + const fixture = await seedNativeRun(); + const resultId = randomUUID(), assessmentId = randomUUID(); + await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId: fixture.issueId, runId: fixture.runId, + completionContractId: fixture.contractId, serverFingerprint: `fp-${resultId}`, schemaStatus: "accepted", + resultJson: {}, canonicalSha256: `sha-${resultId}` }); + await db.insert(workAssessments).values({ id: assessmentId, companyId, issueId: fixture.issueId, runId: fixture.runId, + contractId: fixture.contractId, resultId, triggerKind: "native_result", triggerActorCompanyId: companyId, + priorIssueStatus: "in_progress", priorStatusVersion: 0, policyVersion: "recovery-state-v1", assessmentJson: {}, + inputDigest: `digest-${assessmentId}` }); + await db.insert(nativeRunFinalizations).values({ runId: fixture.runId, companyId, issueId: fixture.issueId, + phase: "arbitrating", resultId, recoveryState: "resuming_session", recoveryRequestId: "restart-request" }); + await commitNativeStatusDecision({ + db, companyId, issueId: fixture.issueId, runId: fixture.runId, assessmentId, + priorStatus: "in_progress", priorStatusVersion: 0, priorDecisionId: null, + decision: { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "cancelled", + toStatus: "cancelled", + reasonCode: "cancellation_issue_authorized", + unblockDescriptor: null, + effects: [{ kind: "release_checkout" }, { kind: "cancel_continuations" }], + }, + }); + expect(await readFinalization(fixture.runId)).toMatchObject({ + phase: "committed", recoveryState: null, recoveryRequestId: null, + }); + }); +}); diff --git a/server/src/services/native-runtime/native-harness-backup-expiration.ts b/server/src/services/native-runtime/native-harness-backup-expiration.ts index d87789f0471..448997670e1 100644 --- a/server/src/services/native-runtime/native-harness-backup-expiration.ts +++ b/server/src/services/native-runtime/native-harness-backup-expiration.ts @@ -180,7 +180,6 @@ async function sessionRetentionBlocker( inArray(nativeRunFinalizations.runId, runIds), or( notInArray(nativeRunFinalizations.phase, SETTLED_FINALIZATION_PHASES), - sql`${nativeRunFinalizations.recoveryState} is not null`, sql`${nativeRunFinalizations.leaseOwner} is not null`, ), ), diff --git a/server/src/services/native-runtime/native-interaction-bridge.ts b/server/src/services/native-runtime/native-interaction-bridge.ts index 42e4d9b06b9..0320564677e 100644 --- a/server/src/services/native-runtime/native-interaction-bridge.ts +++ b/server/src/services/native-runtime/native-interaction-bridge.ts @@ -633,6 +633,8 @@ export async function routePersistedNativeResultAttention(input: { decisionId: existingDecision.id, leaseOwner: null, leaseExpiresAt: null, + recoveryState: null, + recoveryRequestId: null, failureCode: nonEmptyText(finalizationError?.cause), failureDetail: finalizationError ? { originalFailureCode: finalizationError.cause, diff --git a/server/src/services/native-runtime/native-run-coordinator-store.ts b/server/src/services/native-runtime/native-run-coordinator-store.ts index ad9f772023c..99b80907c66 100644 --- a/server/src/services/native-runtime/native-run-coordinator-store.ts +++ b/server/src/services/native-runtime/native-run-coordinator-store.ts @@ -411,6 +411,8 @@ export class NativeRunCoordinatorStore { set: { phase: "workspace_finalizing", resultId: inserted.id, + recoveryState: null, + recoveryRequestId: null, failureCode: null, failureDetail: null, nextAttemptAt: null, diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index b9ded4e21ed..ff5d3267a9d 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -222,6 +222,8 @@ async function claimCoordinator(input: { ? "arbitrating" : "workspace_finalizing" : coordinator.phase, + recoveryState: null, + recoveryRequestId: null, failureCode: null, failureDetail: null, nextAttemptAt: null, diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index ca8889804ed..5321a764f8b 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts @@ -1242,6 +1242,36 @@ describe("PaperclipControlPlanePort conformance", () => { } }); + it.each(["port", "coordinator"])("clears a resumed session's recovery state when %s persists its result", async (kind) => { + const identity = CONTROL_PLANE_CONFORMANCE_OPEN.identity; + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, companyId: identity.companyId, agentId: identity.agentId, + status: "running", runtimeMode: "native", nativeIssueId: identity.issueId, + nativeSessionId: identity.sessionId, runnerInstanceId: conformanceRunnerId, + completionContractId: contractId, completionContractSha256: contractSha, + }); + await db.insert(nativeRunFinalizations).values({ + runId, companyId: identity.companyId, issueId: identity.issueId, + phase: "observed", recoveryState: "resuming_session", recoveryRequestId: "restart-request", + }); + const port = kind === "port" ? new PaperclipControlPlanePort(db, { + ...identity, runId, completionContractId: contractId, completionContractSha256: contractSha, + sourceInstanceId: conformanceRunnerId, controlPlaneSourceInstanceId: "resumed-result", + }) : new NativeRunCoordinatorStore(db, { + ...identity, runId, completionContractId: contractId, completionContractSha256: contractSha, + normalizedSessionId: identity.sessionId, runnerSourceInstanceId: conformanceRunnerId, + completionContractRevision: "standalone-v1", completionContractCriterionIds: ["objective"], + }); + await port.completeRun({ + result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, + callerResultId: "resumed-result", + }); + await expect(db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId))).resolves.toEqual([ + expect.objectContaining({ phase: "workspace_finalizing", recoveryState: null, recoveryRequestId: null }), + ]); + }); + it("does not let native completion bypass a pending issue interaction", async () => { const identity = CONTROL_PLANE_CONFORMANCE_OPEN.identity; const port = new PaperclipControlPlanePort(db, { diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index ecdc639e74c..acb02e19b63 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -420,6 +420,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort { set: { phase: "workspace_finalizing", resultId: inserted.id, + recoveryState: null, + recoveryRequestId: null, failureCode: null, failureDetail: null, nextAttemptAt: null, diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index a7913f060ab..eafd64245c9 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -1863,6 +1863,8 @@ export async function commitNativeStatusDecision(input: { decisionId, leaseOwner: null, leaseExpiresAt: null, + recoveryState: null, + recoveryRequestId: null, failureCode: finalizationError?.cause ?? null, failureDetail: finalizationError ? { diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 7763b3be97c..42765bb3606 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -3487,17 +3487,14 @@ export function recoveryService( result.skipped += 1; continue; } - const wakePolicy = parseObject(action.wakePolicy); - const wakePolicyType = readNonEmptyString(wakePolicy.type); - if ( - wakePolicyType !== "bounded_recovery_owner" && - wakePolicyType !== "bounded_owner_disposition_repair" && - action.ownerType !== "board" - ) { - continue; - } - if (issue.status === "done" || issue.status === "cancelled") { + const systemRecoveryState = readNonEmptyString( + parseObject(parseObject(action.evidence).systemRecovery).state, + ); + if (systemRecoveryState && ["scheduled", "running", "awaiting_evidence"].includes(systemRecoveryState)) { + result.skipped += 1; + continue; + } const resolved = await recoveryActionsSvc.resolveActiveForIssue({ companyId: action.companyId, sourceIssueId: action.sourceIssueId, @@ -3513,6 +3510,16 @@ export function recoveryService( continue; } + const wakePolicy = parseObject(action.wakePolicy); + const wakePolicyType = readNonEmptyString(wakePolicy.type); + if ( + wakePolicyType !== "bounded_recovery_owner" && + wakePolicyType !== "bounded_owner_disposition_repair" && + action.ownerType !== "board" + ) { + continue; + } + // A queued comment or healthy child cannot establish what the stopped // provider already did. Only execution reconciliation can clear this hold. if (requiresExecutionReconciliation(action.cause)