diff --git a/doc/composer-stop.md b/doc/composer-stop.md index bd0165a19ca..1663012b0a1 100644 --- a/doc/composer-stop.md +++ b/doc/composer-stop.md @@ -18,7 +18,9 @@ terminal tasks. Resume/restore retain the optional wake-agents checkbox. ## Resume and execution recovery -Resume releases a hold. Waking agents is optional and only applies to tasks in +Resume releases a hold. Work the pause interrupted is re-dispatched once +automatically, under current admission gates (see `doc/execution-semantics.md`). +Waking other agents is optional and only applies to tasks in `todo`, `in_progress`, or `in_review`; parked and terminal tasks stay untouched. The current execution-recovery policy requires verified outcomes before a stopped provider can restart. If any affected task still needs that reconciliation, diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index f84fbf74e42..f112e1d9391 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1879,6 +1879,30 @@ interrupt. A verified native stop must not become an unexplained failure simply because it came from a subtree action. The explicit pause hold still prevents future execution until Resume, and missing stop proof still blocks continuation. +A run interrupted by a pause is an operator hold, not a failure. The pause's member +snapshot names each interrupted run, and that work is owed one re-dispatch once no +pause covers the task. The obligation belongs to the run, so overlapping pauses keep +it until the last covering release. Releasing or resuming a hold re-dispatches owed +work (`issue_tree_resumed`, with `holdId` and `interruptedRunId` in the wake +context), even without `wakeAgents`. The interrupted run must still be the task's +latest run, assigned to the same agent, and executable; normal admission gates still +apply. A compare-and-set claim on the interrupted run makes the resume happen at most +once. The claim is saved as `pending` and arms the execution-continuation queue in +the same commit; the wake uses the stable idempotency key `tree-hold-resume:`. +If the process stops before that wake is saved, the queue's sweep finds the stale +claim, rechecks current state, and dispatches it once. A task that still needs execution reconciliation is not replayed: the release +records `issue.tree_resume_wake_skipped`, the blocker stays visible, and the +obligation stays unclaimed. + +A legacy run stopped before its gated dispatch entered the adapter has no provider +actions to reconcile. Once its executor confirms no process, session, run-owned lease +or execution event, it records bootstrap evidence and resolves that run's +reconciliation hold as a false positive, including an automatic no-replay +disposition. It restores a `blocked` status only when that disposition made the +change (`legacyCancellationBlock` receipt). The executor then re-dispatches owed work +if the pause was already released. The automatic disposition waits while a legacy +executor still holds its controller lease. A hold can be released only once. + ### Late provider usage Each run measures its usage from the total the previous run accounted for, whether diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index aca401494ff..6a7cb262907 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -592,6 +592,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0414_closed_edwin_jarvis.sql", "0419_slippery_impossible_man.sql", "0432_strong_mole_man.sql", + "0435_premium_puff_adder.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0435_premium_puff_adder.sql b/packages/db/src/migrations/0435_premium_puff_adder.sql new file mode 100644 index 00000000000..48221a4f5b0 --- /dev/null +++ b/packages/db/src/migrations/0435_premium_puff_adder.sql @@ -0,0 +1 @@ +CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "agent_wakeup_requests_tree_hold_resume_uq" ON "agent_wakeup_requests" USING btree ("company_id","idempotency_key") WHERE "agent_wakeup_requests"."idempotency_key" LIKE 'tree-hold-resume:%' AND "agent_wakeup_requests"."status" <> 'skipped'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0430_snapshot.json b/packages/db/src/migrations/meta/0435_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0430_snapshot.json rename to packages/db/src/migrations/meta/0435_snapshot.json index 814a23ca614..0197a7a1c9c 100644 --- a/packages/db/src/migrations/meta/0430_snapshot.json +++ b/packages/db/src/migrations/meta/0435_snapshot.json @@ -1,6 +1,6 @@ { - "id": "b900b172-2cdc-45c6-a964-27e850bc4c74", - "prevId": "809726a1-762c-4c9d-a7fd-55670f1c856f", + "id": "dedfce21-7ff9-4079-b5da-4a74a856a251", + "prevId": "5b40e9ac-12aa-4757-aac7-36218c7500a8", "version": "7", "dialect": "postgresql", "tables": { @@ -713,6 +713,12 @@ "notNull": true, "default": "'{}'::jsonb" }, + "previous_usage_json": { + "name": "previous_usage_json", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "reason": { "name": "reason", "type": "text", @@ -4052,6 +4058,28 @@ "method": "btree", "with": {} }, + "agent_wakeup_requests_tree_hold_resume_uq": { + "name": "agent_wakeup_requests_tree_hold_resume_uq", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "idempotency_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "where": "\"agent_wakeup_requests\".\"idempotency_key\" LIKE 'tree-hold-resume:%' AND \"agent_wakeup_requests\".\"status\" <> 'skipped'", + "concurrently": true, + "method": "btree", + "with": {} + }, "agent_wakeup_requests_chat_completion_uq": { "name": "agent_wakeup_requests_chat_completion_uq", "columns": [ @@ -22033,6 +22061,12 @@ "primaryKey": false, "notNull": false }, + "reported_usage_json": { + "name": "reported_usage_json", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "pricing_provenance": { "name": "pricing_provenance", "type": "jsonb", @@ -34174,6 +34208,50 @@ "concurrently": false, "method": "btree", "with": {} + }, + "heartbeat_run_events_usage_item_run_idx": { + "name": "heartbeat_run_events_usage_item_run_idx", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_run_events\".\"event_type\" = 'item.completed' and (\"heartbeat_run_events\".\"payload\" -> 'prpEvent' -> 'payload' ->> 'kind') = 'usage'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "heartbeat_run_events_compaction_session_idx": { + "name": "heartbeat_run_events_compaction_session_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"payload\" -> 'prpEvent' ->> 'normalizedSessionId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_run_events\".\"event_type\" = 'context.compacted'", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -42436,6 +42514,12 @@ "primaryKey": false, "notNull": false }, + "origin_issue_id": { + "name": "origin_issue_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, "origin_identity_context_id": { "name": "origin_identity_context_id", "type": "uuid", @@ -42879,6 +42963,27 @@ "method": "btree", "with": {} }, + "issues_company_origin_issue_idx": { + "name": "issues_company_origin_issue_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "origin_issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "issues_company_project_workspace_idx": { "name": "issues_company_project_workspace_idx", "columns": [ @@ -43478,6 +43583,19 @@ "onDelete": "no action", "onUpdate": "no action" }, + "issues_origin_issue_id_issues_id_fk": { + "name": "issues_origin_issue_id_issues_id_fk", + "tableFrom": "issues", + "tableTo": "issues", + "columnsFrom": [ + "origin_issue_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + }, "issues_execution_workspace_id_execution_workspaces_id_fk": { "name": "issues_execution_workspace_id_execution_workspaces_id_fk", "tableFrom": "issues", @@ -61828,6 +61946,13 @@ "type": "boolean", "primaryKey": false, "notNull": true, + "default": true + }, + "classic_workspace_enabled": { + "name": "classic_workspace_enabled", + "type": "boolean", + "primaryKey": false, + "notNull": true, "default": false }, "created_at": { diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 1ea6e55327d..0d04f0a001b 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3032,6 +3032,13 @@ "when": 1793924947593, "tag": "0434_brave_leo", "breakpoints": true + }, + { + "idx": 435, + "version": "7", + "when": 1793937048235, + "tag": "0435_premium_puff_adder", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/agent_wakeup_requests.ts b/packages/db/src/schema/agent_wakeup_requests.ts index fc7feaed02f..83006f2616c 100644 --- a/packages/db/src/schema/agent_wakeup_requests.ts +++ b/packages/db/src/schema/agent_wakeup_requests.ts @@ -90,6 +90,12 @@ export const agentWakeupRequests = pgTable( toolActionDeliveryIdempotencyUq: uniqueIndex("agent_wakeup_requests_tool_action_delivery_uq") .on(table.companyId, table.idempotencyKey) .where(sql`${table.idempotencyKey} LIKE 'tool-action-response:%' AND ${table.status} NOT IN ('skipped', 'failed', 'cancelled')`), + // A subtree pause owes its interrupted run one resume wake, ever. A stale + // dispatcher that saves late must not repeat work a takeover already ran. + treeHoldResumeIdempotencyUq: uniqueIndex("agent_wakeup_requests_tree_hold_resume_uq") + .on(table.companyId, table.idempotencyKey) + .concurrently() + .where(sql`${table.idempotencyKey} LIKE 'tree-hold-resume:%' AND ${table.status} <> 'skipped'`), chatCompletionIdempotencyUq: uniqueIndex("agent_wakeup_requests_chat_completion_uq") .on(table.companyId, table.idempotencyKey) .where(sql`${table.idempotencyKey} LIKE 'chat-completion:%'`), diff --git a/server/src/__tests__/issue-tree-control-routes.test.ts b/server/src/__tests__/issue-tree-control-routes.test.ts index 682650b7c5d..28966145ad0 100644 --- a/server/src/__tests__/issue-tree-control-routes.test.ts +++ b/server/src/__tests__/issue-tree-control-routes.test.ts @@ -16,6 +16,7 @@ const mockTreeControlService = vi.hoisted(() => ({ getHold: vi.fn(), getActivePauseHoldGate: vi.fn(), releaseHold: vi.fn(), + listPauseInterruptedResumeTargets: vi.fn(), cancelUnclaimedWakeupsForTree: vi.fn(), })); @@ -79,6 +80,7 @@ describe("issue tree control routes", () => { companyId: "company-2", }); mockTreeControlService.cancelUnclaimedWakeupsForTree.mockResolvedValue([]); + mockTreeControlService.listPauseInterruptedResumeTargets.mockResolvedValue([]); mockTreeControlService.cancelIssueStatusesForHold.mockResolvedValue({ updatedIssueIds: [], updatedIssues: [], diff --git a/server/src/__tests__/issue-tree-pause-resume-run.test.ts b/server/src/__tests__/issue-tree-pause-resume-run.test.ts new file mode 100644 index 00000000000..20288d1be4d --- /dev/null +++ b/server/src/__tests__/issue-tree-pause-resume-run.test.ts @@ -0,0 +1,527 @@ +import { randomUUID } from "node:crypto"; +import express from "express"; +import request from "supertest"; +import { and, eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + activityLog, + agentRuntimeState, + agentWakeupRequests, + agents, + authUsers, + closeRegisteredClients, + companies, + companyMemberships, + companySkills, + costEvents, + createDb, + environmentLeases, + heartbeatRunEvents, + heartbeatRuns, + issueRecoveryActions, + issueTreeHoldMembers, + issueTreeHolds, + issues, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; +import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.ts"; + +// Hold one run in host preparation, before the legacy adapter is entered, so a +// subtree pause can interrupt it exactly where the staging retry was stopped. +const preparationGate = vi.hoisted(() => ({ + blockNext: false, + blockedRunId: null as string | null, + reached: null as null | (() => void), + release: null as null | (() => void), +})); +vi.mock("../services/heartbeat/run-preparation.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + buildPaperclipRuntimeMcpServers: async (input: Parameters[0]) => { + if (preparationGate.blockNext) { + preparationGate.blockNext = false; + preparationGate.blockedRunId = input.runId; + await new Promise((resolve) => { + preparationGate.release = resolve; + preparationGate.reached?.(); + }); + } + return actual.buildPaperclipRuntimeMcpServers(input); + }, + }; +}); + +const externalTestDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL?.trim(); +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; +const ADAPTER = "tree_pause_resume_test"; + +describeEmbeddedPostgres("subtree pause and release of an in-flight child run", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + let heartbeat!: ReturnType; + let getExecutionBlocker!: typeof import("../services/execution-blocker.js").getExecutionBlocker; + let app!: express.Express; + const executedRunIds: string[] = []; + const adapterGate = { block: false, entered: null as null | (() => void) }; + + beforeAll(async () => { + if (externalTestDatabaseUrl) db = createDb(externalTestDatabaseUrl); + else { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-tree-pause-resume-"); + db = createDb(tempDb.connectionString); + } + const [{ heartbeatService }, blocker, { issueTreeControlRoutes }, { errorHandler }] = await Promise.all([ + import("../services/heartbeat.ts"), + import("../services/execution-blocker.js"), + import("../routes/issue-tree-control.js"), + import("../middleware/index.js"), + ]); + heartbeat = heartbeatService(db); + getExecutionBlocker = blocker.getExecutionBlocker; + registerServerAdapter({ + type: ADAPTER, + execute: async (input) => { + executedRunIds.push(input.runId); + if (adapterGate.block) { + adapterGate.entered?.(); + await new Promise((resolve) => { + if (input.signal?.aborted) return resolve(); + input.signal?.addEventListener("abort", () => resolve(), { once: true }); + }); + } + // Finish the task so liveness follow-ups do not add unrelated runs. + const issueId = (input.context as { issueId?: unknown }).issueId; + if (typeof issueId === "string" && !input.signal?.aborted) { + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, issueId)); + } + return { exitCode: 0, signal: null, timedOut: false, resultJson: {} }; + }, + testEnvironment: async () => ({ adapterType: ADAPTER, status: "pass", checks: [], testedAt: new Date().toISOString() }), + }); + app = express(); + app.use(express.json()); + app.use((req, _res, next) => { + (req as any).actor = { type: "board", userId: "board-user", source: "local_implicit", isInstanceAdmin: false }; + next(); + }); + app.use("/api", issueTreeControlRoutes(db)); + app.use(errorHandler); + }, 30_000); + + afterEach(async () => { + preparationGate.release?.(); + Object.assign(preparationGate, { blockNext: false, blockedRunId: null, reached: null, release: null }); + adapterGate.block = false; + adapterGate.entered = null; + await drainHeartbeatRunsToQuiescence(db, heartbeat); + executedRunIds.length = 0; + for (const table of [costEvents, activityLog, issueRecoveryActions, issueTreeHoldMembers, issueTreeHolds, + environmentLeases, heartbeatRunEvents]) await db.delete(table); + await db.update(issues).set({ executionRunId: null, checkoutRunId: null }); + for (const table of [issues, heartbeatRuns, agentWakeupRequests, agentRuntimeState, agents, companySkills, + companyMemberships, companies]) await db.delete(table); + }); + + afterAll(async () => { + unregisterServerAdapter(ADAPTER); + if (externalTestDatabaseUrl) await closeRegisteredClients(externalTestDatabaseUrl); + await db?.$client.end(); + await tempDb?.cleanup(); + }); + + async function seedTree() { + const companyId = randomUUID(); + const agentId = randomUUID(); + const parentId = randomUUID(); + const childId = randomUUID(); + const prefix = `P${companyId.replace(/-/g, "").slice(0, 5).toUpperCase()}`; + const now = new Date(); + await db.insert(companies).values({ id: companyId, name: "Pause co", issuePrefix: prefix, + requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "responsible-user" }); + await db.insert(authUsers).values({ id: "responsible-user", name: "Responsible", email: "tree-pause@example.test", + createdAt: now, updatedAt: now }).onConflictDoNothing(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "responsible-user", + status: "active", membershipRole: "member" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Nibbler", role: "engineer", status: "idle", + adapterType: ADAPTER, adapterConfig: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, permissions: {} }); + await db.insert(issues).values({ id: parentId, companyId, title: "Parent", status: "in_progress", priority: "medium", + responsibleUserId: "responsible-user", assigneeUserId: "responsible-user", issueNumber: 1, identifier: `${prefix}-1` }); + await db.insert(issues).values({ id: childId, companyId, parentId, title: "Child", status: "todo", priority: "medium", + responsibleUserId: "responsible-user", assigneeAgentId: agentId, issueNumber: 2, identifier: `${prefix}-2` }); + return { companyId, agentId, parentId, childId, prefix }; + } + + function wakeChild(fixture: Awaited>) { + return heartbeat.wakeup(fixture.agentId, { + source: "assignment", triggerDetail: "system", reason: "issue_assigned", + payload: { issueId: fixture.childId }, requestedByActorType: "system", requestedByActorId: "system", + contextSnapshot: { issueId: fixture.childId, taskId: fixture.childId, wakeReason: "issue_assigned" }, + }); + } + + async function childRuns(childId: string) { + const rows = await db.select().from(heartbeatRuns); + return rows.filter((run) => run.contextSnapshot?.issueId === childId) + .sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime()); + } + + async function pauseAndRelease( + fixture: Awaited>, + hooks: { afterPause?: () => Promise; whilePaused?: () => Promise } = {}, + ) { + const pause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`) + .send({ mode: "pause", reason: "soak pause" }); + expect(pause.status).toBe(201); + await hooks.afterPause?.(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await hooks.whilePaused?.(); + const release = await request(app) + .post(`/api/issues/${fixture.parentId}/tree-holds/${pause.body.hold.id}/release`) + .send({ reason: "soak resume" }); + expect(release.status).toBe(200); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + return pause.body.hold.id as string; + } + + it("re-dispatches a child whose run was paused during preparation exactly once, without reconciliation", async () => { + const fixture = await seedTree(); + const prepared = new Promise((resolve) => { preparationGate.reached = resolve; }); + preparationGate.blockNext = true; + await wakeChild(fixture); + await prepared; + const interruptedRunId = preparationGate.blockedRunId!; + + const holdId = await pauseAndRelease(fixture, { + // The executor observes the operator Stop at its dispatch gate. + afterPause: async () => { preparationGate.release?.(); }, + whilePaused: async () => { + // Nothing runs during the hold, and the Stop needs no reconciliation. + expect((await childRuns(fixture.childId)).map((run) => run.status)).toEqual(["cancelled"]); + expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull(); + }, + }); + + const runs = await childRuns(fixture.childId); + expect(runs[0]).toMatchObject({ id: interruptedRunId, status: "cancelled" }); + expect(executedRunIds).not.toContain(interruptedRunId); + const resumed = runs.slice(1); + expect(resumed).toHaveLength(1); + expect(resumed[0]).toMatchObject({ status: "succeeded" }); + expect(resumed[0]!.contextSnapshot).toMatchObject({ wakeReason: "issue_tree_resumed", holdId }); + expect(executedRunIds).toEqual([resumed[0]!.id]); + const holds = await db.select().from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.sourceIssueId, fixture.childId), eq(issueRecoveryActions.status, "active"))); + expect(holds).toEqual([]); + expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull(); + const [child] = await db.select().from(issues).where(eq(issues.id, fixture.childId)); + expect(child!.status).toBe("done"); + + // The released hold owes its work once; a repeated release cannot replay it. + const again = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`) + .send({ reason: "duplicate resume" }); + expect(again.status).toBe(409); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + expect(await childRuns(fixture.childId)).toHaveLength(2); + }); + + async function pauseWhilePreparing(fixture: Awaited>, rootId = fixture.parentId) { + const prepared = new Promise((resolve) => { preparationGate.reached = resolve; }); + preparationGate.blockNext = true; + await wakeChild(fixture); + await prepared; + const pause = await request(app).post(`/api/issues/${rootId}/tree-holds`).send({ mode: "pause" }); + expect(pause.status).toBe(201); + return { interruptedRunId: preparationGate.blockedRunId!, holdId: pause.body.hold.id as string }; + } + + it("resumes once when the pause is released before the stopped preparation finishes unwinding", async () => { + const fixture = await seedTree(); + const { interruptedRunId, holdId } = await pauseWhilePreparing(fixture); + // The periodic disposition must not block a Stop its executor still owns. + const { settleUnrecoverableExecutions } = await import("../services/execution-recovery-resolution.js"); + await settleUnrecoverableExecutions(db); + const release = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`).send({}); + expect(release.status).toBe(200); + expect((await childRuns(fixture.childId)).map((run) => run.status)).toEqual(["cancelled"]); + const [held] = await db.select().from(issues).where(eq(issues.id, fixture.childId)); + expect(held!.status).not.toBe("blocked"); + + preparationGate.release?.(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + + const runs = await childRuns(fixture.childId); + expect(runs.map((run) => run.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]); + expect(runs[1]).toMatchObject({ status: "succeeded" }); + expect(runs[1]!.contextSnapshot).toMatchObject({ holdId, interruptedRunId }); + expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull(); + }); + + it("clears an automatic no-replay block recorded before the executor proved no adapter started", async () => { + const fixture = await seedTree(); + const { interruptedRunId, holdId } = await pauseWhilePreparing(fixture); + // Model a disposition that ran after the controller lease lapsed. + await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(0) }).where(eq(heartbeatRuns.id, interruptedRunId)); + const { settleUnrecoverableExecutions } = await import("../services/execution-recovery-resolution.js"); + await settleUnrecoverableExecutions(db); + const [blocked] = await db.select().from(issues).where(eq(issues.id, fixture.childId)); + expect(blocked!.status).toBe("blocked"); + + preparationGate.release?.(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull(); + const [restored] = await db.select().from(issues).where(eq(issues.id, fixture.childId)); + expect(restored!.status).toBe("in_progress"); + + const release = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`).send({}); + expect(release.status).toBe(200); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const runs = await childRuns(fixture.childId); + expect(runs.map((run) => run.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]); + }); + + it("keeps the interrupted work owed across overlapping pauses until the last one is released", async () => { + const fixture = await seedTree(); + const { interruptedRunId, holdId: childHoldId } = await pauseWhilePreparing(fixture, fixture.childId); + preparationGate.release?.(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const parentPause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "pause" }); + expect(parentPause.status).toBe(201); + + const childRelease = await request(app) + .post(`/api/issues/${fixture.childId}/tree-holds/${childHoldId}/release`).send({}); + expect(childRelease.status).toBe(200); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + expect(await childRuns(fixture.childId)).toHaveLength(1); + + const parentRelease = await request(app) + .post(`/api/issues/${fixture.parentId}/tree-holds/${parentPause.body.hold.id}/release`).send({}); + expect(parentRelease.status).toBe(200); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const runs = await childRuns(fixture.childId); + expect(runs.map((run) => run.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]); + expect(runs[1]!.contextSnapshot).toMatchObject({ holdId: childHoldId, interruptedRunId }); + }); + + it("lets only one of two concurrent releases resume the interrupted work", async () => { + const fixture = await seedTree(); + const { holdId } = await pauseWhilePreparing(fixture); + preparationGate.release?.(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + + const url = `/api/issues/${fixture.parentId}/tree-holds/${holdId}/release`; + const responses = await Promise.all([request(app).post(url).send({}), request(app).post(url).send({})]); + expect(responses.map((response) => response.status).sort()).toEqual([200, 409]); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const runs = await childRuns(fixture.childId); + expect(runs.map((run) => run.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]); + }); + + it("finishes a resume whose wake was not saved before a restart, exactly once", async () => { + const fixture = await seedTree(); + const { interruptedRunId, holdId } = await pauseWhilePreparing(fixture); + preparationGate.release?.(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + // The release claimed the work, then the process stopped before the wake was saved. + const claimedAt = new Date(Date.now() - 5 * 60_000).toISOString(); + await db.update(issueTreeHolds).set({ status: "released", releasedAt: new Date() }).where(eq(issueTreeHolds.id, holdId)); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, interruptedRunId)); + await db.update(heartbeatRuns).set({ resultJson: { ...run!.resultJson, treeHoldResume: { holdId, state: "pending", claimedAt } } }) + .where(eq(heartbeatRuns.id, interruptedRunId)); + + const { deliverPendingPauseResumes, hasPendingPauseResume } = await import("../services/issue-tree-resume.js"); + expect(await hasPendingPauseResume(db)).toBe(true); + await deliverPendingPauseResumes(db, (agentId, opts) => heartbeat.wakeup(agentId, opts)); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await deliverPendingPauseResumes(db, (agentId, opts) => heartbeat.wakeup(agentId, opts), new Date(Date.now() + 5 * 60_000)); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + + const runs = await childRuns(fixture.childId); + expect(runs.map((row) => row.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]); + expect(runs[1]).toMatchObject({ status: "succeeded" }); + expect(await hasPendingPauseResume(db)).toBe(false); + }); + + it("does not let a stalled dispatcher repeat a resume the sweep already took over", async () => { + const fixture = await seedTree(); + const { holdId } = await pauseWhilePreparing(fixture); + preparationGate.release?.(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await db.update(issueTreeHolds).set({ status: "released", releasedAt: new Date() }).where(eq(issueTreeHolds.id, holdId)); + + // The first dispatcher claims the work, then stalls before its wake is saved. + const { resumePauseInterruptedWork, deliverPendingPauseResumes } = await import("../services/issue-tree-resume.js"); + let resumeStalled!: () => void; + const stalled = new Promise((resolve) => { resumeStalled = resolve; }); + let continueStalled!: () => void; + const gate = new Promise((resolve) => { continueStalled = resolve; }); + const slow = resumePauseInterruptedWork(db, { + companyId: fixture.companyId, issueIds: [fixture.childId], actor: { actorType: "system", actorId: "system" }, + wakeup: async (agentId, opts) => { resumeStalled(); await gate; return heartbeat.wakeup(agentId, opts); }, + }); + await stalled; + + // The sweep takes over the stale claim and the replacement run finishes + // while the task stays open. + await deliverPendingPauseResumes(db, (agentId, opts) => heartbeat.wakeup(agentId, opts), new Date(Date.now() + 5 * 60_000)); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await db.update(issues).set({ status: "in_progress", completedAt: null }).where(eq(issues.id, fixture.childId)); + + continueStalled(); + expect(await slow).toEqual([]); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const runs = await childRuns(fixture.childId); + expect(runs.map((run) => run.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]); + const wakes = await db.select().from(agentWakeupRequests) + .where(eq(agentWakeupRequests.idempotencyKey, `tree-hold-resume:${runs[0]!.id}`)); + expect(wakes).toHaveLength(1); + }); + + it("keeps a resume owed when a new pause lands before its wake is admitted", async () => { + const fixture = await seedTree(); + const { holdId } = await pauseWhilePreparing(fixture); + preparationGate.release?.(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await db.update(issueTreeHolds).set({ status: "released", releasedAt: new Date() }).where(eq(issueTreeHolds.id, holdId)); + + // A second pause lands between the eligibility check and queue admission. + const { resumePauseInterruptedWork, deliverPendingPauseResumes } = await import("../services/issue-tree-resume.js"); + let secondHoldId = ""; + expect(await resumePauseInterruptedWork(db, { + companyId: fixture.companyId, issueIds: [fixture.childId], actor: { actorType: "system", actorId: "system" }, + wakeup: async (agentId, opts) => { + const pause = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds`).send({ mode: "pause" }); + expect(pause.status).toBe(201); + secondHoldId = pause.body.hold.id; + return heartbeat.wakeup(agentId, opts); + }, + })).toEqual([]); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + expect(await childRuns(fixture.childId)).toHaveLength(1); + + const release = await request(app).post(`/api/issues/${fixture.parentId}/tree-holds/${secondHoldId}/release`).send({}); + expect(release.status).toBe(200); + await deliverPendingPauseResumes(db, (agentId, opts) => heartbeat.wakeup(agentId, opts), new Date(Date.now() + 5 * 60_000)); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const runs = await childRuns(fixture.childId); + expect(runs.map((run) => run.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]); + expect(runs[1]).toMatchObject({ status: "succeeded" }); + }); + + it("visits every stale resume claim even when the first page is still paused", async () => { + const fixture = await seedTree(); + const { interruptedRunId, holdId } = await pauseWhilePreparing(fixture); + preparationGate.release?.(); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await db.update(issueTreeHolds).set({ status: "released", releasedAt: new Date() }).where(eq(issueTreeHolds.id, holdId)); + + // A full page of claims on tasks under another active pause sorts first. + const pausedRootId = randomUUID(); + await db.insert(issues).values({ id: pausedRootId, companyId: fixture.companyId, title: "Paused root", status: "in_progress", + priority: "medium", responsibleUserId: "responsible-user", assigneeUserId: "responsible-user", issueNumber: 10, + identifier: `${fixture.prefix}-10` }); + const pausedChildIds = Array.from({ length: 25 }, () => randomUUID()); + for (const [index, id] of pausedChildIds.entries()) { + await db.insert(issues).values({ id, companyId: fixture.companyId, parentId: pausedRootId, title: `Paused ${index}`, + status: "in_progress", priority: "medium", responsibleUserId: "responsible-user", assigneeAgentId: fixture.agentId, + issueNumber: 11 + index, identifier: `${fixture.prefix}-${11 + index}` }); + } + const otherPause = await request(app).post(`/api/issues/${pausedRootId}/tree-holds`).send({ mode: "pause" }); + expect(otherPause.status).toBe(201); + const claimedAt = new Date(Date.now() - 5 * 60_000).toISOString(); + const pendingClaim = { treeHoldResume: { holdId: otherPause.body.hold.id, state: "pending", claimedAt } }; + await db.insert(heartbeatRuns).values(pausedChildIds.map((issueId, index) => ({ + id: `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, companyId: fixture.companyId, agentId: fixture.agentId, + status: "cancelled", finishedAt: new Date(Date.now() - 10 * 60_000), createdAt: new Date(Date.now() - 10 * 60_000), + contextSnapshot: { issueId }, resultJson: pendingClaim, + }))); + // The runnable claim sorts after that page. + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, interruptedRunId)); + await db.update(heartbeatRuns).set({ resultJson: { ...run!.resultJson, treeHoldResume: { holdId, state: "pending", claimedAt } } }) + .where(eq(heartbeatRuns.id, interruptedRunId)); + expect(interruptedRunId > `00000000-0000-4000-8000-${String(24).padStart(12, "0")}`).toBe(true); + + const { deliverPendingPauseResumes } = await import("../services/issue-tree-resume.js"); + await deliverPendingPauseResumes(db, (agentId, opts) => heartbeat.wakeup(agentId, opts)); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + + const runs = await childRuns(fixture.childId); + expect(runs.map((row) => row.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]); + }); + + it("re-dispatches a queued child run that the pause cancelled before it started", async () => { + const fixture = await seedTree(); + // Occupy the agent's only slot so the child's run stays queued. + const otherIssueId = randomUUID(); + await db.insert(issues).values({ id: otherIssueId, companyId: fixture.companyId, title: "Other", status: "todo", + priority: "medium", responsibleUserId: "responsible-user", assigneeAgentId: fixture.agentId, issueNumber: 3, + identifier: `${fixture.prefix}-3` }); + const prepared = new Promise((resolve) => { preparationGate.reached = resolve; }); + preparationGate.blockNext = true; + await heartbeat.wakeup(fixture.agentId, { + source: "assignment", triggerDetail: "system", reason: "issue_assigned", payload: { issueId: otherIssueId }, + requestedByActorType: "system", requestedByActorId: "system", + contextSnapshot: { issueId: otherIssueId, taskId: otherIssueId, wakeReason: "issue_assigned" }, + }); + await prepared; + const queued = await wakeChild(fixture); + expect(queued).toMatchObject({ status: "queued" }); + + const holdId = await pauseAndRelease(fixture, { + afterPause: async () => { + expect((await childRuns(fixture.childId)).map((run) => run.status)).toEqual(["cancelled"]); + preparationGate.release?.(); + }, + whilePaused: async () => { + expect((await childRuns(fixture.childId)).map((run) => run.status)).toEqual(["cancelled"]); + }, + }); + + const runs = await childRuns(fixture.childId); + expect(runs.map((run) => run.contextSnapshot?.wakeReason)).toEqual(["issue_assigned", "issue_tree_resumed"]); + expect(runs[1]).toMatchObject({ status: "succeeded" }); + expect(runs[1]!.contextSnapshot).toMatchObject({ holdId, interruptedRunId: queued!.id }); + expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)).toBeNull(); + }); + + it("does not resume work the board reassigned while the tree was paused", async () => { + const fixture = await seedTree(); + const prepared = new Promise((resolve) => { preparationGate.reached = resolve; }); + preparationGate.blockNext = true; + await wakeChild(fixture); + await prepared; + + await pauseAndRelease(fixture, { + afterPause: async () => { preparationGate.release?.(); }, + whilePaused: async () => { + await db.update(issues).set({ assigneeAgentId: null, assigneeUserId: "responsible-user" }) + .where(eq(issues.id, fixture.childId)); + }, + }); + + expect((await childRuns(fixture.childId)).map((run) => run.status)).toEqual(["cancelled"]); + expect(executedRunIds).toEqual([]); + }); + + it("keeps an uncertain mid-turn Stop held for reconciliation instead of replaying it on release", async () => { + const fixture = await seedTree(); + const entered = new Promise((resolve) => { adapterGate.entered = resolve; }); + adapterGate.block = true; + const run = await wakeChild(fixture); + await entered; + + await pauseAndRelease(fixture, { whilePaused: async () => { adapterGate.block = false; } }); + + const runs = await childRuns(fixture.childId); + expect(runs.map((row) => row.id)).toEqual([run!.id]); + expect(runs[0]).toMatchObject({ status: "cancelled" }); + expect(await getExecutionBlocker(db, fixture.companyId, fixture.childId)) + .toMatchObject({ cause: "legacy_execution_requires_reconciliation", runId: run!.id }); + const skipped = await db.select().from(activityLog).where(eq(activityLog.action, "issue.tree_resume_wake_skipped")); + expect(skipped).toHaveLength(1); + expect(skipped[0]!.details).toMatchObject({ interruptedRunId: run!.id, cause: "legacy_execution_requires_reconciliation" }); + }); +}); diff --git a/server/src/index.ts b/server/src/index.ts index fca1599fae3..d16023336b8 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -80,6 +80,7 @@ import { verifyStoppedNativeSessionForReplacement } from "./services/native-runt import { embeddedPostgresOwnerPort } from "./embedded-postgres-owner.js"; import { deliverExecutionStatuses } from "./services/execution-status-delivery.js"; import { deliverReconciledExecutions, settleUnrecoverableExecutions } from "./services/execution-recovery-resolution.js"; +import { deliverPendingPauseResumes } from "./services/issue-tree-resume.js"; import { reconcileSafeNativeReplacements } from "./services/native-runtime/native-safe-replacement.js"; import { reconcileAbandonedExecutionControl } from "./services/execution-control-reconciliation.js"; import { acknowledgeNativeStartupCleanup } from "./services/native-runtime/native-startup-cleanup.js"; @@ -1324,7 +1325,10 @@ async function startServerWithDatabaseTeardown( { queue: DELIVERY_QUEUES.executionDisposition, run: () => settleUnrecoverableExecutions(db), next: executionSchedule.disposition }, ...(heartbeat ? [ { queue: DELIVERY_QUEUES.executionReplacement, run: () => reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: run => verifyStoppedNativeSessionForReplacement(db, run) }), next: executionSchedule.replacement }, - { queue: DELIVERY_QUEUES.executionContinuation, run: () => deliverReconciledExecutions(db, heartbeat.wakeup), next: executionSchedule.continuation }, + { queue: DELIVERY_QUEUES.executionContinuation, run: async () => { + await deliverReconciledExecutions(db, heartbeat.wakeup); + await deliverPendingPauseResumes(db, heartbeat.wakeup); + }, next: executionSchedule.continuation }, ] : []), ], () => !heartbeatSchedulerStopped); diff --git a/server/src/routes/issue-tree-control.ts b/server/src/routes/issue-tree-control.ts index 05c208756e9..eb90f99f59d 100644 --- a/server/src/routes/issue-tree-control.ts +++ b/server/src/routes/issue-tree-control.ts @@ -6,6 +6,7 @@ import { } from "@paperclipai/db"; import { and, eq, inArray, isNotNull } from "drizzle-orm"; import { getExecutionBlocker } from "../services/execution-blocker.js"; +import { resumePauseInterruptedWork as resumeInterruptedWork } from "../services/issue-tree-resume.js"; import { ISSUE_TREE_RESUME_EXECUTABLE_STATUSES } from "../services/issue-tree-control.js"; import { conflict } from "../errors.js"; import { @@ -92,6 +93,24 @@ export function issueTreeControlRoutes( return root; } + async function resumePauseInterruptedWork( + root: { id: string; companyId: string }, + holdId: string, + actor: ReturnType, + ) { + // A pause Stop is an operator hold. Re-dispatch the work it interrupted + // once, under current admission gates; never replay uncertain actions. + const hold = await treeControlSvc.getHold(root.companyId, holdId); + if (!hold || hold.mode !== "pause") return []; + return resumeInterruptedWork(db, { + companyId: root.companyId, + issueIds: (hold.members ?? []).map((member) => member.issueId), + rootIssueId: root.id, + actor, + wakeup: (agentId, opts) => heartbeat.wakeup(agentId, opts), + }); + } + router.post("/issues/:id/tree-control/preview", validate(previewIssueTreeControlSchema), async (req, res) => { assertBoard(req); const root = await getAccessibleResource(req, res, resolveRootIssue(req), "Root issue not found"); @@ -356,6 +375,14 @@ export function issueTreeControlRoutes( } } + if (result.hold.mode === "resume") { + const wakeFailures: Array<{ issueId: string; message: string }> = []; + for (const pauseHoldId of result.resumedPauseHoldIds ?? []) { + wakeFailures.push(...await resumePauseInterruptedWork(root, pauseHoldId, actor)); + } + if (wakeFailures.length) result = { ...result, wakeFailures } as typeof result; + } + res .status(result.hold.mode === "restore" || result.hold.mode === "resume" ? 200 : 201) .json(result); @@ -544,6 +571,11 @@ export function issueTreeControlRoutes( } } + // An explicit wake above already covers every executable member. + if (hold.mode === "pause" && req.body.metadata?.wakeAgents !== true) { + wakeFailures.push(...await resumePauseInterruptedWork(root, hold.id, actor)); + } + res.json({ ...hold, ...(wakeFailures.length ? { wakeFailures } : {}) }); }, ); diff --git a/server/src/services/execution-control-work.ts b/server/src/services/execution-control-work.ts index 69863954dbc..f282b9d5820 100644 --- a/server/src/services/execution-control-work.ts +++ b/server/src/services/execution-control-work.ts @@ -6,6 +6,7 @@ import { type DeliveryQueue } from "./delivery-work-notifications.js"; import { EXECUTION_RECONCILIATION_INTERVAL_MS } from "./execution-control-deadline.js"; import { obsoleteExecutionConversationHold, unresolvedExecutionDispositions } from "./execution-recovery-resolution.js"; import { nativeReplacementPending } from "./native-runtime/native-safe-replacement.js"; +import { hasPendingPauseResume } from "./issue-tree-resume.js"; export function registerExecutionControlWork( coordinator: ReturnType, @@ -81,7 +82,7 @@ export function executionControlSchedule(db: Db) { eq(issueRecoveryActions.status, "resolved"), sql`${issueRecoveryActions.evidence}->>'continuationDelivery' = 'pending'`, )).limit(1); - return rows.length ? retry() : null; + return rows.length || await hasPendingPauseResume(db) ? retry() : null; }, async status() { const rows = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index 1d27e125d02..f579d908d14 100644 --- a/server/src/services/execution-recovery-resolution.ts +++ b/server/src/services/execution-recovery-resolution.ts @@ -600,6 +600,9 @@ export async function settleUnrecoverableExecutions( }); return; } + // A legacy executor still holding its controller lease may be unwinding + // a Stop and can prove that no adapter started. Let it settle first. + if (run.runtimeMode === "legacy" && run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > now) return; // Adoption transfers maintenance to the retained-workspace owner. // This compatibility resolver must not close its hold or relabel it // as a user repair merely because old return-owner metadata is absent. @@ -631,6 +634,7 @@ export async function settleUnrecoverableExecutions( await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: task.companyId, issueId: task.id }); let nativeFailureBlock = action.evidence.nativeFailureBlock; let nativeBootstrapFailureBlock = action.evidence.nativeBootstrapFailureBlock; + let legacyCancellationBlock = action.evidence.legacyCancellationBlock; if (current) { await notifyNativeStatusReassessment(tx, { companyId: task.companyId, issueId: task.id }); if (task.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: task.companyId, issueId: task.id }); @@ -651,6 +655,9 @@ export async function settleUnrecoverableExecutions( if (task.status !== "blocked" && run.runtimeMode === "native") { nativeFailureBlock = { runId: run.id, statusVersion: projected!.statusVersion }; } + if (task.status !== "blocked" && run.runtimeMode === "legacy" && run.status === "cancelled") { + legacyCancellationBlock = { runId: run.id, statusVersion: projected!.statusVersion, previousStatus: task.status }; + } if (task.status !== "blocked" && isInterruptedNativePreparationCandidate(run)) { nativeBootstrapFailureBlock = { runId: run.id, statusVersion: projected!.statusVersion, previousStatus: task.status }; } @@ -670,6 +677,7 @@ export async function settleUnrecoverableExecutions( ...action.evidence, ...(nativeFailureBlock ? { nativeFailureBlock } : {}), ...(nativeBootstrapFailureBlock ? { nativeBootstrapFailureBlock } : {}), + ...(legacyCancellationBlock ? { legacyCancellationBlock } : {}), automaticRecovery: { policy: "preserve_without_replay_v1", runId: run.id, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 6d0821d2764..42062c37ec0 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -303,7 +303,9 @@ import { import { settleInterruptedNativeBootstrap, + settleCancelledLegacyPreparation, } from "./legacy-execution-recovery.js"; +import { resumePauseInterruptedWork } from "./issue-tree-resume.js"; import { adapterExecutionControls, createAdapterExecutionControl, @@ -3962,6 +3964,19 @@ export function heartbeatService( providerDispatchStarted: legacyAdapterEntered || nativeDispatchStarted || nativeOwnershipHeld, }) ?? latestRun; } + if (latestRun?.status === "cancelled" && latestRun.runtimeMode === "legacy") { + const settled = await settleCancelledLegacyPreparation(db, { run: latestRun, + providerDispatchStarted: legacyAdapterEntered || nativeDispatchStarted || nativeOwnershipHeld, + }); + latestRun = settled ?? latestRun; + // A pause may already have been released while this Stop unwound. + const settledIssueId = settled?.issueId ?? settled?.contextSnapshot?.issueId; + if (typeof settledIssueId === "string") { + await resumePauseInterruptedWork(db, { companyId: settled!.companyId, issueIds: [settledIssueId], + actor: { actorType: "system", actorId: "system" }, wakeup: enqueueWakeup, + }).catch(error => logger.warn({ err: error, runId: run.id }, "Could not resume work after a released pause")); + } + } if (latestRun?.status === "cancelled" && !nativeDispatchStarted && !nativeOwnershipHeld && (latestRun.runtimeMode === "native" || parseObject(latestRun.resultJson?.startupCancellation).beforeNativeSelection === true)) { diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index eb2965650cf..e66d7cb5e69 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -34,6 +34,7 @@ import { import { conflict, notFound, unprocessable } from "../errors.js"; import type { IssuePostCommitAction } from "./issues.js"; import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; +import { readRunCancellation } from "./run-cancellation.js"; type IssueRow = typeof issues.$inferSelect; type HoldRow = typeof issueTreeHolds.$inferSelect; @@ -1208,9 +1209,12 @@ export function issueTreeControlService(db: Db) { releaseMetadata: input.metadata ?? null, updatedAt: new Date(), }) - .where(and(eq(issueTreeHolds.id, holdId), eq(issueTreeHolds.companyId, companyId))) + .where(and(eq(issueTreeHolds.id, holdId), eq(issueTreeHolds.companyId, companyId), + eq(issueTreeHolds.status, "active"))) .returning(); }); + // A concurrent release won; it alone owns the resumed work. + if (!updated) throw conflict("Issue tree hold is already released"); const members = await db .select() @@ -1221,6 +1225,47 @@ export function issueTreeControlService(db: Db) { return toHold(updated, members); } + /** + * A pause interrupts runs as an operator hold, not as a failure. A pause + * hold's member snapshot names each run it interrupted, and that pending work + * is owed one re-dispatch once no pause covers the task. The obligation + * belongs to the run, so overlapping holds keep it until the last release. + * Current state governs: the interrupted run must still be the task's latest + * run, the task must be executable by the same agent, and the claim marker + * makes the resume happen at most once. Execution blockers are the caller's. + */ + async function findOwedPauseResume(companyId: string, issueId: string) { + const [issue] = await db.select({ + status: issues.status, assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId, + }).from(issues).where(and(eq(issues.companyId, companyId), eq(issues.id, issueId))); + if (!issue?.assigneeAgentId || issue.assigneeUserId + || !ISSUE_TREE_RESUME_EXECUTABLE_STATUSES.includes(issue.status)) return null; + const [run] = await db.select({ + id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, status: heartbeatRuns.status, + finishedAt: heartbeatRuns.finishedAt, resultJson: heartbeatRuns.resultJson, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), + sql`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueId}`, + )).orderBy(sql`${heartbeatRuns.createdAt} desc`).limit(1); + if (!run || run.status !== "cancelled" || !run.finishedAt || run.agentId !== issue.assigneeAgentId + || run.resultJson?.treeHoldResume !== undefined + || readRunCancellation(run.resultJson)?.source !== "operator") return null; + // Only a Stop recorded while a pause naming this run was active is owed. + const [hold] = await db.select({ id: issueTreeHolds.id }).from(issueTreeHoldMembers) + .innerJoin(issueTreeHolds, eq(issueTreeHolds.id, issueTreeHoldMembers.holdId)) + .where(and( + eq(issueTreeHoldMembers.companyId, companyId), eq(issueTreeHoldMembers.issueId, issueId), + eq(issueTreeHoldMembers.activeRunId, run.id), eq(issueTreeHoldMembers.skipped, false), + eq(issueTreeHolds.companyId, companyId), eq(issueTreeHolds.mode, "pause"), + sql`${issueTreeHolds.createdAt} <= ${run.finishedAt.toISOString()}::timestamptz`, + or(isNull(issueTreeHolds.releasedAt), + sql`${issueTreeHolds.releasedAt} >= ${run.finishedAt.toISOString()}::timestamptz`), + )).orderBy(asc(issueTreeHolds.createdAt)).limit(1); + if (!hold) return null; + if (await getActivePauseHoldGate(companyId, issueId)) return null; + return { issueId, agentId: run.agentId, interruptedRunId: run.id, holdId: hold.id }; + } + async function cancelUnclaimedWakeupsForTree(companyId: string, rootIssueId: string, reason: string) { const treeIssues = await listTreeIssues(companyId, rootIssueId); const issueIds = treeIssues.map((issue) => issue.id); @@ -1268,6 +1313,7 @@ export function issueTreeControlService(db: Db) { listHolds, getActivePauseHoldGate, releaseHold, + findOwedPauseResume, cancelUnclaimedWakeupsForTree, }; } diff --git a/server/src/services/issue-tree-resume.ts b/server/src/services/issue-tree-resume.ts new file mode 100644 index 00000000000..ff73374fb6f --- /dev/null +++ b/server/src/services/issue-tree-resume.ts @@ -0,0 +1,239 @@ +import { and, asc, eq, gt, ne, sql } from "drizzle-orm"; +import { agentWakeupRequests, heartbeatRuns, issues, type Db } from "@paperclipai/db"; +import { isUniqueViolation } from "../db-errors.js"; +import { logActivity } from "./activity-log.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import { getExecutionBlocker } from "./execution-blocker.js"; +import { ISSUE_TREE_RESUME_EXECUTABLE_STATUSES, issueTreeControlService } from "./issue-tree-control.js"; + +type ResumeActor = { + actorType: "user" | "agent" | "system"; + actorId: string; + agentId?: string | null; + runId?: string | null; + agentApiKeyId?: string | null; +}; + +type Wakeup = (agentId: string, opts: { + source: "assignment"; + triggerDetail: "system"; + reason: string; + payload: Record; + requestedByActorType: ResumeActor["actorType"]; + requestedByActorId: string; + idempotencyKey: string; + contextSnapshot: Record; +}) => Promise; + +/** A claim older than this with no saved wake belongs to an interrupted dispatcher. */ +export const PAUSE_RESUME_CLAIM_STALE_MS = 60_000; + +/** Queue admission accepts one saved wake per interrupted run under this key. */ +export const PAUSE_RESUME_WAKE_INDEX = "agent_wakeup_requests_tree_hold_resume_uq"; + +const SWEEP_PAGE_SIZE = 25; + +export function pauseResumeWakeKey(runId: string) { + return `tree-hold-resume:${runId}`; +} + +const SYSTEM_ACTOR: ResumeActor = { actorType: "system", actorId: "system" }; + +function activity(db: Db, companyId: string, actor: ResumeActor, action: string, issueId: string, + details: Record) { + return logActivity(db, { + companyId, actorType: actor.actorType, actorId: actor.actorId, agentId: actor.agentId ?? null, + runId: actor.runId ?? null, agentApiKeyId: actor.agentApiKeyId ?? null, action, + entityType: "issue", entityId: issueId, details, + }); +} + +/** Compare-and-set the run's resume record. Only the writer that wins dispatches. */ +async function setResumeState(db: Db, input: { + companyId: string; runId: string; next: Record; expectedClaimedAt?: string | null; +}) { + const guard = input.expectedClaimedAt === undefined + ? sql`not (coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) ? 'treeHoldResume')` + : sql`${heartbeatRuns.resultJson}->'treeHoldResume'->>'claimedAt' is not distinct from ${input.expectedClaimedAt}`; + return db.transaction(async tx => { + // A pending claim arms the recovery sweep in the same commit, so a crash + // before the wake is saved is retried. Intent is registered before the write. + if (input.next.state === "pending") await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation); + const [row] = await tx.update(heartbeatRuns).set({ + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ treeHoldResume: input.next })}::jsonb`, + }).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.status, "cancelled"), guard)).returning({ id: heartbeatRuns.id }); + return Boolean(row); + }); +} + +/** A saved, admitted resume wake (queued, deferred, coalesced, or run) for the interrupted run. */ +async function findPauseResumeReceipt(db: Db, companyId: string, agentId: string, runId: string) { + const [receipt] = await db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId), + eq(agentWakeupRequests.idempotencyKey, pauseResumeWakeKey(runId)), + ne(agentWakeupRequests.status, "skipped"))).limit(1); + return Boolean(receipt); +} + +async function dispatch(db: Db, input: { + companyId: string; issueId: string; agentId: string; runId: string; holdId: string; + claimedAt: string; rootIssueId: string | null; actor: ResumeActor; wakeup: Wakeup; +}) { + const saved = await input.wakeup(input.agentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_tree_resumed", + payload: { issueId: input.issueId, rootIssueId: input.rootIssueId, holdId: input.holdId, interruptedRunId: input.runId }, + requestedByActorType: input.actor.actorType, + requestedByActorId: input.actor.actorId, + // The saved wake request is the durable receipt the recovery sweep checks. + idempotencyKey: pauseResumeWakeKey(input.runId), + contextSnapshot: { + issueId: input.issueId, + taskId: input.issueId, + wakeReason: "issue_tree_resumed", + source: "issue.tree_resume", + rootIssueId: input.rootIssueId, + holdId: input.holdId, + interruptedRunId: input.runId, + }, + }).then((run) => run ? "saved" as const : "not_admitted" as const, (error: unknown) => { + // Another dispatcher already saved this run's resume wake; it owns the work. + if (isUniqueViolation(error, PAUSE_RESUME_WAKE_INDEX)) return "duplicate" as const; + throw error; + }); + if (saved === "duplicate") return; + // Admission can skip the wake, for example when a new pause landed after the + // eligibility check. A skipped request is no receipt, so the claim stays + // pending and the sweep retries once the task is runnable again. + if (saved === "not_admitted" && !await findPauseResumeReceipt(db, input.companyId, input.agentId, input.runId)) return; + await setResumeState(db, { companyId: input.companyId, runId: input.runId, expectedClaimedAt: input.claimedAt, + next: { holdId: input.holdId, state: "delivered", claimedAt: input.claimedAt, deliveredAt: new Date().toISOString() } }); +} + +/** + * Re-dispatch, once, the work a subtree pause interrupted on each task that is + * no longer paused. Both a hold release and the executor that settles a + * cancelled preparation call this, so neither ordering loses the work. A task + * that still needs execution reconciliation keeps its blocker and is skipped; + * its obligation stays unclaimed for a later settlement or release. A claim + * whose wake was never saved (for example after a restart) stays pending and + * is finished by `deliverPendingPauseResumes`. + */ +export async function resumePauseInterruptedWork( + db: Db, + input: { companyId: string; issueIds: string[]; actor: ResumeActor; wakeup: Wakeup; rootIssueId?: string | null }, +) { + const treeControl = issueTreeControlService(db); + const failures: Array<{ issueId: string; message: string }> = []; + for (const issueId of [...new Set(input.issueIds)]) { + try { + const target = await treeControl.findOwedPauseResume(input.companyId, issueId); + if (!target) continue; + const blocker = await getExecutionBlocker(db, input.companyId, issueId); + if (blocker) { + await activity(db, input.companyId, input.actor, "issue.tree_resume_wake_skipped", issueId, { + holdId: target.holdId, interruptedRunId: target.interruptedRunId, cause: blocker.cause, + }); + continue; + } + const claimedAt = new Date().toISOString(); + if (!await setResumeState(db, { companyId: input.companyId, runId: target.interruptedRunId, + next: { holdId: target.holdId, state: "pending", claimedAt } })) continue; + await dispatch(db, { companyId: input.companyId, issueId, agentId: target.agentId, runId: target.interruptedRunId, + holdId: target.holdId, claimedAt, rootIssueId: input.rootIssueId ?? null, actor: input.actor, wakeup: input.wakeup }); + } catch (error) { + // A pending claim stays recoverable; the sweep retries it after it goes stale. + const message = error instanceof Error ? error.message : String(error); + failures.push({ issueId, message }); + await Promise.resolve(activity(db, input.companyId, input.actor, "issue.tree_resume_wake_failed", issueId, + { error: message })).catch(() => null); + } + } + return failures; +} + +function pendingResumePredicate(staleBefore: Date) { + return and(eq(heartbeatRuns.status, "cancelled"), + sql`${heartbeatRuns.resultJson}->'treeHoldResume'->>'state' = 'pending'`, + sql`(${heartbeatRuns.resultJson}->'treeHoldResume'->>'claimedAt')::timestamptz <= ${staleBefore.toISOString()}::timestamptz`); +} + +/** True while a claimed resume has no durable receipt yet. */ +export async function hasPendingPauseResume(db: Db) { + const [row] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(sql`${heartbeatRuns.status} = 'cancelled' and ${heartbeatRuns.resultJson}->'treeHoldResume'->>'state' = 'pending'`) + .limit(1); + return Boolean(row); +} + +/** + * Finish resume claims whose dispatcher stopped between claiming and saving the + * wake (for example a restart). A saved wake with the stable key is the receipt. + * Otherwise the task's current state is checked again before one dispatch. + */ +export async function deliverPendingPauseResumes(db: Db, wakeup: Wakeup, now = new Date()) { + const staleBefore = new Date(now.getTime() - PAUSE_RESUME_CLAIM_STALE_MS); + // Page through the stale claims: a claim still waiting on a pause or blocker + // is skipped without a write, so any fixed batch could starve the rest. Each + // pass visits every stale claim in bounded pages. + let after: string | null = null; + for (;;) { + const rows: PendingResumeRow[] = await db.select({ + id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + createdAt: heartbeatRuns.createdAt, resultJson: heartbeatRuns.resultJson, + issueId: sql`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')`, + }).from(heartbeatRuns) + .where(and(pendingResumePredicate(staleBefore), after ? gt(heartbeatRuns.id, after) : undefined)) + .orderBy(asc(heartbeatRuns.id)).limit(SWEEP_PAGE_SIZE); + if (!rows.length) return; + after = rows[rows.length - 1]!.id; + await deliverPendingPage(db, wakeup, now, rows); + if (rows.length < SWEEP_PAGE_SIZE) return; + } +} + +type PendingResumeRow = { + id: string; companyId: string; agentId: string; createdAt: Date; + resultJson: Record | null; issueId: string | null; +}; + +async function deliverPendingPage(db: Db, wakeup: Wakeup, now: Date, rows: PendingResumeRow[]) { + for (const run of rows) { + const claim = run.resultJson?.treeHoldResume as { holdId?: unknown; claimedAt?: unknown } | undefined; + if (typeof claim?.claimedAt !== "string" || typeof claim.holdId !== "string" || !run.issueId) continue; + const finish = (state: "delivered" | "abandoned", reason?: string) => setResumeState(db, { + companyId: run.companyId, runId: run.id, expectedClaimedAt: claim.claimedAt as string, + next: { holdId: claim.holdId, state, claimedAt: claim.claimedAt, settledAt: now.toISOString(), ...(reason ? { reason } : {}) }, + }); + try { + if (await findPauseResumeReceipt(db, run.companyId, run.agentId, run.id)) { await finish("delivered"); continue; } + const [issue] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId }).from(issues) + .where(and(eq(issues.companyId, run.companyId), eq(issues.id, run.issueId))); + const [newer] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), + sql`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${run.issueId}`, + // JavaScript dates drop microseconds, so exclude the run itself by id. + sql`${heartbeatRuns.id} <> ${run.id}`, + sql`${heartbeatRuns.createdAt} >= ${run.createdAt.toISOString()}::timestamptz`)).limit(1); + if (!issue || issue.assigneeAgentId !== run.agentId || issue.assigneeUserId || newer + || !ISSUE_TREE_RESUME_EXECUTABLE_STATUSES.includes(issue.status)) { + await finish("abandoned", "task_changed"); + continue; + } + // A paused or blocked task keeps its claim; release or settlement revisits it. + if (await issueTreeControlService(db).getActivePauseHoldGate(run.companyId, run.issueId) + || await getExecutionBlocker(db, run.companyId, run.issueId)) continue; + const claimedAt = now.toISOString(); + if (!await setResumeState(db, { companyId: run.companyId, runId: run.id, expectedClaimedAt: claim.claimedAt, + next: { holdId: claim.holdId, state: "pending", claimedAt } })) continue; + await dispatch(db, { companyId: run.companyId, issueId: run.issueId, agentId: run.agentId, runId: run.id, + holdId: claim.holdId, claimedAt, rootIssueId: null, actor: SYSTEM_ACTOR, wakeup }); + } catch (error) { + await Promise.resolve(activity(db, run.companyId, SYSTEM_ACTOR, "issue.tree_resume_wake_failed", run.issueId, + { error: error instanceof Error ? error.message : String(error), interruptedRunId: run.id })).catch(() => null); + } + } +} diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 1059a7978a5..0b3255ae97a 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -16,7 +16,10 @@ import { claimedAdapterType, hasConversationContinuationPolicy } from "./convers import { canonicalExecutionUuid } from "./execution-identity.js"; import { randomUUID } from "node:crypto"; import { and, desc, eq, getTableColumns, inArray, isNull, or, sql, type SQL } from "drizzle-orm"; -import { environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { environmentLeases, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { executionStartedEventPredicate } from "./pre-dispatch-execution-evidence.js"; +import { readRunCancellation } from "./run-cancellation.js"; +import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { issueRecoveryActionService } from "./issue-recovery-actions.js"; import { parseIssueExecutionState } from "./issue-execution-policy.js"; import { executionFailureRetryCount } from "./execution-recovery-attempt.js"; @@ -382,6 +385,87 @@ export async function settleInterruptedNativeBootstrap( }); } +/** + * The executor that owned a cancelled legacy run proves that its gated + * dispatch never entered the adapter. Such a Stop (for example a subtree pause + * during workspace preparation) left no provider actions to reconcile, so the + * hold that terminalization conservatively recorded is a false positive. Every + * retained execution reference must agree; otherwise the hold stays in force. + */ +export async function settleCancelledLegacyPreparation( + db: Db, input: { run: Run; providerDispatchStarted: boolean }, +): Promise { + if (input.providerDispatchStarted || input.run.runtimeMode !== "legacy" || input.run.status !== "cancelled") return null; + const issueId = input.run.issueId ?? input.run.contextSnapshot?.issueId; + if (typeof issueId !== "string") return null; + return db.transaction(async tx => { + const [task] = await tx.select().from(issues) + .where(and(eq(issues.id, issueId), eq(issues.companyId, input.run.companyId))).for("update"); + const [run] = await tx.select().from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, input.run.id), eq(heartbeatRuns.companyId, input.run.companyId))).for("update"); + const recovery = run?.resultJson?.executionRecovery as Record | undefined; + // Only a requested Stop qualifies, and only while it is still classified as + // an unknown outcome: existing wait or bootstrap receipts already decide it. + if (!task || !run || run.runtimeMode !== "legacy" || run.status !== "cancelled" || !run.finishedAt + || readRunCancellation(run.resultJson)?.expected !== true || recovery !== undefined + || !legacyExecutionNeedsReconciliation(run) + || run.processPid || run.processGroupId || run.processStartedAt || run.sessionIdAfter + || run.nativeIssueId || run.nativeSessionId + || hasWorkspaceRestoreFailure(run.resultJson) || hasRequiredWorkspaceRecovery(run.resultJson) + || run.runnerProfileJson?.nativeExecutionInput != null) return null; + const [native] = await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .where(and(eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id))).limit(1); + const leases = await tx.select().from(environmentLeases) + .where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id))); + const [execution] = await tx.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), executionStartedEventPredicate(), + )).limit(1); + if (native || execution || leases.some(lease => lease.provider === "local" + ? !lease.releasedAt || lease.status === "pending_cleanup" || lease.cleanupStatus === "failed" + : !hasRemoteTerminationReceipt(lease))) return null; + const now = new Date(); + const [settled] = await tx.update(heartbeatRuns).set({ resultJson: { ...run.resultJson, + executionRecovery: { kind: "bootstrap", providerWorkStarted: false, preparationSettledAt: now.toISOString() }, + }, updatedAt: now }).where(eq(heartbeatRuns.id, run.id)).returning(); + // The automatic disposition may already have closed the incident with + // replay blocked, and moved the task to blocked. Correct both verdicts. + const holds = await tx.select().from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, task.id), + eq(issueRecoveryActions.cause, LEGACY_RECOVERY_CAUSE), eq(issueRecoveryActions.fingerprint, `legacy-execution:${run.id}`), + sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, + sql`not (${issueRecoveryActions.evidence} ? 'executionReconciliation')`, + or(inArray(issueRecoveryActions.status, ["active", "escalated"]), and( + eq(issueRecoveryActions.status, "resolved"), eq(issueRecoveryActions.outcome, "blocked"), + sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'policy' = 'preserve_without_replay_v1'`, + sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`, + )), + )).for("update"); + const ownedBlock = holds.map(hold => hold.evidence.legacyCancellationBlock as Record | undefined) + .find(receipt => receipt?.runId === run.id && receipt.statusVersion === task.statusVersion && + ["todo", "in_progress", "in_review"].includes(String(receipt.previousStatus))); + if (task.status === "blocked" && ownedBlock && task.assigneeAgentId === run.agentId && + !isSupersededConversationRun(task, run) && !task.executionRunId && !task.checkoutRunId) { + // Restore only the unchanged projection made by this exact incident. + await issueService(tx as unknown as Db).update(task.id, { status: String(ownedBlock.previousStatus) }, tx); + } + const resolved = holds.length ? await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "false_positive", + resolutionNote: "The run was stopped before its adapter started; there are no provider actions to reconcile.", + resolvedAt: now, updatedAt: now, + evidence: sql`case when ${issueRecoveryActions.evidence} ? 'automaticRecovery' + then jsonb_set(${issueRecoveryActions.evidence}, '{automaticRecovery,replay}', '"allowed"'::jsonb, false) + else ${issueRecoveryActions.evidence} end`, + }).where(inArray(issueRecoveryActions.id, holds.map(hold => hold.id))).returning({ id: issueRecoveryActions.id }) : []; + if (resolved.length) { + await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId: task.id }); + await logActivity(tx as unknown as Db, { companyId: run.companyId, actorType: "system", actorId: "system", + action: "issue.recovery_action_resolved", entityType: "issue", entityId: task.id, agentId: run.agentId, runId: run.id, + details: { reason: "legacy_cancelled_before_adapter", recoveryActionIds: resolved.map(r => r.id) }, + }); + } + return settled; + }); +} + /** Capture the required-file obligation before unrelated host writes. Keeps * execution ownership while the adapter's host finalization is still running. */ export async function recordLegacyWorkspaceRestoreFailure(db: Db, run: Run, evidence: Record, source?: LegacyWorkspaceRestoreSource) {