From 6364fa7327530d0e54ca3970568ce5d6c41dff86 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 4 Nov 2026 11:16:54 -0600 Subject: [PATCH] fix(heartbeat): wait for retry predecessor cleanup (#18989) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip manages agent work and preserves its task ownership. > - Heartbeat admission controls when a queued run can start. > - A failed run can still own workspace saving and environment cleanup. > - Automatic recovery can transfer the task execution lock to its queued retry before that cleanup finishes. > - The old queue check then sees the retry itself and misses the predecessor's remaining work. > - This pull request checks the exact retry predecessor before claiming the queued run. > - Existing cleanup notifications can then resume that same run through normal admission. ## Linked Issues or Issue Description Refs #18898 and #18691. The first repair covers canonical connected-task retries with a native source. The second covers live remote workspace writers. This change covers ordinary automatic continuation when the predecessor failed before native runtime selection. The search found no open repair for that path. **What happened?** A preparation failure saved a terminal legacy run row while its remote workspace save and environment cleanup were still active. Generic continuation created one queued successor and moved the task execution lock to it. The queue checked that current lock but did not check the successor's predecessor. The new run started and failed at the existing unfinished-writer guard. **Expected behavior** Keep that same successor queued until its exact predecessor finishes execution and cleanup. Check company, agent and task identity before using retry provenance as a wait condition. Preserve Stop, reassignment, failed cleanup and independent recovery gates. **Steps to reproduce** 1. Save a failed legacy preparation run with an active environment lease. 2. Queue its ordinary automatic continuation and bind the task execution lock to the successor. 3. Claim the queued run before the predecessor finishes cleanup. 4. The old code starts the successor. The fixed code retains it in the queue. 5. Commit successful cleanup through the existing cleanup owner. The ordinary dispatcher starts that same run once. **Paperclip version or commit** The causal regression fails on master `86814c6550f52e99ede150a883bdc773cbb540e3`. **Deployment mode** Source checkout on macOS arm64 with disposable PostgreSQL, the assembled heartbeat service and the production work coordinator. The adapter boundary is controlled. No hosted provider replay or deployment is claimed. ## What Changed - Check the current task execution owner and the retry's exact predecessor under the existing task claim lock. - Use the saved native task binding, ordinary task binding and historical context fallback in that order. Ignore unrelated retry provenance. - Reuse the existing live-executor and pending-cleanup checks. Reuse committed cleanup notifications and the existing dispatcher. - Add real PostgreSQL claim controls and assembled cleanup-event acceptance. Keep the original source failure, saved files and current task ownership intact. - Document the queued predecessor-cleanup rule. ## Verification - RED: the unchanged queue claimed the successor and wrote a new controller lease while its legacy predecessor retained an active remote environment lease. The regression expected the successor to remain queued. - GREEN: the same regression passes after the change. The complete queue suite passes all 71 tests. - The assembled heartbeat suite passes all six tests. The added cases cover one continuation after committed cleanup, saved-file reuse, final task lock release, board Stop, reassignment and failed cleanup. - Scope controls cover a deleted source after scan, another company, another task, another agent, canonical native task binding and the existing current-owner hold. A failed cleanup and a still-live local executor retain the wait. - Node 24.19.0 and pnpm 9.15.4. Module boundaries and `git diff --check` pass. - Initial fixture attempts failed on database constraints and observer timing. The final fixtures use valid immutable bindings, the board Stop attribution and a completed real dispatcher call after reading the exact committed lease state. Production ownership guards were retained throughout. - Full low-memory repository typecheck passed at `5cddb49322`. The sync to master `04a24f4e06` adds only the independent retained-allocation test fixture from #18984; production code is unchanged. - Full `pnpm build` passes on current head `c7c2bcf08ffd05be5ce714683ae1dcfe6f5b49f1`. - The post-sync run executes all 80 behavior assertions successfully across the queue, assembled cleanup and updated upstream retention suites. Its queue `afterAll` then exceeds the existing 30-second database shutdown hook. That failed suite result is retained. The controlled post-build queue rerun passes all 71 tests with normal teardown. The earlier shutdown cause remains undetermined. Its owned queue-test PostgreSQL process has exited. The separate local default `pnpm test:run` was intentionally stopped after full unfiltered CI passed on the exact head. Four local chat partitions completed before that stop; this is not a local full-suite pass. The verified owned process group and its PostgreSQL fixture have exited. Logs and retained fixture data remain available. The timeout has not been increased. - Full unfiltered CI passes on current head `c7c2bcf08ffd05be5ce714683ae1dcfe6f5b49f1`: all 12 server shards, all six internal chat partitions, all nine serialized server shards, the workspace groups, native Runner lanes, typecheck, build and browser E2E. Both required [ci / verify](https://github.com/paperclipai/paperclip/actions/runs/41866478617/job/126711606613) and [ci / e2e](https://github.com/paperclipai/paperclip/actions/runs/41866478617/job/126711263146) pass. The file selector is skipped, so this is the full CI test matrix. The head has 55 successful checks and three expected skipped checks. The ready-for-review contributor trust and security scan also pass. - Current-head Greptile is 5/5, its check is successful, and no inline review threads remain. The ordinary review check also passes. No merge, deployment or original task retry was performed. ## Risks - A predecessor with unfinished cleanup keeps its successor queued. Its existing cleanup owner remains responsible for settlement. - A task claim reads at most one additional predecessor row. A matching source also uses the existing ownership and cleanup queries. The change adds no timer, schema, credential, provider or model policy. - A committed cleanup signal only requests normal admission. Current ownership and independent execution gates still apply. ## Model Used OpenAI GPT-6 through Codex, with reasoning, code editing, tool use and PostgreSQL verification. The session does not expose a more specific serving model ID or 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 (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass — focused suites pass; the separate local full-suite attempt is explicitly incomplete above - [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 | 7 ++ ...heartbeat-native-cleanup-admission.test.ts | 98 ++++++++++++++++++- server/src/services/heartbeat/queue.test.ts | 90 ++++++++++++++++- server/src/services/heartbeat/queue.ts | 12 ++- 4 files changed, 202 insertions(+), 5 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 74422770352..0e93aefda20 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -370,6 +370,13 @@ The active-lock lifecycle is part of the checkout contract: Stale-lock recovery is crash recovery, not a retry loop. Paperclip must not clear or adopt locks held by non-terminal runs. After stale cleanup, a checkout `409` should mean a real live owner, status/assignee mismatch, unresolved blocker, or active gate still prevents checkout. Agents must treat that `409` as an ownership conflict and stop rather than retrying the same checkout. +A queued retry waits for its same-company, same-task and same-agent predecessor +to finish execution and cleanup, including when the task execution lock already +names the retry. This also applies to failures before native runtime selection. +Committed cleanup notifications reconsider that same queued run through ordinary +admission. Failed cleanup, Stop, reassignment and independent recovery gates +retain their existing effect; a terminal run row alone cannot release the wait. + Stopping a queued legacy run before runtime selection clears only its own task locks. It preserves deferred messages without promoting another run. The stop must have a recorded, expected operator cancellation and no runtime selection, diff --git a/server/src/__tests__/heartbeat-native-cleanup-admission.test.ts b/server/src/__tests__/heartbeat-native-cleanup-admission.test.ts index b5153c32d7d..dd82048651f 100644 --- a/server/src/__tests__/heartbeat-native-cleanup-admission.test.ts +++ b/server/src/__tests__/heartbeat-native-cleanup-admission.test.ts @@ -1,8 +1,10 @@ import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; import { randomUUID } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import path from "node:path"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, authUsers, companies, companyMemberships, createDb, environmentLeases, environments, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { agents, agentWakeupRequests, authUsers, companies, companyMemberships, createDb, environmentLeases, environments, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { heartbeatService } from "../services/heartbeat.js"; import { NativeStartupCleanupPendingError, readNativeStartupCleanup } from "../services/native-runtime/native-startup-cleanup.js"; @@ -10,6 +12,9 @@ import { prepareNativeHeartbeatRun } from "../services/native-runtime/prepare-na import { environmentService } from "../services/environments.js"; import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; import { createIsolatedTestHome } from "./helpers/isolated-test-home.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { writeEnvironmentCleanup } from "../services/environment-cleanup-notifications.js"; +import { registerRunDispatchWork } from "../services/heartbeat/dispatch-work.js"; const execute = vi.hoisted(() => vi.fn()); vi.mock("../adapters/index.js", async () => { @@ -76,6 +81,97 @@ suite("native terminal-run cleanup admission", () => { expect(execute).toHaveBeenCalledTimes(3); }, 30_000); + it.each(["continue", "stop", "reassign", "failed cleanup"] as const)( + "settles an automatic legacy retry from the committed cleanup event: %s", async mode => { + const home = createIsolatedTestHome(), previousHome = process.env.PAPERCLIP_HOME; + process.env.PAPERCLIP_HOME = home.home; + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), sourceId = randomUUID(); + const successorId = randomUUID(), wakeId = randomUUID(), ownerId = randomUUID(); + const cwd = path.join(home.home, "saved-task"); + await mkdir(cwd, { recursive: true }); + await writeFile(path.join(cwd, "saved.txt"), "Preserved work before preparation failed.\n"); + await db.insert(authUsers).values({ id: ownerId, name: "Task owner", email: `${ownerId}@example.test`, + createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companies).values({ id: companyId, name: "Automatic cleanup retry", issuePrefix: `R${companyId.slice(0, 6)}`, defaultResponsibleUserId: ownerId }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: ownerId, status: "active", membershipRole: "owner" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Continuation worker", status: "idle", adapterType: "process", + adapterConfig: { cwd }, runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true, maxConcurrentRuns: 1 } } }); + await db.insert(issues).values({ id: issueId, companyId, title: "Continue preserved work", status: "in_progress", + assigneeAgentId: agentId, responsibleUserId: ownerId }); + await db.insert(heartbeatRuns).values({ id: sourceId, companyId, agentId, issueId, status: "failed", runtimeMode: "legacy", + startedAt: new Date(Date.now() - 10_000), finishedAt: new Date(), executionStage: "preparing", errorCode: "adapter_failed", + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, contextSnapshot: { issueId } }); + const [lease] = await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: sourceId, + provider: "daytona", providerLeaseId: randomUUID(), status: "active" }).returning(); + await db.insert(agentWakeupRequests).values({ id: wakeId, companyId, agentId, source: "automation", triggerDetail: "system", + reason: "issue_continuation_needed", requestedByActorType: "system", status: "queued", runId: successorId, + payload: { issueId } }); + await db.insert(heartbeatRuns).values({ id: successorId, companyId, agentId, issueId, status: "queued", invocationSource: "automation", + responsibleUserId: ownerId, retryOfRunId: sourceId, wakeupRequestId: wakeId, + contextSnapshot: { issueId, retryOfRunId: sourceId, source: "issue.continuation_recovery", + wakeReason: "issue_continuation_needed", skipIssueComment: true } }); + await db.update(issues).set({ executionRunId: successorId, executionAgentNameKey: "continuation-worker", executionLockedAt: new Date() }) + .where(eq(issues.id, issueId)); + execute.mockReset(); + execute.mockImplementation(async (context: { runId: string }) => { + expect(context.runId).toBe(successorId); + expect(await readFile(path.join(cwd, "saved.txt"), "utf8")).toBe("Preserved work before preparation failed.\n"); + await writeFile(path.join(cwd, "continued.txt"), "Continued once after cleanup.\n"); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId)); + return { exitCode: 0, signal: null, timedOut: false, summary: "Continued preserved work after cleanup." }; + }); + const heartbeat = heartbeatService(db), errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + let completedCleanupDispatches = 0; + const cleanupStatus = mode === "failed cleanup" ? "failed" : "success"; + try { + await registerRunDispatchWork(coordinator, { promoteDueScheduledRetries: async () => undefined, + nextScheduledRetryAt: async () => null, schedulingSuppressed: async () => false, + resumeQueuedRunsAfterDrain: async () => { + const [current] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)); + await heartbeat.resumeQueuedRunsAfterDrain({ companyId }); + if (current?.status === "released" && current.releasedAt && current.cleanupStatus === cleanupStatus) + completedCleanupDispatches++; + }, + }, () => true).ready; + expect((await heartbeat.getRun(successorId))?.status).toBe("queued"); + expect(execute).not.toHaveBeenCalled(); + if (mode === "stop") await heartbeat.cancelRun(successorId, "Cancelled by a board operator", { + resultJson: { cancelledByActorType: "user", cancelledByUserId: ownerId }, + }); + if (mode === "reassign") { + const [replacement] = await db.insert(agents).values({ companyId, name: "New owner", adapterType: "process" }).returning(); + await db.update(issues).set({ assigneeAgentId: replacement.id }).where(eq(issues.id, issueId)); + } + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "released", releasedAt: new Date(), + cleanupStatus }).where(eq(environmentLeases.id, lease.id)), + { companyId, runId: sourceId }); + await vi.waitFor(() => expect(completedCleanupDispatches).toBeGreaterThan(0), { timeout: 10_000 }); + if (mode === "continue") { + await heartbeat.drainActiveRunExecutions(); + expect((await heartbeat.getRun(successorId))?.status).toBe("succeeded"); + expect(execute).toHaveBeenCalledTimes(1); + expect(await readFile(path.join(cwd, "continued.txt"), "utf8")).toBe("Continued once after cleanup.\n"); + const [done] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(done).toMatchObject({ status: "done", executionRunId: null, checkoutRunId: null }); + } else { + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 }); + expect((await heartbeat.getRun(successorId))?.status).toBe(mode === "failed cleanup" ? "queued" : "cancelled"); + expect(execute).not.toHaveBeenCalled(); + } + expect((await heartbeat.getRun(sourceId))?.status).toBe("failed"); + expect(await readFile(path.join(cwd, "saved.txt"), "utf8")).toBe("Preserved work before preparation failed.\n"); + expect(errors).toEqual([]); + } finally { + await coordinator.stop(); + await heartbeat.drainActiveRunExecutions(); + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, agentId)); + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome; + home.cleanup(); + } + }, 30_000, + ); + it("retries a failed startup intent through timer failure and drain before verified cancellation", async () => { const home = createIsolatedTestHome(), originalHome = process.env.PAPERCLIP_HOME; process.env.PAPERCLIP_HOME = home.home; diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index cf6e2f7bffb..9e87edb1c00 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -1,6 +1,6 @@ import { claimNativeSessionResumptions } from "../native-runtime/native-finalization-reconciler.js"; import type { Db } from "@paperclipai/db"; -import { authUsers, agents, agentWakeupRequests, caseHandoffs, cases, chatCompletionDeliveries, companies, companyMemberships, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issueComments, issueThreadInteractions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { authUsers, agents, agentWakeupRequests, caseHandoffs, cases, chatCompletionDeliveries, companies, companyMemberships, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environmentLeases, heartbeatRuns, issueComments, issueThreadInteractions, issues, nativeRunFinalizations } from "@paperclipai/db"; import { eq, inArray, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; @@ -259,6 +259,94 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { return { company, agent, issue, wake, run }; } + async function pendingPredecessorFixture() { + const f = await fixture("queued"); + const [source] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, + issueId: f.issue.id, status: "failed", runtimeMode: "legacy", finishedAt: new Date(), + executionStage: "preparing", errorCode: "adapter_failed", + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + contextSnapshot: { issueId: f.issue.id } }).returning(); + const [lease] = await db.insert(environmentLeases).values({ companyId: f.company.id, issueId: f.issue.id, + heartbeatRunId: source.id, provider: "daytona", providerLeaseId: crypto.randomUUID(), status: "active" }).returning(); + const [successor] = await db.update(heartbeatRuns).set({ retryOfRunId: source.id, responsibleUserId: "queue-test-user", + contextSnapshot: { issueId: f.issue.id, source: "issue.continuation_recovery", retryOfRunId: source.id, + wakeReason: "issue_continuation_needed" } }).where(eq(heartbeatRuns.id, f.run.id)).returning(); + const deps = callbacks(db); + deps.getAgent.mockResolvedValue(f.agent); + return { ...f, source, lease, successor, deps, queue: createHeartbeatQueue(db, deps) }; + } + + it("keeps an automatic continuation queued while its legacy predecessor still owns cleanup", async () => { + const { successor, lease, queue, deps, ...f } = await pendingPredecessorFixture(); + expect(await queue.claimQueuedRun(successor)).toBeNull(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, successor.id)))[0].status).toBe("queued"); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0].executionRunId).toBe(successor.id); + expect(deps.executeRun).not.toHaveBeenCalled(); + + await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success" }) + .where(eq(environmentLeases.id, lease.id)); + expect(await queue.claimQueuedRun(successor)).toMatchObject({ id: successor.id, status: "running" }); + expect(await queue.claimQueuedRun(successor)).toBeNull(); + }); + + it("keeps a retry queued after failed cleanup and until its local executor releases", async () => { + const f = await pendingPredecessorFixture(); + await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "failed" }) + .where(eq(environmentLeases.id, f.lease.id)); + expect(await f.queue.claimQueuedRun(f.successor)).toBeNull(); + await db.update(environmentLeases).set({ cleanupStatus: "success" }).where(eq(environmentLeases.id, f.lease.id)); + f.deps.liveRunExecutions.has.mockImplementation(id => id === f.source.id); + expect(await f.queue.claimQueuedRun(f.successor)).toBeNull(); + f.deps.liveRunExecutions.has.mockReturnValue(false); + expect(await f.queue.claimQueuedRun(f.successor)).toMatchObject({ id: f.successor.id, status: "running" }); + }); + + it.each(["missing", "foreign company", "other task", "other agent", "native task overrides context"] as const)( + "does not hold a queued run for a %s retry source", async mismatch => { + const f = await pendingPredecessorFixture(); + let sourceId = f.source.id; + if (mismatch === "missing") { + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, sourceId)); + expect(await f.queue.claimQueuedRun(f.successor)).toMatchObject({ id: f.successor.id, status: "running" }); + return; + } else if (mismatch === "foreign company") { + const [company] = await db.insert(companies).values({ name: "Unrelated company", issuePrefix: "OTHER" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Other company worker", adapterType: "process" }).returning(); + const [source] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, + status: "failed", finishedAt: new Date() }).returning(); + sourceId = source.id; + await db.insert(environmentLeases).values({ companyId: company.id, heartbeatRunId: sourceId, status: "active", provider: "daytona" }); + } else if (mismatch === "other agent") { + const [other] = await db.insert(agents).values({ companyId: f.company.id, name: "Another worker", adapterType: "process" }).returning(); + const [source] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: other.id, issueId: f.issue.id, + status: "failed", finishedAt: new Date() }).returning(); + sourceId = source.id; + await db.insert(environmentLeases).values({ companyId: f.company.id, heartbeatRunId: sourceId, status: "active", provider: "daytona" }); + } else { + const [other] = await db.insert(issues).values({ companyId: f.company.id, title: "Another task" }).returning(); + const [source] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, issueId: other.id, + ...(mismatch === "native task overrides context" ? { nativeIssueId: other.id } : {}), + contextSnapshot: { issueId: f.issue.id }, status: "failed", finishedAt: new Date() }).returning(); + sourceId = source.id; + await db.insert(environmentLeases).values({ companyId: f.company.id, heartbeatRunId: sourceId, status: "active", provider: "daytona" }); + } + const [successor] = await db.update(heartbeatRuns).set({ retryOfRunId: sourceId }) + .where(eq(heartbeatRuns.id, f.successor.id)).returning(); + expect(await f.queue.claimQueuedRun(successor)).toMatchObject({ id: successor.id, status: "running" }); + }, + ); + + it("retains the current execution owner's hold when retry provenance is unrelated", async () => { + const f = await pendingPredecessorFixture(); + const [unrelated] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, + status: "succeeded", finishedAt: new Date() }).returning(); + const [successor] = await db.update(heartbeatRuns).set({ retryOfRunId: unrelated.id }) + .where(eq(heartbeatRuns.id, f.successor.id)).returning(); + await db.update(issues).set({ executionRunId: f.source.id }).where(eq(issues.id, f.issue.id)); + expect(await f.queue.claimQueuedRun(successor)).toBeNull(); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0].executionRunId).toBe(f.source.id); + }); + it.each((["process", "paperclip_runner"] as const).flatMap(adapterType => (["coalesced", "claimed-primary", "legacy-envelope", "delete-old", "delete-new", "stopped", "running", "system-primary", "foreign-primary", "same-named-agent", "mixed-author"] as const).map(scenario => ({ adapterType, scenario }))))("protects a queued comment batch across $adapterType/$scenario", async ({ adapterType, scenario }) => { const f = await fixture("queued"); let admissionAgent: typeof agents.$inferSelect = { ...f.agent, adapterType, adapterConfig: adapterType === "paperclip_runner" ? { provider: "codex" } : {} }; diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index aa0977e5659..dff68ca5b46 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1079,14 +1079,20 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (ownsIssue && issueId && await taskChatRetrySourceCleanupPending(tx, run, issueId)) { return { ownsIssue, blocked: true }; } - const previousRunId = run.scheduledRetryReason === "native_provider_overloaded" + const executionOwnerId = run.scheduledRetryReason === "native_provider_overloaded" ? run.retryOfRunId : ownsIssue ? owner?.executionRunId : null; - if (previousRunId && previousRunId !== run.id) { - const [previous] = await tx.select({ status: heartbeatRuns.status }) + for (const previousRunId of new Set([executionOwnerId, ownsIssue ? run.retryOfRunId : null])) { + if (!previousRunId || previousRunId === run.id) continue; + const [previous] = await tx.select({ status: heartbeatRuns.status, agentId: heartbeatRuns.agentId, + issueId: heartbeatRuns.issueId, nativeIssueId: heartbeatRuns.nativeIssueId, + contextIssueId: sql`coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId')`, + }) .from(heartbeatRuns).where(and( eq(heartbeatRuns.id, previousRunId), eq(heartbeatRuns.companyId, run.companyId), )); + if (previousRunId !== executionOwnerId && (previous?.agentId !== run.agentId || + (previous.nativeIssueId ?? previous.issueId ?? previous.contextIssueId) !== issueId)) continue; // A terminal result can precede workspace/lease cleanup on this or // another controller. Local absence alone is not a release receipt. if (!isHeartbeatRunTerminalStatus(previous?.status) ||