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; }