mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 10:13:11 +02:00
Clear stale native recovery state so closed work expires (#17359)
## 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 837add0caa
**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 <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip authored and GitHub committed
1 parent
dfe7df0015
commit
f73ace9f1a
19 files changed
+656
-20
No files matched your search
@@ -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;
|
||||
+258
-2
@@ -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",
|
||||
+7
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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<typeof nativeRunFinalizations.$inferInsert>; action?: Partial<typeof issueRecoveryActions.$inferInsert> };
|
||||
async function seedStaleState(f: Awaited<ReturnType<typeof fixture>>, 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);
|
||||
});
|
||||
});
|
||||
@@ -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`);
|
||||
|
||||
@@ -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<typeof issueRecoveryActions.$inferInsert>) => {
|
||||
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) => {
|
||||
|
||||
@@ -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<string, string>();
|
||||
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);
|
||||
});
|
||||
@@ -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, {
|
||||
|
||||
@@ -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 }),
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -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'
|
||||
|
||||
@@ -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<typeof createDb>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<ReturnType<typeof seedNativeRun>>) {
|
||||
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,
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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`,
|
||||
),
|
||||
),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -411,6 +411,8 @@ export class NativeRunCoordinatorStore {
|
||||
set: {
|
||||
phase: "workspace_finalizing",
|
||||
resultId: inserted.id,
|
||||
recoveryState: null,
|
||||
recoveryRequestId: null,
|
||||
failureCode: null,
|
||||
failureDetail: null,
|
||||
nextAttemptAt: null,
|
||||
|
||||
@@ -222,6 +222,8 @@ async function claimCoordinator(input: {
|
||||
? "arbitrating"
|
||||
: "workspace_finalizing"
|
||||
: coordinator.phase,
|
||||
recoveryState: null,
|
||||
recoveryRequestId: null,
|
||||
failureCode: null,
|
||||
failureDetail: null,
|
||||
nextAttemptAt: null,
|
||||
|
||||
@@ -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, {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -1863,6 +1863,8 @@ export async function commitNativeStatusDecision(input: {
|
||||
decisionId,
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
recoveryState: null,
|
||||
recoveryRequestId: null,
|
||||
failureCode: finalizationError?.cause ?? null,
|
||||
failureDetail: finalizationError
|
||||
? {
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user