From 4808a3a2b1d3a447bdda5fac83f043fb2fabfb32 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 28 Oct 2026 11:46:58 -0500 Subject: [PATCH] fix(heartbeat): retry operator Stop after a transaction deadlock (#17895) https://entire.com/gh/paperclipai/-/trails/69/changes/paperclip/69 ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Operator Stop must revoke task authority before it stops provider execution. > - Its transaction locks the task and then the run. A concurrent writer can hold the run while it waits for the task. > - PostgreSQL can abort the Stop transaction with a confirmed deadlock. The cancel API then returns 500. > - This pull request retries only that aborted database transaction, at most three times. > - Provider Stop and activity publication stay after commit, so the retry cannot repeat those effects. ## Linked Issues or Issue Description Refs #17639. Its agent-chat project browser test completed the product assertions, then failed during cleanup when the cancel API returned 500. A real PostgreSQL lock-cycle test reproduces the same `40P01` error in the Stop transaction. Related: #13911 addresses audit foreign-key lock contention. This change handles a different concurrent run-writer path without changing the schema or task/run lock order. Stack base: #17639. This is a separate correctness repair for the sleep migration stack, not another polling migration. ## What Changed - Control the runner timeout regression fixture's clock so it reaches the pending append before expiry; retain production cancellation and late-append assertions. - Retry the Stop authority transaction only after PostgreSQL reports `40P01`. Make at most three attempts, with short bounded backoff. - Discard rolled-back local run state, claim flags, and queued activity publications before each retry. Re-read current authority under fresh locks. - Keep provider cancellation and activity publication outside the retried transaction. Preserve other errors and the existing native lock-conflict response. - Add a real two-transaction deadlock test. Add server-side failure tests for retry exhaustion and errors that must not retry. - Let run-control tests use a validated disposable PostgreSQL database URL. ## Verification Published head: `60ba892ae9fc5faaf8cd0833694b4a66c13323c3`, targeting master after #17639 merged at `1ed0af41867ba40fd1d23292d395588d18c2fcbe`. The normal ancestry merge preserves the complete five-path tested cancellation patch byte-for-byte, including the review repair, and all four unrelated current master paths. The production retry logic remains unchanged from this slice's reviewed implementation; the follow-up adds coverage and its audit. - **228 distinct complete cases pass:** all 48 run-control cases and all 180 native-session runtime cases. The two new targeted cases are included in 48 and are not counted again. Server typechecking, build and module-boundary checks also pass. - The new real PostgreSQL cases reach a failed native run's correlated Stop claim, cancellation fence, board handoff retirement and activity write before raising `40P01`. One case preserves retry authority and commits exactly one cancellation activity/live notice. The other changes coordinator authority between attempts, returns 409, preserves the failed run and handoff, and performs no provider/workspace cleanup. - A negative control removes only the publication-list reset. It fails with two live notices for one durable activity. Production was restored byte-for-byte before final validation. The existing actual two-connection deadlock, bounded attempts, non-deadlock rejection and post-commit provider behavior remain covered. - Preserve the initial new-assertion typecheck failure (optional live-event payload) and the corrected final pass. Service-local test files are typechecked; `src/__tests__` is excluded. The negative-control harness initially expected an untruncated Vitest assertion line; the actual named 1-versus-2 failure and restoration were inspected before correcting its result matcher. - [Current-parent acceptance and failure ledger](https://github.com/paperclipai/paperclip/blob/60ba892ae9fc5faaf8cd0833694b4a66c13323c3/doc/plans/2026-10-28-cancellation-deadlock-acceptance.md) records current scope, commands, isolated fixtures, mutation evidence and limits. Earlier combined-stack counts retain their original revision scope. The existing controlled runner timeout/late-append fixture is retained and passes within the complete 180-case runtime suite. - This addresses the prior Ponytail request to exercise native claimed state before deadlock rollback and re-evaluate authority afterward. The advisory loop-invariant snapshot/named-attempt-bound note remains a maintainability suggestion, not a runtime defect. No production loop refactor or new worker duty is introduced. - **All 58 current checks are clean; Greptile and Ponytail both rate this exact head 5/5.** Ponytail explicitly confirms that the native-claim coverage request is resolved and says Ship. The formerly queued dependency/quality review workflow has completed successfully. Old `fffc`/parent checks and reviews are not current passes; the former pending Snyk status is not called successful. Actual failing Ponytail findings must be fixed. Missing/stopped/paused turns without a verdict are authorized nonblockers. Active unfinished reviews retain their original per-head timer, with a 25-minute monitor cutoff respecting the 30-minute maximum. - No full local repository matrix, live provider execution, external alarm, physical host/database suspension or savings claim is made. No schema, dependency, workflow, shared scheduler contract or shared OS resource changes are introduced. ## Risks - Persistent contention still returns an error after the third attempt. This change does not promise to eliminate every possible lock cycle. - Retrying a write after an ambiguous connection failure would be unsafe. The retry accepts only PostgreSQL's confirmed deadlock abort code; it does not accept transport errors. - No schema or public API contract changes. Current task/run authority and cancellation eligibility are checked again on each attempt. ## Model Used OpenAI GPT-6 via Codex, with shell, GitHub CLI, TypeScript, PostgreSQL, and Playwright tools. The exact serving model ID and context window are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [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 --- ...-10-28-cancellation-deadlock-acceptance.md | 77 ++++++++++ .../src/native-session-runtime.test.ts | 60 ++++---- .../__tests__/run-control-fixtures.ts | 11 +- .../services/heartbeat/run-control.test.ts | 133 +++++++++++++++++- server/src/services/heartbeat/run-control.ts | 24 +++- 5 files changed, 271 insertions(+), 34 deletions(-) create mode 100644 doc/plans/2026-10-28-cancellation-deadlock-acceptance.md diff --git a/doc/plans/2026-10-28-cancellation-deadlock-acceptance.md b/doc/plans/2026-10-28-cancellation-deadlock-acceptance.md new file mode 100644 index 00000000000..491240ace26 --- /dev/null +++ b/doc/plans/2026-10-28-cancellation-deadlock-acceptance.md @@ -0,0 +1,77 @@ +# Cancellation deadlock retry acceptance + +This existing correctness slice (#17895) retries only the operator Stop database +transaction after PostgreSQL reports `40P01`, at most three times. Provider Stop, +workspace settlement, and activity publication remain after commit. This is not +a new polling or scheduling duty. + +## Current parent and outstanding review + +The previous published head is +`fffc8d0c46fbbeb3da4a23a36e638fefd3b8fe3d`, originally based on native progress +`08d07ead105dc89c2af9b62f4d90bff8cc93540c`. The slice is prepared by a normal +merge of verified native progress `6f365285f5b233a89acc714eadfead5ab145100e`. +All four original changed-file hunks retain their exact content and context. +Only the run-control blob index hashes differ because the parent also changed +that file. The automatic composition is clean and preserves those parent edits. + +The prior Ponytail review said Ship with an actionable coverage request: existing +deadlock tests did not reach a correlated native failed run's claimed state before +rollback. The follow-up adds two real PostgreSQL cases in the existing run-control +suite, with no new production behavior: + +1. Keep retry authority: claim the correlated Stop, fence the failed native run, + cancel its real board terminal handoff, and persist cancellation activity inside + the first transaction. Raise PostgreSQL `40P01` before commit. Retry succeeds; + exactly one cancellation activity is durable and exactly one live notice is + published. The correlation belongs to the committed attempt. +2. Revoke retry authority: after that aborted transaction rolls back, an independent + coordinator update retires retry eligibility. The second attempt returns 409, + preserves the failed run and uncancelled handoff, publishes no cancellation + activity, and performs no provider or workspace cleanup. + +The existing actual two-connection lock-cycle regression and the bounded retry/ +non-deadlock cases remain in the complete suite. The new tests exercise claims +and activity produced before a real PostgreSQL transaction abort; they do not +infer correctness from a mocked retry callback alone. + +## Evidence and failure accounting + +- Both new cases pass against the retained production reset logic. +- Removing only `publications.length = 0` makes the preserved-authority case fail: + two live cancellation notices appear for one committed activity. Production was + restored byte-for-byte to its integrated index before further validation. + `/tmp/sleep-aux-native-retry-mutation-proof.json` records that comparison. +- The negative-control harness initially expected an untruncated assertion line; + Vitest abbreviates that line. Its actual named failure and 1-versus-2 result were + inspected, and the matcher was corrected without rerunning or altering production. +- Typechecking caught an optional live-event payload in the new assertion. The + assertion now uses optional access. Preserve the original failed typecheck at + `/tmp/sleep-aux-current-parent-types.log`; the final typecheck passes at + `/tmp/sleep-aux-current-parent-types-final.log`. +- **228 distinct complete cases pass:** all 48 run-control cases, including the two + additions, and all 180 native-session runtime cases, including the existing + controlled timeout/late-append fixture. The two targeted probes are included in + 48 and are not counted again. +- Server typechecking, build, and module-boundary checks pass. `server/tsconfig.json` + excludes `src/__tests__`, but includes service-local test files such as + `run-control.test.ts`; typechecking does not replace test execution. Earlier + statements about tests omitted by production typechecking must be read with + that path boundary. + +The three PostgreSQL probes use separate new disposable databases on the owned +local cluster. Full-run results and exit receipts are in +`/tmp/sleep-aux-current-parent-{types-final,build,boundaries,control,runner}.log`. +The targeted positive and mutation logs are +`/tmp/sleep-aux-native-retry-regression.log` and +`/tmp/sleep-aux-native-retry-mutant-red.log`. + +## Remaining gates and limits + +This is local preparation. The dependency must merge, the normal merge onto its +actual landed master must preserve this tested patch and all master changes, and +fresh exact-head CI plus Greptile 5/5 must pass before protected merge. Prior +review scores and checks on older heads are not current passes. The previous +Snyk status was pending and is not called successful here. No shared OS resource +cleanup, schema change, whole-repository local matrix, live provider execution, +external alarm, physical suspension, or savings qualification is claimed. diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 588b64556df..caba0b66d12 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -3499,10 +3499,6 @@ describe("executeNativeSession recovery", () => { }); it("stops and closes a timed-out consumer even when the caller requested a warm session", async () => { - let markAppendStarted = () => {}; - const appendStarted = new Promise((resolve) => { - markAppendStarted = resolve; - }); let releaseAppend = () => {}; const appendReleased = new Promise((resolve) => { releaseAppend = resolve; @@ -3515,7 +3511,6 @@ describe("executeNativeSession recovery", () => { let appendCommitted = false; const appendEvent = vi.fn( async (_event: PrpEvent, options?: { signal: AbortSignal }) => { - markAppendStarted(); await Promise.race([ appendReleased, new Promise((_resolve, reject) => { @@ -3612,28 +3607,39 @@ describe("executeNativeSession recovery", () => { async completeRun() {}, }; - const execution = executeNativeSession({ - input, - backend, - controlPlane: port, - runnerInstanceId: "runner-recovery", - controlPlaneInstanceId: "control-recovery", - timeoutMs: 1, - keepSessionOpen: true, - }); - const rejection = expect(execution).rejects.toThrow( - "native session timed out", - ); - await appendStarted; - await vi.waitFor(() => expect(iteratorTeardown).toHaveBeenCalledOnce()); - await rejection; - expect(cancel).toHaveBeenCalledOnce(); - expect(close).toHaveBeenCalled(); - expect(appendEvent).toHaveBeenCalledOnce(); - expect(appendCommitted).toBe(false); - releaseAppend(); - await new Promise((resolve) => setImmediate(resolve)); - expect(appendCommitted).toBe(false); + // Reach the pending append before expiring the consumer. A real 1 ms + // deadline can expire during setup before the consumer reaches this case. + vi.useFakeTimers(); + try { + const execution = executeNativeSession({ + input, + backend, + controlPlane: port, + runnerInstanceId: "runner-recovery", + controlPlaneInstanceId: "control-recovery", + timeoutMs: 1, + keepSessionOpen: true, + }); + const rejection = expect(execution).rejects.toThrow( + "native session timed out", + ); + await vi.advanceTimersByTimeAsync(0); + expect(appendEvent).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + await rejection; + expect(iteratorTeardown).toHaveBeenCalledOnce(); + expect(cancel).toHaveBeenCalledOnce(); + expect(close).toHaveBeenCalled(); + expect(appendEvent).toHaveBeenCalledOnce(); + expect(appendCommitted).toBe(false); + releaseAppend(); + await vi.advanceTimersByTimeAsync(0); + expect(appendCommitted).toBe(false); + } finally { + releaseAppend(); + releaseTeardown(); + vi.useRealTimers(); + } }); it("closes a failed session while retaining an uncancellable event read", async () => { diff --git a/server/src/services/heartbeat/__tests__/run-control-fixtures.ts b/server/src/services/heartbeat/__tests__/run-control-fixtures.ts index 2adce9077c1..06ed71c35d1 100644 --- a/server/src/services/heartbeat/__tests__/run-control-fixtures.ts +++ b/server/src/services/heartbeat/__tests__/run-control-fixtures.ts @@ -1,8 +1,9 @@ import type { Db } from "@paperclipai/db"; -import { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns } from "@paperclipai/db"; +import { agents, companies, createDb, applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns } from "@paperclipai/db"; import { sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, vi, type Mock } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js"; import { runningProcesses } from "../../../adapters/index.js"; import { terminateLocalService } from "../../local-service-supervisor.js"; import { createHeartbeatRunState } from "../run-state.js"; @@ -65,15 +66,17 @@ export function callbacks(db: Db): TestCallbacks { failedProcessRunCancellations: new Map; failed: boolean; error?: unknown }>(), } satisfies HeartbeatRunControlDependencies; } -export const runControlPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_RUN_CONTROL_TEST_DATABASE_URL"); +export const runControlPostgresSupport = externalTestDatabaseUrl ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); if (!runControlPostgresSupport.supported) console.warn(`Skipping run-control database tests: ${runControlPostgresSupport.reason}`); export function setupRunControlDatabase(name: string, ready: (db: Db) => void) { let database: Awaited>; let db: Db; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase(name); - db = createDb(database.connectionString); + if (externalTestDatabaseUrl) await applyPendingMigrations(externalTestDatabaseUrl); + else database = await startEmbeddedPostgresTestDatabase(name); + db = createDb(externalTestDatabaseUrl ?? database!.connectionString); ready(db); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => { diff --git a/server/src/services/heartbeat/run-control.test.ts b/server/src/services/heartbeat/run-control.test.ts index ab0a9fbb5a8..c73dba4d120 100644 --- a/server/src/services/heartbeat/run-control.test.ts +++ b/server/src/services/heartbeat/run-control.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import type { Db } from "@paperclipai/db"; import { activityLog, agents, agentWakeupRequests, caseEvents, caseHandoffs, cases, companies, environmentLeases, heartbeatRuns, issueComments, issues, nativeRunFinalizations } from "@paperclipai/db"; -import { eq, inArray } from "drizzle-orm"; +import { eq, inArray, sql } from "drizzle-orm"; import { afterEach, describe, expect, it, vi } from "vitest"; import { runningProcesses } from "../../adapters/index.js"; import { @@ -14,6 +14,7 @@ import { AdapterStopTimeoutError } from "../adapter-stop-timeout.js"; import { HttpError } from "../../errors.js"; import { captureException } from "../../sentry.js"; import { terminateLocalService } from "../local-service-supervisor.js"; +import { publishLiveEvent } from "../live-events.js"; import { createHeartbeatRunState } from "./run-state.js"; import { createHeartbeatRunControl } from "./run-control.js"; import { preserveCancelledNativeWorkspaces } from "../native-runtime/native-failed-workspace-recovery.js"; @@ -173,6 +174,79 @@ describe.skipIf(!runControlPostgresSupport.supported)("heartbeat run-control dat )).toMatchObject({ status: "cancelled" }); }); + it.each([["40P01", 3], ["XX000", 1]] as const)("bounds cancellation transaction attempts for PostgreSQL %s", async (code, attempts) => { + const { company, agent } = await fixture(); + const run = await liveRun(company.id, agent.id); + const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Cancellation failure", status: "in_progress", + assigneeAgentId: agent.id, executionRunId: run.id }).returning(); + await db.update(heartbeatRuns).set({ issueId: issue.id, contextSnapshot: { issueId: issue.id } }).where(eq(heartbeatRuns.id, run.id)); + await db.execute(sql`create sequence run_control_cancel_attempts`); + await db.execute(sql.raw(`create function run_control_cancel_fault() returns trigger language plpgsql as $$ + begin + if new.result_json->'executionCancellation'->>'state' = 'requested' then + perform nextval('run_control_cancel_attempts'); + raise exception using errcode = '${code}', message = 'Synthetic cancellation transaction failure'; + end if; + return new; + end $$`)); + await db.execute(sql`create trigger run_control_cancel_fault before update on heartbeat_runs + for each row execute function run_control_cancel_fault()`); + const deps = callbacks(db); + try { + await expect(createHeartbeatRunControl(db, deps).cancelRunInternal(run.id, "Operator Stop", { + resultJson: { cancelledByActorType: "user", cancelledByUserId: "operator" }, + })).rejects.toMatchObject({ cause: { code } }); + const [counter] = await db.execute<{ count: number }>(sql`select last_value::int as count from run_control_cancel_attempts`); + expect(counter.count).toBe(attempts); + expect(await deps.getRun(run.id)).toMatchObject({ status: "running", resultJson: null }); + expect(terminateLocalService).not.toHaveBeenCalled(); + expect(deps.appendRunEvent).not.toHaveBeenCalled(); + } finally { + await db.execute(sql`drop trigger run_control_cancel_fault on heartbeat_runs`); + await db.execute(sql`drop function run_control_cancel_fault()`); + await db.execute(sql`drop sequence run_control_cancel_attempts`); + } + }); + + it("finishes operator Stop after a concurrent run writer causes a transaction deadlock", async () => { + const { company, agent } = await fixture(); + const run = await liveRun(company.id, agent.id); + const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Stop contention", status: "in_progress", + assigneeAgentId: agent.id, executionRunId: run.id, checkoutRunId: run.id }).returning(); + await db.update(heartbeatRuns).set({ issueId: issue.id, contextSnapshot: { issueId: issue.id } }).where(eq(heartbeatRuns.id, run.id)); + let enter!: () => void, release!: () => void; + const entered = new Promise(resolve => { enter = resolve; }); + const released = new Promise(resolve => { release = resolve; }); + let blockerPid = 0; + const writer = db.transaction(async tx => { + // Make Stop's default detector see the cycle first, so this reproduces + // the API-side victim observed in CI rather than a random victim. + await tx.execute(sql`set local deadlock_timeout = '10s'`); + const [backend] = await tx.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`); + blockerPid = backend.pid; + await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)).for("update"); + enter(); await released; + await tx.update(issues).set({ updatedAt: new Date() }).where(eq(issues.id, issue.id)); + }); + await entered; + const deps = callbacks(db); + const stopping = createHeartbeatRunControl(db, deps).cancelRunInternal(run.id, "Operator Stop", { + resultJson: { cancelledByActorType: "user", cancelledByUserId: "operator" }, + }).then(value => ({ value, error: undefined }), error => ({ value: undefined, error })); + try { + await vi.waitFor(async () => { + const waiting = await db.execute(sql`select pid from pg_stat_activity + where ${blockerPid} = any(pg_blocking_pids(pid)) and wait_event_type = 'Lock'`); + expect(waiting.length).toBeGreaterThan(0); + }, { timeout: 5000 }); + } finally { release(); } + await writer; + const result = await stopping; + expect(result.error).toBeUndefined(); + expect(result.value).toMatchObject({ status: "cancelled" }); + expect(await deps.getRun(run.id)).toMatchObject({ status: "cancelled" }); + }); + it.each(["queued", "running"] as const)("preserves case intent when a guarded Stop loses to %s promotion", async status => { const { company, agent } = await fixture(); const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Promoted case work", status: "todo", assigneeAgentId: agent.id }).returning(); @@ -393,6 +467,63 @@ describe.skipIf(!runControlPostgresSupport.supported)("heartbeat run-control dat .toMatchObject({ status: "in_progress", phase: "verification", completedAt: null }); }); + it.each([false, true])("discards rolled-back native Stop claims and activity before retry (authority revoked=%s)", async revokeAuthority => { + const f = await boardTerminalFixture("running", "native"); + const requestId = randomUUID(), deps = callbacks(db); + vi.mocked(publishLiveEvent).mockClear(); + const transact = db.transaction.bind(db); + let injected = false; + const transaction = vi.spyOn(db, "transaction").mockImplementationOnce(async (operation, config) => { + try { + return await transact(async tx => { + const result = await operation(tx); + const [claimed] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)); + expect(claimed.resultJson?.startupCancellation).toMatchObject({ cancellationRequestId: requestId }); + expect(claimed.resultJson?.executionCancellation).toMatchObject({ state: "requested" }); + expect((await tx.select().from(caseHandoffs).where(eq(caseHandoffs.id, f.receipt.handoffId)))[0].state).toBe("cancelled"); + expect((await tx.select().from(activityLog).where(eq(activityLog.companyId, f.company.id))) + .filter(row => row.action === "case.handoff_cancelled")).toHaveLength(1); + expect(deps.getAgent).not.toHaveBeenCalled(); + injected = true; + await tx.execute(sql`do $$ begin raise exception using errcode = '40P01', message = 'Deadlock after native Stop claim'; end $$`); + return result; + }, config); + } catch (error) { + if (injected && revokeAuthority) await db.update(nativeRunFinalizations) + .set({ phase: "terminal_failure", failureCode: "retry_authority_revoked", nextAttemptAt: null }) + .where(eq(nativeRunFinalizations.runId, f.run.id)); + throw error; + } + }); + try { + const stopping = createHeartbeatRunControl(db, deps).cancelRunInternal(f.run.id, "Retry native Stop", { + resultJson: { cancelledByActorType: "user", cancelledByUserId: "operator" }, + cancellationRequestId: requestId, cancellationRequestedByUserId: "operator", + }); + if (revokeAuthority) { + await expect(stopping).rejects.toMatchObject({ status: 409 }); + expect(await deps.getRun(f.run.id)).toMatchObject({ status: "failed", resultJson: null }); + expect(deps.getAgent).not.toHaveBeenCalled(); + expect(deps.options.closeWarmNativeSessionsForRun).not.toHaveBeenCalled(); + expect(deps.envOrchestrator.releaseForRun).not.toHaveBeenCalled(); + } else { + await expect(stopping).resolves.toMatchObject({ status: "cancelled" }); + expect((await deps.getRun(f.run.id))?.resultJson?.startupCancellation).toMatchObject({ cancellationRequestId: requestId }); + } + expect(injected).toBe(true); + expect(transaction.mock.calls.length).toBeGreaterThanOrEqual(2); + expect((await db.select().from(caseHandoffs).where(eq(caseHandoffs.id, f.receipt.handoffId)))[0].state) + .toBe(revokeAuthority ? "awaiting_completion" : "cancelled"); + const expected = revokeAuthority ? 0 : 1; + expect((await db.select().from(activityLog).where(eq(activityLog.companyId, f.company.id))) + .filter(row => row.action === "case.handoff_cancelled")).toHaveLength(expected); + expect(vi.mocked(publishLiveEvent).mock.calls.map(([event]) => event) + .filter(event => event.companyId === f.company.id && event.type === "activity.logged" + && event.payload?.action === "case.handoff_cancelled")).toHaveLength(expected); + expect(terminateLocalService).not.toHaveBeenCalled(); + } finally { transaction.mockRestore(); } + }); + it("stops current preparing execution despite a prior terminal checkout when cancelling board intent", async () => { // Queue admission binds the new execution before preparation adopts its // predecessor's checkout. The execution owner is authoritative in this gap. diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index c1b50a46517..c32cebc2013 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -1134,7 +1134,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont let caseHandoffRetryClaimed = false; if (handoffIssueId && supersedesCaseHandoff && options.resultJson?.caseHandoffYieldStop !== true) { const publications: ActivityPublication[] = []; - const supersessionApplied = await db.transaction(async tx => { + const applySupersession = () => db.transaction(async tx => { const [issue] = await tx.select({ id: issues.id, executionRunId: issues.executionRunId, checkoutRunId: issues.checkoutRunId, assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId }).from(issues) .where(and(eq(issues.companyId, run!.companyId), eq(issues.id, handoffIssueId))).for("update"); @@ -1189,7 +1189,27 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont await cancelCaseHandoffsForIssue(tx as unknown as Db, { companyId: run!.companyId, issueId: handoffIssueId, sourceRunId: runId, includeUnboundTerminalIntent: Boolean(fenced && ownsTask), reason, publications }); return true; - }).catch(rethrowNativeCancellationLockConflict); + }); + let supersessionApplied = false; + for (let attempt = 0; ; attempt++) { + const previousRun = run; + try { + supersessionApplied = await applySupersession(); + break; + } catch (error) { + const value = parseObject(error); + const deadlock = value.code === "40P01" || parseObject(value.cause).code === "40P01"; + if (!deadlock || attempt >= 2) rethrowNativeCancellationLockConflict(error); + // PostgreSQL aborted the entire transaction. Discard its local + // receipts before re-reading current authority under fresh locks. + // Provider Stop and publication happen only after this boundary. + run = previousRun; + cancellationClaimed = false; + caseHandoffRetryClaimed = false; + publications.length = 0; + await new Promise(resolve => setTimeout(resolve, 25 * (attempt + 1))); + } + } publications.forEach(publishActivity); if (!supersessionApplied) return null; }