From d20cb69ce5b1cdde08eea36c5a4badc72e3d7326 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 19 Oct 2026 11:34:27 -0500 Subject: [PATCH] fix(heartbeat): preserve active retries when comments arrive (#16855) ## Thinking Path > - Paperclip manages agents and their work. > - Heartbeat retries preserve task ownership after a recoverable failure. > - The issue read model includes retries that are scheduled, queued, or running. > - The human comment route treated every retry in that model as parked work. > - A normal status comment could therefore cancel a running continuation and return its task to Todo. > - This pull request limits comment supersession to parked retries and checks that state at the cancellation fence. > - The agent can keep working when a person adds context. Explicit Stop and Interrupt remain available. ## Linked Issues or Issue Description **What happened?** A Board comment on an in-progress task cancelled its running recovery continuation. The handler read a live retry from `getCurrentScheduledRetry` but did not check its status. The cancellation also forced workspace recovery when copy-back did not complete. **Expected behavior** A comment must preserve queued and running execution. Only a retry that is still `scheduled_retry` may be superseded. If promotion wins the race, the comment must keep the task in progress and follow the normal wake path. **Steps to reproduce** 1. Let a task retry advance from `scheduled_retry` to `running`. 2. Keep its retry reason and source run identity in the stored row. 3. Post a normal human comment, without an Interrupt request. 4. Observe the old handler cancel the run and move the task to Todo. **Paperclip version or commit** Reproduced on master base `9966f31bee`. The staging incident served `f3cd3f4ea89539b68410841d39ec15a5c626007f`. **Deployment mode** Staging server with a remote runner. The defect is in the core comment and cancellation path. Refs #16835 and #16843. Those repairs cover runner health and workspace preservation. This change prevents a normal comment from requesting the wrong cancellation. Related #9732 promotes human wakes on parked retries; it does not protect a retry that is already running. ## What Changed - Check the live retry status in both comment routes: POST comments and PATCH issue with a comment. - Add an optional expected-status fence to cancellation. A concurrent promotion or terminal outcome returns a no-op result. - Make due-retry promotion honor a cancellation claim under its issue and run row locks. A scheduler cannot promote work after cancellation has won. - Finish persisted cancellations of parked retries through the existing startup and periodic recovery sweep. Preserve their original request and retry a failed settlement without starting provider work. - Keep the task active when guarded cancellation does not win. Do not record a cancellation that did not occur. - Add route regressions, real PostgreSQL race tests, and pending-retry and explicit Stop safety coverage. - Exercise both normal HTTP comment routes with real services, stored Board membership, PostgreSQL, and a live child process. Assert that the process can still perform work, the task keeps its original execution, and no second executor starts. - Document the comment supersession contract and update the task feature/access maps with the local regression coverage and deployed acceptance recipe. ## Verification - Before the fix, all four POST/PATCH regressions cancelled queued or running work. - Before the fix, the real PostgreSQL race test cancelled a retry that became running after lookup. - The comment-route suite passes **103 tests**. The final dispatch, cancellation, restart recovery and HTTP/live-process suites pass **119 tests**, for **222 affected tests** across these runs. They cover comment delivery, both sides of the promotion race, concurrent terminal outcomes, pending retries, explicit Stop, and a live process that can continue work. - Command: `node node_modules/vitest/vitest.mjs run server/src/__tests__/issue-comment-reopen-routes.test.ts server/src/services/heartbeat/run-control.test.ts`. - The normal HTTP/live-process acceptance suite passes all **14 tests**, including both comment endpoints with stored Board membership and deterministic wakeup teardown. - Review exposed a second race after the cancellation claim. Its new PostgreSQL test failed before the dispatch guard with `promoted` and passes afterward with `not_promoted`, followed by completed cancellation. - A real PostgreSQL constraint interrupts the terminal cancellation write after its intent commits. The assembled recovery service failed to settle that row before the repair. Afterward a fresh service settles it, clears its task lock, cancels its saved wake, preserves the original request and leaves unrelated parked retries alone. A repeated database failure preserves the intent for a later successful sweep; a second successful sweep adds no duplicate event. - Full `pnpm -r typecheck` and `pnpm build` pass. Server typecheck and build pass again after the final server-only review fix. A local full-suite attempt started on `f8773bf` did not return an overall result; its log is preserved and no local full-suite pass is claimed. The new recovery change passes the focused suites and fresh server typecheck/build. Latest-head CI on `8f928adee67c089e1ac405d2ad3b169eedb8da27` is green: the full PR workflow and commitperclip review passed; only the two optional Storybook jobs skipped. - Greptile reviewed final head `8f928adee67c089e1ac405d2ad3b169eedb8da27` at **5/5**, with no unresolved findings. All actual latest-head CI checks passed; the two optional Storybook jobs skipped. - `node scripts/check-maps.mjs` passes after the map updates (34 areas, 172 capability IDs); it checks structure only. - Fixed-revision staging acceptance is pending deployment. The original workspace remains held and its latest source checkpoint is backed up and verified. ## Risks - A promoted retry must keep its execution and ownership when a comment arrives. The database status predicate and route no-op handling enforce this at the race boundary. - A cancellation intent must prevent promotion without leaving work parked forever after a restart. The recovery sweep consumes at most 20 such intents per pass through guarded cancellation, and a failed settlement keeps its intent for retry. - A parked retry must still accept fresh instructions. Its existing cancellation and wake behavior remains covered. - Ordinary Stop and Interrupt calls omit the expected-status option. Their cancellation authority is unchanged and remains covered by the run-control suite. - No schema migration or API request contract changes are required. ## Model Used OpenAI `gpt-6.1-sol` in Codex, with extra-high reasoning, tool use, and code execution. The session does not expose its context-window limit. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/execution-semantics.md | 11 +++ maps/agent-access-map/tasks.md | 5 +- maps/feature-map/tasks.md | 8 ++ .../issue-comment-reopen-routes.test.ts | 60 +++++++++++- .../issue-scheduled-retry-routes.test.ts | 92 ++++++++++++++++++- .../modules/run-dispatch/adapters/postgres.ts | 3 + server/src/routes/issues.ts | 36 +++++--- server/src/services/heartbeat/recovery.ts | 29 +++++- .../services/heartbeat/run-control.test.ts | 83 +++++++++++++++++ server/src/services/heartbeat/run-control.ts | 13 ++- 10 files changed, 316 insertions(+), 24 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index e9dd8187545..519e58b67a5 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -242,6 +242,17 @@ The **Retry now** action returns `waiting` with the saved retry schedule while cleanup holds ownership. Its controls show the wait without claiming that a run started, and another click can check the gate again. +A human comment may supersede a retry only while it remains `scheduled_retry`. +Queued and running retries retain their execution and receive the comment through +normal wake coalescing. Supersession rechecks the scheduled status atomically with +the cancellation fence; a concurrent promotion must not cancel the active run or +move its task back to `todo`. Explicit Stop and Interrupt keep their authority. +Once cancellation wins that fence, due-retry promotion respects its claim until +the cancellation owner finishes settlement. Startup and periodic recovery finish +persisted cancellations of parked retries through the normal guarded cancellation +path if the original owner stops or its final database write fails. They preserve +the original cancellation request and never start a provider turn for that work. + Recovery instructions in a wake describe only an active or escalated recovery action. Resolving or cancelling that action removes its repair instructions from subsequent wakes, so the original worker can resume the task. This prompt change diff --git a/maps/agent-access-map/tasks.md b/maps/agent-access-map/tasks.md index c2148348330..28ade0fa130 100644 --- a/maps/agent-access-map/tasks.md +++ b/maps/agent-access-map/tasks.md @@ -22,7 +22,10 @@ API: `present` — representative operation IDs: Agent access: `available` — Native create_task; REST / CLI / MCP. Policy: Company/task visibility, assignment/change policy, active-run checkout ownership -and lifecycle rules. +and lifecycle rules. Ordinary Board comments preserve queued/running retries; +only a parked retry can be superseded. Explicit Stop/Interrupt retain their +existing actor and execution authority. Agent comment tools do not acquire Board +identity or cancellation authority through this behavior. Evidence: `source-only` — [implementation](../../server/src/routes/issues.ts) at the reviewed commit above. diff --git a/maps/feature-map/tasks.md b/maps/feature-map/tasks.md index 0b2d90936ad..ded0795a166 100644 --- a/maps/feature-map/tasks.md +++ b/maps/feature-map/tasks.md @@ -70,6 +70,14 @@ reassignment and confirm that the outgoing run still stops. Manual: Add a comment and edit authorized properties. Reload and verify history, actor, and values. Complete a disposable task and inspect its deliverable; reopen through an available action and confirm the next execution follows current policy. +Manual: Add an ordinary Board comment while a task retry is queued or running. +Confirm that the same execution keeps working and receives the saved message, +without a second executor or an implicit move to Todo. Repeat with a parked retry +and verify that fresh instructions can supersede it. Explicit Stop must still +cancel active work. The [comment route regressions](../../server/src/__tests__/issue-comment-reopen-routes.test.ts) +and [HTTP/process and interrupted-cancellation tests](../../server/src/__tests__/issue-scheduled-retry-routes.test.ts) +cover these boundaries locally; deployed runner acceptance remains separate. + ## Gotchas - Priority remains an API field, but ordinary priority UI editing is hidden. diff --git a/server/src/__tests__/issue-comment-reopen-routes.test.ts b/server/src/__tests__/issue-comment-reopen-routes.test.ts index 0043cd24623..2585c5e5f20 100644 --- a/server/src/__tests__/issue-comment-reopen-routes.test.ts +++ b/server/src/__tests__/issue-comment-reopen-routes.test.ts @@ -1215,7 +1215,7 @@ describe("issue comment reopen routes", () => { "11111111-1111-4111-8111-111111111111", { status: "todo" }, ); - expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1"); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1", undefined, { expectedStatus: "scheduled_retry" }); expect(mockLogActivity).toHaveBeenCalledWith( expect.anything(), expect.objectContaining({ @@ -1246,6 +1246,58 @@ describe("issue comment reopen routes", () => { ); }); + it.each([ + ["post", "queued"], + ["post", "running"], + ["patch", "queued"], + ["patch", "running"], + ] as const)("preserves a %s comment's %s retry execution", async (method, status) => { + const issue = { + ...makeIssue("in_progress"), + executionRunId: "retry-run-1", + }; + mockIssueService.getById.mockResolvedValue(issue); + mockIssueService.getCurrentScheduledRetry.mockResolvedValue({ + ...createScheduledRetry(), + status, + }); + mockIssueService.update.mockImplementation(async (_id, patch) => ({ ...issue, ...patch })); + + const app = await installActor(createApp()); + const res = method === "post" + ? await request(app).post(`/api/issues/${issue.id}/comments`).send({ body: "Recovery evidence is attached." }) + : await request(app).patch(`/api/issues/${issue.id}`).send({ comment: "Recovery evidence is attached." }); + + expect(res.status).toBe(method === "post" ? 201 : 200); + expect(mockHeartbeatService.cancelRun).not.toHaveBeenCalled(); + expect(mockIssueService.update).not.toHaveBeenCalledWith(issue.id, expect.objectContaining({ status: "todo" })); + await waitForWakeup(() => expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith( + issue.assigneeAgentId, + expect.objectContaining({ reason: "issue_commented" }), + )); + }); + + it.each(["post", "patch"] as const)("keeps the task active when a retry is promoted during a %s comment", async method => { + const issue = { ...makeIssue("in_progress"), executionRunId: "retry-run-1" }; + mockIssueService.getById.mockResolvedValue(issue); + mockIssueService.getCurrentScheduledRetry.mockResolvedValue(createScheduledRetry()); + mockHeartbeatService.cancelRun.mockResolvedValue({ id: "retry-run-1", status: "running" }); + mockIssueService.update.mockImplementation(async (_id, patch) => ({ ...issue, ...patch })); + const app = await installActor(createApp()); + + const res = method === "post" + ? await request(app).post(`/api/issues/${issue.id}/comments`).send({ body: "Additional context for the active run." }) + : await request(app).patch(`/api/issues/${issue.id}`).send({ comment: "Additional context for the active run." }); + + expect(res.status).toBe(method === "post" ? 201 : 200); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1", undefined, { expectedStatus: "scheduled_retry" }); + expect(mockIssueService.update).not.toHaveBeenCalledWith(issue.id, expect.objectContaining({ status: "todo" })); + expect(mockLogActivity).not.toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ action: "heartbeat.cancelled" })); + await waitForWakeup(() => expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith( + issue.assigneeAgentId, expect.objectContaining({ reason: "issue_commented" }), + )); + }); + it("does not move scheduled-retry issues to todo when POST comment retry cancellation fails", async () => { const issue = { ...makeIssue("in_progress"), @@ -1262,7 +1314,7 @@ describe("issue comment reopen routes", () => { .send({ body: "I added the missing detail; please continue." }); expect(res.status).toBe(500); - expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1"); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1", undefined, { expectedStatus: "scheduled_retry" }); expect(mockIssueService.update).not.toHaveBeenCalled(); expect(mockIssueService.addComment).not.toHaveBeenCalled(); expect(mockLogActivity).not.toHaveBeenCalledWith( @@ -1945,7 +1997,7 @@ describe("issue comment reopen routes", () => { actorUserId: "local-board", }), ); - expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1"); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1", undefined, { expectedStatus: "scheduled_retry" }); await waitForWakeup(() => expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith( "22222222-2222-4222-8222-222222222222", @@ -1976,7 +2028,7 @@ describe("issue comment reopen routes", () => { .send({ comment: "Retry window is over; please continue." }); expect(res.status).toBe(500); - expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1"); + expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("retry-run-1", undefined, { expectedStatus: "scheduled_retry" }); expect(mockIssueService.update).not.toHaveBeenCalled(); expect(mockIssueService.addComment).not.toHaveBeenCalled(); expect(mockLogActivity).not.toHaveBeenCalledWith( diff --git a/server/src/__tests__/issue-scheduled-retry-routes.test.ts b/server/src/__tests__/issue-scheduled-retry-routes.test.ts index bd5e31432c7..65adabfc09c 100644 --- a/server/src/__tests__/issue-scheduled-retry-routes.test.ts +++ b/server/src/__tests__/issue-scheduled-retry-routes.test.ts @@ -1,13 +1,16 @@ import { randomUUID } from "node:crypto"; +import { spawn } from "node:child_process"; +import { once } from "node:events"; import express from "express"; import request from "supertest"; -import { and, eq, inArray } from "drizzle-orm"; +import { and, eq, inArray, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { activityLog, agents, agentWakeupRequests, companies, + companyMemberships, createDb, environmentLeases, heartbeatRunEvents, @@ -23,6 +26,8 @@ import { } from "./helpers/embedded-postgres.js"; import { errorHandler } from "../middleware/index.js"; import { issueRoutes } from "../routes/issues.js"; +import { runningProcesses } from "../adapters/index.js"; +import { heartbeatService } from "../services/heartbeat.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -55,6 +60,7 @@ describeEmbeddedPostgres("issue scheduled retry routes", () => { await db.delete(heartbeatRuns); await db.delete(agentWakeupRequests); await db.delete(agents); + await db.delete(companyMemberships); await db.delete(companies); }); @@ -207,6 +213,90 @@ describeEmbeddedPostgres("issue scheduled retry routes", () => { return { companyId, agentId, issueId, sourceRunId, retryRunId, scheduledRetryAt }; } + it("finishes a parked retry cancellation after its owner loses the final database write", async () => { + const f = await seedIssueWithRetry({ agentStatus: "paused" }); + const unrelated = await seedIssueWithRetry({ agentStatus: "paused" }); + await db.execute(sql`ALTER TABLE heartbeat_runs ADD CONSTRAINT reject_retry_cancellation CHECK (status <> 'cancelled') NOT VALID`); + try { + await expect(heartbeatService(db).cancelRun(f.retryRunId, "Superseded by a human comment", { + expectedStatus: "scheduled_retry", + })).rejects.toThrow(); + // A failed sweep leaves the same durable intent available for a later one. + await heartbeatService(db).reapOrphanedRuns(); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.retryRunId))) + .toEqual([expect.objectContaining({ status: "scheduled_retry" })]); + } finally { + await db.execute(sql`ALTER TABLE heartbeat_runs DROP CONSTRAINT reject_retry_cancellation`); + } + const [interrupted] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.retryRunId)); + expect(interrupted).toMatchObject({ status: "scheduled_retry", resultJson: { + startupCancellation: expect.any(Object), cancellation: { reason: "Superseded by a human comment" }, + } }); + + // A fresh service has no process-local cancellation owner or settlement promise. + await heartbeatService(db).reapOrphanedRuns(); + const [cancelled] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.retryRunId)); + expect(cancelled).toMatchObject({ status: "cancelled", error: "Superseded by a human comment" }); + expect(cancelled!.resultJson!.cancellation).toEqual(interrupted!.resultJson!.cancellation); + expect(await db.select().from(issues).where(eq(issues.id, f.issueId))) + .toEqual([expect.objectContaining({ executionRunId: null, assigneeAgentId: f.agentId })]); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.runId, f.retryRunId))) + .toEqual([expect.objectContaining({ status: "cancelled" })]); + const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.retryRunId)); + await heartbeatService(db).reapOrphanedRuns(); + expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.retryRunId))).toEqual(events); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, unrelated.retryRunId))) + .toEqual([expect.objectContaining({ status: "scheduled_retry", resultJson: null })]); + }); + + it.each(["post", "patch"] as const)("keeps a retry's live process and saved work owner when a %s comment arrives", async method => { + const f = await seedIssueWithRetry({ retryStatus: "running" }); + await db.insert(companyMemberships).values({ + companyId: f.companyId, principalType: "user", principalId: "board-user", + membershipRole: "admin", status: "active", + }); + const child = spawn(process.execPath, ["-e", ` + process.on('message', message => { + if (message === 'continue-work') process.send('work-continues'); + }); + process.send('ready'); + `], { stdio: ["ignore", "ignore", "pipe", "ipc"] }); + const exited = once(child, "exit"); + const heartbeat = heartbeatService(db); + try { + await once(child, "message", { signal: AbortSignal.timeout(5_000) }); + runningProcesses.set(f.retryRunId, { child, graceSec: 1, processGroupId: null }); + await db.update(heartbeatRuns).set({ + startedAt: new Date(), runtimeModeResolvedAt: new Date(), processPid: child.pid, + }).where(eq(heartbeatRuns.id, f.retryRunId)); + const app = createApp(boardActor(f.companyId)); + + const res = method === "post" + ? await request(app).post(`/api/issues/${f.issueId}/comments`).send({ body: "Verified source recovery; continue the original work." }) + : await request(app).patch(`/api/issues/${f.issueId}`).send({ comment: "Verified source recovery; continue the original work." }); + + expect(res.status, JSON.stringify(res.body)).toBe(method === "post" ? 201 : 200); + await heartbeat.drainActiveRunExecutions(); + const progress = once(child, "message", { signal: AbortSignal.timeout(5_000) }); + child.send("continue-work"); + expect((await progress)[0]).toBe("work-continues"); + const [savedIssue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const [savedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.retryRunId)); + expect(savedIssue).toMatchObject({ status: "in_progress", executionRunId: f.retryRunId }); + expect(savedRun).toMatchObject({ status: "running", processPid: child.pid }); + expect(await db.select().from(issueComments).where(eq(issueComments.issueId, f.issueId))) + .toHaveLength(1); + expect(await db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "running"), + ))).toHaveLength(1); + } finally { + await heartbeat.drainActiveRunExecutions(); + runningProcesses.delete(f.retryRunId); + if (child.exitCode === null && child.signalCode === null) child.kill(); + await exited; + } + }); + it("surfaces the current scheduled retry in the issue read model", async () => { const { companyId, issueId, agentId, sourceRunId, retryRunId, scheduledRetryAt } = await seedIssueWithRetry(); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 28070336533..22ebe0f98f2 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -760,6 +760,9 @@ export function createPostgresRunDispatchAdapter( const promoteLockedRun = async (tx: Db, run: HeartbeatRun, publications: ActivityPublication[]) => { if ( run.status !== "scheduled_retry" || + // The locked row is also the cancellation claim. If Stop won first, + // leave settlement to that owner rather than promote its parked run. + run.resultJson?.startupCancellation || !run.scheduledRetryAt || new Date(run.scheduledRetryAt).getTime() > now.getTime() ) { diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index d64c054025f..c7139f83572 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -4134,8 +4134,11 @@ export function issueRoutes( if (!scheduledRetryRunId) return null; try { - const cancelled = await heartbeat.cancelRun(scheduledRetryRunId); - const cancelledRunId = cancelled?.id ?? scheduledRetryRunId; + const cancelled = await heartbeat.cancelRun(scheduledRetryRunId, undefined, { + expectedStatus: "scheduled_retry", + }); + if (cancelled?.status !== "cancelled") return null; + const cancelledRunId = cancelled.id; await logActivity(db, { companyId: input.issue.companyId, actorType: input.actor.actorType, @@ -13791,8 +13794,8 @@ export function issueRoutes( }) ? await svc.getCurrentScheduledRetry(existing.id) : null; - const shouldResumeInProgressScheduledRetry = - !!scheduledRetryForHumanComment && + let shouldResumeInProgressScheduledRetry = + scheduledRetryForHumanComment?.status === "scheduled_retry" && scheduledRetryForHumanComment.agentId === requestedAssigneeAgentId; const assigneeSelfCommentOnTerminal = isAssigneeSelfCommentOnTerminalIssue({ @@ -13945,6 +13948,12 @@ export function issueRoutes( issue: existing, actor, }); + if (!cancelledScheduledRetryRunId) { + shouldResumeInProgressScheduledRetry = false; + if (!explicitMoveToTodoRequested && req.body.status === undefined) { + delete updateFields.status; + } + } } if (req.body.executionPolicy !== undefined) { updateFields.executionPolicy = applyActorMonitorScheduledBy( @@ -18265,8 +18274,8 @@ export function issueRoutes( }) ? await svc.getCurrentScheduledRetry(issue.id) : null; - const shouldResumeInProgressScheduledRetry = - !!scheduledRetryForHumanComment && + let shouldResumeInProgressScheduledRetry = + scheduledRetryForHumanComment?.status === "scheduled_retry" && scheduledRetryForHumanComment.agentId === issue.assigneeAgentId; const assigneeSelfCommentOnTerminal = isAssigneeSelfCommentOnTerminalIssue({ @@ -18366,6 +18375,14 @@ export function issueRoutes( let scheduledRetrySupersededByComment = false; let cancelledScheduledRetryRunId: string | null = null; + if (effectiveMoveToTodoRequested && shouldResumeInProgressScheduledRetry) { + cancelledScheduledRetryRunId = await cancelScheduledRetrySupersededByComment({ + scheduledRetryRunId: scheduledRetryForHumanComment?.runId, + issue, + actor, + }); + shouldResumeInProgressScheduledRetry = cancelledScheduledRetryRunId !== null; + } if ( effectiveMoveToTodoRequested && (isClosed || @@ -18375,13 +18392,6 @@ export function issueRoutes( scheduledRetrySupersededByComment = shouldResumeInProgressScheduledRetry && issue.status === "in_progress"; - cancelledScheduledRetryRunId = scheduledRetrySupersededByComment - ? await cancelScheduledRetrySupersededByComment({ - scheduledRetryRunId: scheduledRetryForHumanComment?.runId, - issue, - actor, - }) - : null; const reopenedIssue = await svc.update(id, { status: "todo" }); if (!reopenedIssue) { res.status(404).json({ error: "Issue not found" }); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 5d5bcf6d534..cf6a17c46f4 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -75,6 +75,7 @@ import { } from "../hot-restart.js"; import { serverVersion } from "../../version.js"; import { adapterExecutionControls, waitForAdapterStop } from "../adapter-execution-control.js"; +import { readRunCancellation } from "../run-cancellation.js"; import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js"; import type { environmentService } from "../environments.js"; @@ -82,6 +83,7 @@ import type { environmentRuntimeService, ProviderResourceDisposition } from "../ import type { agentInstructionWorkingCopyService } from "../agent-instruction-working-copies.js"; import type { createHeartbeatRunState } from "./run-state.js"; import type { HeartbeatRetryDependencies, createHeartbeatRetries } from "./retries.js"; +import type { CancelRunOptions } from "./run-control.js"; type HeartbeatRun = typeof heartbeatRuns.$inferSelect; type HeartbeatRetries = ReturnType; @@ -144,7 +146,7 @@ export interface HeartbeatRecoveryDependencies extends Pick Promise; resumeRemoteStopComments: (run: HeartbeatRun) => Promise; dispatchPendingNativeStatusWakeups: () => Promise; - cancelRunInternal: (runId: string, reason?: string, options?: { resultJson?: Record }) => Promise; + cancelRunInternal: (runId: string, reason?: string, options?: Pick) => Promise; startNextQueuedRunForAgent: (agentId: string) => Promise; } @@ -1916,6 +1918,31 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); + // A parked retry has no executor to finish a cancellation after a restart + // or a failed status write. Consume its durable intent through normal run + // control; the guarded CAS and terminal settlement make competing sweeps + // safe, and dispatch cannot promote a fenced retry in the meantime. + const interruptedRetryCancellations = await db.select().from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.status, "scheduled_retry"), + sql`jsonb_typeof(${heartbeatRuns.resultJson}->'startupCancellation') = 'object'`, + )) + .orderBy(asc(heartbeatRuns.updatedAt), asc(heartbeatRuns.id)) + .limit(20); + for (const run of interruptedRetryCancellations) { + const cancellation = readRunCancellation(run.resultJson); + if (!cancellation?.expected) continue; + try { + await cancelRunInternal(run.id, cancellation.reason, { + expectedStatus: "scheduled_retry", + errorCode: readNonEmptyString(parseObject(run.resultJson?.startupCancellation).errorCode) ?? "cancelled", + resultJson: parseObject(run.resultJson), + }); + } catch (error) { + // Leave the durable intent for the next bounded periodic sweep. + logger.warn({ err: error, runId: run.id }, "failed to finish interrupted retry cancellation"); + } + } // Recovery never launches a provider or infers stopped ownership from // terminal status. Uncaptured local copies require durable stop evidence. await instructionCopies.recoverStopped().catch(error => { diff --git a/server/src/services/heartbeat/run-control.test.ts b/server/src/services/heartbeat/run-control.test.ts index c92207fbdba..6f6fce28413 100644 --- a/server/src/services/heartbeat/run-control.test.ts +++ b/server/src/services/heartbeat/run-control.test.ts @@ -20,6 +20,7 @@ import { createHeartbeatRunPreparation } from "./run-preparation.js"; import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "./run-lifecycle.js"; import { createHeartbeatRunControl, type HeartbeatRunControlDependencies } from "./run-control.js"; import { preserveCancelledNativeWorkspaces } from "../native-runtime/native-failed-workspace-recovery.js"; +import { createPostgresRunDispatchAdapter } from "../../modules/run-dispatch/adapters/postgres.js"; vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() })); vi.mock("../adapter-execution-control.js", async (importOriginal) => { @@ -197,6 +198,88 @@ describe.skipIf(!support.supported)("heartbeat run-control database wiring", () return run; } + it.each(["queued", "running", "cancelled", "succeeded"] as const)("does not supersede a retry that becomes %s between lookup and the cancellation fence", async status => { + const { company, agent } = await fixture(); + const [run] = await db.insert(heartbeatRuns).values({ + companyId: company.id, agentId: agent.id, status: "scheduled_retry", + runtimeMode: "legacy", runtimeModeResolvedAt: new Date(), + scheduledRetryReason: "execution_reconciliation_continuation", + }).returning(); + const deps = callbacks(db); + const originalGetRun = deps.getRun.getMockImplementation()!; + deps.getRun.mockImplementationOnce(async id => { + const snapshot = await originalGetRun(id); + await db.update(heartbeatRuns).set({ status, startedAt: new Date() }).where(eq(heartbeatRuns.id, id)); + return snapshot; + }); + + const control = createHeartbeatRunControl(db, deps); + const result = await control.cancelRunInternal(run.id, undefined, { expectedStatus: "scheduled_retry" }); + + expect(result).toBeNull(); + expect(await originalGetRun(run.id)).toMatchObject({ status, resultJson: null }); + expect(deps.envOrchestrator.releaseForRun).not.toHaveBeenCalled(); + expect(deps.options.closeWarmNativeSessionsForRun).not.toHaveBeenCalled(); + expect(deps.appendRunEvent).not.toHaveBeenCalled(); + + // An actual Stop still has authority over promoted, nonterminal work. + if (status === "queued" || status === "running") { + expect(await control.cancelRunInternal(run.id)).toMatchObject({ status: "cancelled" }); + } + }); + + it("can supersede a retry that is still scheduled", async () => { + const { company, agent } = await fixture(); + const [run] = await db.insert(heartbeatRuns).values({ + companyId: company.id, agentId: agent.id, status: "scheduled_retry", + runtimeMode: "legacy", runtimeModeResolvedAt: new Date(), + }).returning(); + const deps = callbacks(db); + + expect(await createHeartbeatRunControl(db, deps).cancelRunInternal( + run.id, undefined, { expectedStatus: "scheduled_retry" }, + )).toMatchObject({ status: "cancelled" }); + }); + + it("blocks promotion after guarded cancellation wins but before it settles", async () => { + const { company, agent } = await fixture(); + await db.update(companies).set({ defaultResponsibleUserId: "operator" }).where(eq(companies.id, company.id)); + const [issue] = await db.insert(issues).values({ + companyId: company.id, title: "Keep the cancellation claim", status: "in_progress", assigneeAgentId: agent.id, + }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ + companyId: company.id, agentId: agent.id, issueId: issue.id, status: "scheduled_retry", + runtimeMode: "legacy", runtimeModeResolvedAt: new Date(), responsibleUserId: "operator", + scheduledRetryAt: new Date(), scheduledRetryReason: "transient_failure", contextSnapshot: { issueId: issue.id }, + }).returning(); + await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, issue.id)); + const deps = callbacks(db); + const settle = deps.setRunStatusFromLive.getMockImplementation()!; + let claimed!: () => void, release!: () => void; + const claimReached = new Promise(resolve => { claimed = resolve; }); + const continueSettlement = new Promise(resolve => { release = resolve; }); + deps.setRunStatusFromLive.mockImplementation(async (...args) => { + claimed(); + await continueSettlement; + return settle(...args); + }); + const cancellation = createHeartbeatRunControl(db, deps).cancelRunInternal( + run.id, undefined, { expectedStatus: "scheduled_retry" }, + ); + try { + await Promise.race([claimReached, cancellation.then(() => { throw new Error("Cancellation did not reach settlement"); })]); + expect((await deps.getRun(run.id))?.resultJson?.startupCancellation).toBeTruthy(); + expect(await createPostgresRunDispatchAdapter(db).promoteOrCancelDueRetry({ + companyId: company.id, runId: run.id, now: new Date(), + })).toMatchObject({ outcome: "not_promoted" }); + expect(await deps.getRun(run.id)).toMatchObject({ status: "scheduled_retry" }); + } finally { + release(); + await cancellation; + } + expect(await deps.getRun(run.id)).toMatchObject({ status: "cancelled" }); + }); + it.each(["ephemeral", "reuse_by_environment"] as const)("keeps %s cancellation eligible for retention when teardown loses the task lock", async leasePolicy => { const { company, agent } = await fixture(); const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Retain cancelled work", diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index 3e4e0ff026f..5f1cb8c67ce 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -233,6 +233,8 @@ function readNonEmptyString(value: unknown): string | null { } export type CancelRunOptions = { + /** Return null if a parked retry was promoted or settled before this cancellation. */ + expectedStatus?: "scheduled_retry"; budgetEnforcement?: BudgetEnforcementScope; /** Optional board request identity, atomically reserved for native Stop. */ cancellationRequestId?: string; @@ -899,6 +901,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont ) { let run = await getRun(runId); if (!run) throw notFound("Heartbeat run not found"); + if (options.expectedStatus && run.status !== options.expectedStatus) return null; if (options.cancellationRequestId) { run = await claimCancellationRequest(db, runId, run.companyId, options.cancellationRequestId, options.cancellationRequestedByUserId ?? null); } @@ -929,7 +932,8 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont return run; const agent = await getAgent(run.agentId); const errorCode = options.errorCode ?? "cancelled"; - const cancellation = requestedRunCancellation(options.resultJson ?? {}, reason); + const cancellation = (options.expectedStatus && run.resultJson?.startupCancellation + ? readRunCancellation(run.resultJson) : null) ?? requestedRunCancellation(options.resultJson ?? {}, reason); const pendingProcessCancellation = processRunCancellationSettlements.get( run.id, @@ -952,7 +956,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont // Established legacy processes must still be stopped if the database is // unavailable. Only native or not-yet-dispatched preparation needs this // additional durable fence before its existing cancellation path. - if (options.budgetEnforcement || (!options.cancellationRequestId && (run.runtimeMode === "native" || (!run.runtimeModeResolvedAt && !running && !control)))) { + if (options.expectedStatus || options.budgetEnforcement || (!options.cancellationRequestId && (run.runtimeMode === "native" || (!run.runtimeModeResolvedAt && !running && !control)))) { const fence = async (tx: Db) => { const [fenced] = await tx.update(heartbeatRuns).set({ // Record handoff intent before native cancellation can finalize and release @@ -963,13 +967,14 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont ${JSON.stringify({ cancellation })}::jsonb || jsonb_build_object('startupCancellation', jsonb_build_object( 'requestedAt', ${new Date().toISOString()}::text, + 'errorCode', ${errorCode}::text, 'beforeNativeSelection', ${heartbeatRuns.runtimeMode} = 'legacy' and ${heartbeatRuns.runtimeModeResolvedAt} is null and ${heartbeatRuns.executionStage} = 'preparing' and coalesce(${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->>'adapterType' = 'paperclip_runner', false) ))`, }).where(and(eq(heartbeatRuns.id, runId), inArray(heartbeatRuns.status, - pendingNativeRetry ? [...CANCELLABLE_HEARTBEAT_RUN_STATUSES, "failed"] : [...CANCELLABLE_HEARTBEAT_RUN_STATUSES], + options.expectedStatus ? [options.expectedStatus] : pendingNativeRetry ? [...CANCELLABLE_HEARTBEAT_RUN_STATUSES, "failed"] : [...CANCELLABLE_HEARTBEAT_RUN_STATUSES], ))).returning(); return fenced ?? null; }; @@ -984,7 +989,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont } if (!fenced) { stopOwnership?.release(); - return getRun(runId); + return options.expectedStatus ? null : getRun(runId); } run = fenced; }