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)