fix(heartbeat): clear orphan execution locks on every issue when a run finalizes (#4318)
> **Note (rebase, 2026-06-11):** this PR was rebased onto current `master` again after #6008 (`Clear stale checkoutRunId on run finalization and add backstop sweeper`) landed. See "What Changed" below for how the previous narrow per-issue checkoutRunId clear from #6008 is now subsumed by a single bulk-update pass over every sibling that still references the finalizing run, with the two columns cleared in separate, scoped UPDATEs so retry pointers are not clobbered. ## Thinking Path > - Paperclip orchestrates AI-agent companies; issue execution ownership is gated by `executionRunId` / `executionAgentNameKey` / `executionLockedAt`, and any checkout whose run doesn't match the stored `executionRunId` is rejected with 409 "Issue run ownership conflict" > - In production, a running company silently got stuck: multiple in-progress issues ended up with `executionRunId` pointing at heartbeat runs that had already finalized hours earlier, so every new agent checkout returned 409 and the issues stayed marked blocked forever > - Root cause: `releaseIssueExecutionAndPromote` only resolved and cleared the execution lock on one issue per finalizing run (the run's `contextSnapshot.issueId`, or `rows[0]` when no context issue existed), but `enqueueWakeup`'s "legacy run" fallback can stamp the same `run.id` onto sibling issues' `execution_run_id`, so the siblings were left as orphans > - #4258 shipped a *reactive* fix for this bug class in `issueService`: `clearExecutionRunIfTerminal` now self-heals a stale execution lock on the next ownership-gated access (`checkout`, `assertCheckoutOwner`, `release`) to each affected issue, and `release` now unconditionally clears the three execution-lock fields > - #6008 shipped the *symmetric* fix for the `checkoutRunId` column (per-issue self-heal in `releaseIssueExecutionAndPromote`, `clearCheckoutRunIfTerminal` helper, and a backstop sweeper) > - This PR adds the *proactive* half at the point of run finalization, and generalizes #6008's per-issue checkoutRunId clear to every sibling that still references the finalizing run. After this PR + #4258 + #6008, orphan locks are cleared at the moment the run ends (across both execution and checkout columns, on every affected sibling), not only on the next access attempt to one of them ## Linked Issues or Issue Description - Closes #4194 - Closes #201 - Closes #3904 ## What Changed - **`server/src/services/heartbeat.ts` — `releaseIssueExecutionAndPromote`:** lock the context issue (when set) **and** every issue still referencing the finalizing run via either `execution_run_id` or `checkout_run_id`, under a single `SELECT ... FOR UPDATE ORDER BY id` (deterministic lock acquisition across concurrent finalizations). Then issue two scoped bulk `UPDATE`s in the same transaction: - one clears `executionRunId` / `executionAgentNameKey` / `executionLockedAt` on every issue whose `executionRunId` still matches this run, - the other clears `checkoutRunId` on every issue whose `checkoutRunId` still matches this run. The split avoids clobbering a retry's `executionRunId` pointer: in the codex-transient-upstream and process-loss retry paths, `executionRunId` is moved from this run to the retry run before `releaseIssueExecutionAndPromote` runs, while `checkoutRunId` is left pinned at the failed run. A single combined `UPDATE` with an `OR` predicate would null the retry's `executionRunId` in that case — these two scoped UPDATEs do not. The deferred-wake promotion contract is preserved: pick the run's context issue when present, else the first candidate (matching the legacy `rows[0]` selection under the new ordering). Recovery-agent fields added by a concurrent master change (`taskKey`, `recoveryAgent`, `recoverySessionBefore`, `recoveryAgentNameKey`, and the extra `assigneeAgentId`/`assigneeUserId` columns used downstream for `issueNeedsImmediateRecovery`) are fully preserved through the merge. The workspace-validation-failed recovery-comment path added by master is also preserved on the primary issue. - **`server/src/__tests__/execution-lock-orphan-cleanup.test.ts` (new, 6 tests):** multi-issue cleanup on finalize (2 issues); higher fan-out (4 issues) exercising the bulk `UPDATE` path; finalization of a run without a `contextSnapshot.issueId`; cross-company isolation under a pathologically cross-tenant `executionRunId`; unrelated-run locks are never touched by a sibling run's finalization; and a dedicated test for the `checkoutRunId` bulk-clear path that proves the split-UPDATE invariant by seeding a sibling whose `executionRunId` already points at a retry run while `checkoutRunId` is still pinned at the finalizing run — the test asserts the retry pointer is preserved and the checkout column is cleared. **Not in this PR:** `server/src/services/issues.ts` is intentionally unchanged. The release-side changes from the previous revision of this PR are fully subsumed by #4258 (`clearExecutionRunIfTerminal` plus unconditional clear in `release`) and #6008 (`clearCheckoutRunIfTerminal`). The per-issue checkoutRunId clear added in `releaseIssueExecutionAndPromote` by #6008 is replaced by the bulk path here, which strictly widens coverage from "primary issue only" to "every sibling that still references this run". ## Verification - `pnpm install --frozen-lockfile` — clean - `pnpm typecheck` (server workspace) — passes on the rebased branch - Focused suite (8 files, 147 tests — `execution-lock-orphan-cleanup` (6), `heartbeat-run-log`, `heartbeat-run-summary`, `issues-checkout-wakeup`, `issue-execution-policy-routes`, `issue-agent-mutation-ownership-routes`, `issues-service`, and `issue-stale-execution-lock-routes`): **147/147 pass**. - `heartbeat-process-recovery.test.ts` (52 tests): 51 pass; the one failure (`queues exactly one retry when the recorded local pid is dead`) reproduces verbatim on raw `master` with the PR's changes reverted, so it is a pre-existing flake (also noted by the earlier CI-retrigger commit on this branch). - Regression evidence: reverting `server/src/services/heartbeat.ts` to `master` while keeping the 6 new tests causes 5 of them to fail (the finalization-cleanup tests, including the new checkoutRunId-pointer-preservation test; the "unrelated-run locks never touched" test passes either way — that's its purpose as a negative control); restoring the fix returns to 6/6 green. - `pnpm-lock.yaml` untouched; no migration required; no public API shape change. Repro-ability of the original production symptom: ``` # Seed an issue with executionRunId pointing at a finalized run # (matches what enqueueWakeup's legacy-run fallback can produce) UPDATE issues SET execution_run_id = '<finalized-run-id>', execution_agent_name_key = 'ceo', execution_locked_at = NOW() WHERE id = '<issue-id>'; # Any subsequent svc.checkout against this issue 409s until the # lock is cleared. Before #4258, the lock stayed forever. After # #4258, it self-heals on next ownership-gated access. After this # PR, it's cleared at the moment the run finalizes so an untouched # sibling issue doesn't rely on a later access to recover. ``` ## Risks - **Same bug class exists in two adjacent, untouched code paths in this file** — `enqueueProcessLossRetry` repoints `executionRunId` only for the context issue (not siblings stamped with the failed run id), and `enqueueMissingIssueCommentRetry` locks all matching issues under `FOR UPDATE` but only updates the first row returned. Filed separately as #4319 with exact line refs so this PR can stay narrow. - **No `activity_log` entry is emitted for secondary orphan issues** whose locks are cleared by the new bulk UPDATEs — only the single primary issue retains its existing promotion event stream. Callers that audit lock transitions purely via `activity_log` may see orphan issues flip to `execution_run_id = null` (or `checkout_run_id = null`) without a matching event. Easy to add a batched log line in a follow-up if audit completeness matters; the #6008 backstop sweeper already emits `issue.stale_lock_cleared` for the catch-up path so this is mainly an observability nicety for the proactive path. - **UI polling shift** — `ui/src/pages/IssueDetail.tsx` and `ui/src/lib/issueActiveRun.ts` key off `execution_run_id` for active-run polling; clearing orphan locks at finalize (rather than waiting for next access as in #4258's flow) means the "executing" UI state falls back slightly faster when the underlying run has genuinely finalized. Observable, but an improvement over showing stuck state. - **Lockfile and manifests untouched**; no migration required; no public API shape change. ## Model Used - **Provider/model:** Anthropic Claude Opus 4.7 (the model this session is running on, per Cursor IDE system context) - **Harness:** Cursor IDE - **Capabilities used:** extended/reasoning thinking mode; filesystem and shell tool use; parallel subagent orchestration for a three-lens readonly self-review (correctness+concurrency, regression blast radius, test adequacy+style) before the initial push; targeted rebase conflict resolution after #4258 landed and again after #6008 landed, combining all three branches' changes to `releaseIssueExecutionAndPromote` - **Context:** full repository plus live access to the running Paperclip instance that exhibited the bug (the instance was unblocked via a targeted DB intervention before this code fix was authored; the live observation drove the root-cause analysis) ## 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 (the roadmap contains no references to heartbeat execution-lock management or `releaseIssueExecutionAndPromote`) - [x] I have searched GitHub for duplicate or related PRs and linked them above (#4258 and #6008 are the closest prior art and are explicitly cross-linked in the thinking path and "What Changed" sections; no other open or merged PR touches `releaseIssueExecutionAndPromote`) - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template — see "Linked Issues or Issue Description" above - [x] I have run tests locally and they pass (typecheck workspace-wide; 147/147 in the focused suite including #4258's and #6008's new tests) - [x] I have added or updated tests where applicable (6 regression tests; 5 of them provably fail on `master` without the code change) - [ ] If this change affects the UI, I have included before/after screenshots — *n/a, this is a server-only change; any UI polling effect is documented under Risks* - [ ] I have updated relevant documentation to reflect my changes — *n/a, no user-facing or API docs reference `releaseIssueExecutionAndPromote`* - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green (the prior `verify` flake is unrelated to this PR's change path and reproduces on unrelated branches; documented above and in follow-up #4328) - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups (the only test-coverage gap Greptile flagged on the latest review — the `checkoutRunId` bulk-clear branch — is now covered by the new 6th test in this revision) - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Devin Foley <devin@paperclip.ing> Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
AyeletMorris-ShieldFC
committed by
GitHub
GitHub
parent
6f9801a46b
commit
d2ef767712
@@ -0,0 +1,461 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
agents,
|
||||
agentWakeupRequests,
|
||||
companies,
|
||||
createDb,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
issues,
|
||||
} from "@paperclipai/db";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
import { heartbeatService } from "../services/heartbeat.ts";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
if (!embeddedPostgresSupport.supported) {
|
||||
console.warn(
|
||||
`Skipping execution-lock orphan-cleanup tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
|
||||
);
|
||||
}
|
||||
|
||||
describeEmbeddedPostgres("execution lock orphan cleanup", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
|
||||
beforeAll(async () => {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-exec-lock-orphan-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
}, 30_000);
|
||||
|
||||
afterEach(async () => {
|
||||
await db.delete(agentWakeupRequests);
|
||||
await db.delete(activityLog);
|
||||
await db.delete(heartbeatRunEvents);
|
||||
await db.delete(issues);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(agents);
|
||||
await db.delete(companies);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await tempDb?.cleanup();
|
||||
});
|
||||
|
||||
async function seedCompany() {
|
||||
const companyId = randomUUID();
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
return companyId;
|
||||
}
|
||||
|
||||
async function seedAgent(companyId: string, name = "CEO") {
|
||||
const agentId = randomUUID();
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name,
|
||||
role: "engineer",
|
||||
status: "running",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
return agentId;
|
||||
}
|
||||
|
||||
async function seedIssue(companyId: string, overrides: Partial<typeof issues.$inferInsert> = {}) {
|
||||
const issueId = randomUUID();
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Fixture issue",
|
||||
status: "in_progress",
|
||||
priority: "medium",
|
||||
...overrides,
|
||||
});
|
||||
return issueId;
|
||||
}
|
||||
|
||||
describe("heartbeat run finalization", () => {
|
||||
it("clears execution_run_id on every issue that references the finalized run, not just the run's contextSnapshot issue", async () => {
|
||||
// Regression test for the "stale execution lock" bug:
|
||||
//
|
||||
// A single heartbeat run can end up stamped onto multiple issues' execution_run_id:
|
||||
// - the issue it explicitly checked out (run.contextSnapshot.issueId)
|
||||
// - any *other* issue whose enqueueWakeup hit the "legacy run" fallback
|
||||
// and reattached this same run to that issue's execution_run_id.
|
||||
//
|
||||
// The original releaseIssueExecutionAndPromote implementation only resolved the
|
||||
// single context issue (or rows[0] when no context issue existed), so all the
|
||||
// other issues were left with execution_run_id pointing at a finalized run —
|
||||
// an orphan lock that blocked any future agent from checking them out.
|
||||
const companyId = await seedCompany();
|
||||
const ceoAgentId = await seedAgent(companyId, "CEO");
|
||||
|
||||
const contextIssueId = await seedIssue(companyId);
|
||||
const orphanIssueId = await seedIssue(companyId);
|
||||
|
||||
const runId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId: ceoAgentId,
|
||||
invocationSource: "assignment",
|
||||
status: "queued",
|
||||
contextSnapshot: { issueId: contextIssueId },
|
||||
});
|
||||
|
||||
// Both issues reference the same run — exactly the state produced when the
|
||||
// CEO explicitly checks out one issue and enqueueWakeup's legacy-run fallback
|
||||
// stamps the same run.id onto a sibling issue.
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: runId,
|
||||
executionAgentNameKey: "ceo",
|
||||
executionLockedAt: new Date(),
|
||||
})
|
||||
.where(eq(issues.id, contextIssueId));
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: runId,
|
||||
executionAgentNameKey: "ceo",
|
||||
executionLockedAt: new Date(),
|
||||
})
|
||||
.where(eq(issues.id, orphanIssueId));
|
||||
|
||||
await heartbeatService(db).cancelRun(runId);
|
||||
|
||||
const [contextAfter] = await db.select().from(issues).where(eq(issues.id, contextIssueId));
|
||||
const [orphanAfter] = await db.select().from(issues).where(eq(issues.id, orphanIssueId));
|
||||
|
||||
expect(contextAfter?.executionRunId).toBeNull();
|
||||
expect(contextAfter?.executionAgentNameKey).toBeNull();
|
||||
expect(contextAfter?.executionLockedAt).toBeNull();
|
||||
|
||||
expect(orphanAfter?.executionRunId).toBeNull();
|
||||
expect(orphanAfter?.executionAgentNameKey).toBeNull();
|
||||
expect(orphanAfter?.executionLockedAt).toBeNull();
|
||||
});
|
||||
|
||||
it("clears the execution lock on every orphan when three or more issues share the finalized run", async () => {
|
||||
// The production symptom of this bug surfaced with four issues pointing at
|
||||
// the same run id; two is the minimum to reproduce but higher fan-out is
|
||||
// what we actually observed in the field. The bulk UPDATE path should be
|
||||
// O(n) without per-row round-trips, so exercise n>2 explicitly.
|
||||
const companyId = await seedCompany();
|
||||
const ceoAgentId = await seedAgent(companyId, "CEO");
|
||||
|
||||
const contextIssueId = await seedIssue(companyId);
|
||||
const orphanAId = await seedIssue(companyId);
|
||||
const orphanBId = await seedIssue(companyId);
|
||||
const orphanCId = await seedIssue(companyId);
|
||||
|
||||
const runId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId: ceoAgentId,
|
||||
invocationSource: "assignment",
|
||||
status: "queued",
|
||||
contextSnapshot: { issueId: contextIssueId },
|
||||
});
|
||||
|
||||
for (const issueId of [contextIssueId, orphanAId, orphanBId, orphanCId]) {
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: runId,
|
||||
executionAgentNameKey: "ceo",
|
||||
executionLockedAt: new Date(),
|
||||
})
|
||||
.where(eq(issues.id, issueId));
|
||||
}
|
||||
|
||||
await heartbeatService(db).cancelRun(runId);
|
||||
|
||||
const rows = await db.select().from(issues).where(eq(issues.companyId, companyId));
|
||||
expect(rows).toHaveLength(4);
|
||||
for (const row of rows) {
|
||||
expect(row.executionRunId).toBeNull();
|
||||
expect(row.executionAgentNameKey).toBeNull();
|
||||
expect(row.executionLockedAt).toBeNull();
|
||||
}
|
||||
});
|
||||
|
||||
it("clears orphan locks even when the finalizing run has no contextSnapshot issueId", async () => {
|
||||
// Not every heartbeat run is tied to a specific issue via contextSnapshot
|
||||
// (e.g. assignment-less wakeups). releaseIssueExecutionAndPromote must
|
||||
// still clear every issue whose execution_run_id matches the run in that
|
||||
// case — the legacy behavior picked rows[0] as the "primary" issue for
|
||||
// deferred-wake promotion, which we intentionally preserve, but cleanup
|
||||
// must span all matches.
|
||||
const companyId = await seedCompany();
|
||||
const ceoAgentId = await seedAgent(companyId, "CEO");
|
||||
|
||||
const orphanAId = await seedIssue(companyId);
|
||||
const orphanBId = await seedIssue(companyId);
|
||||
|
||||
const runId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId: ceoAgentId,
|
||||
invocationSource: "assignment",
|
||||
status: "queued",
|
||||
contextSnapshot: {},
|
||||
});
|
||||
|
||||
for (const issueId of [orphanAId, orphanBId]) {
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: runId,
|
||||
executionAgentNameKey: "ceo",
|
||||
executionLockedAt: new Date(),
|
||||
})
|
||||
.where(eq(issues.id, issueId));
|
||||
}
|
||||
|
||||
await heartbeatService(db).cancelRun(runId);
|
||||
|
||||
const [orphanAAfter] = await db.select().from(issues).where(eq(issues.id, orphanAId));
|
||||
const [orphanBAfter] = await db.select().from(issues).where(eq(issues.id, orphanBId));
|
||||
|
||||
expect(orphanAAfter?.executionRunId).toBeNull();
|
||||
expect(orphanBAfter?.executionRunId).toBeNull();
|
||||
});
|
||||
|
||||
it("never clears execution locks on issues in another company", async () => {
|
||||
// Defense-in-depth regression: a finalizing run in company A must never
|
||||
// affect rows in company B even if (pathologically) an issue in B carries
|
||||
// the same execution_run_id. The `company_id = run.companyId` scope in
|
||||
// the cleanup query is the only line protecting tenant isolation here.
|
||||
const companyAId = await seedCompany();
|
||||
const companyBId = await seedCompany();
|
||||
const agentAId = await seedAgent(companyAId, "CEO-A");
|
||||
|
||||
const issueAId = await seedIssue(companyAId);
|
||||
const issueBId = await seedIssue(companyBId);
|
||||
|
||||
const runAId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runAId,
|
||||
companyId: companyAId,
|
||||
agentId: agentAId,
|
||||
invocationSource: "assignment",
|
||||
status: "queued",
|
||||
contextSnapshot: { issueId: issueAId },
|
||||
});
|
||||
|
||||
// Company B's issue pathologically references company A's run id. The FK
|
||||
// from issues.execution_run_id → heartbeat_runs.id permits this because
|
||||
// it doesn't enforce a company-match constraint; in production this
|
||||
// state is only reachable via a bug, but the scoping predicate should
|
||||
// make us robust against it.
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: runAId,
|
||||
executionAgentNameKey: "ceo-a",
|
||||
executionLockedAt: new Date(),
|
||||
})
|
||||
.where(eq(issues.id, issueAId));
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: runAId,
|
||||
executionAgentNameKey: "ceo-b",
|
||||
executionLockedAt: new Date(),
|
||||
})
|
||||
.where(eq(issues.id, issueBId));
|
||||
|
||||
await heartbeatService(db).cancelRun(runAId);
|
||||
|
||||
const [issueAAfter] = await db.select().from(issues).where(eq(issues.id, issueAId));
|
||||
const [issueBAfter] = await db.select().from(issues).where(eq(issues.id, issueBId));
|
||||
|
||||
expect(issueAAfter?.executionRunId).toBeNull();
|
||||
expect(issueAAfter?.companyId).toBe(companyAId);
|
||||
|
||||
expect(issueBAfter?.executionRunId).toBe(runAId);
|
||||
expect(issueBAfter?.executionAgentNameKey).toBe("ceo-b");
|
||||
expect(issueBAfter?.executionLockedAt).not.toBeNull();
|
||||
expect(issueBAfter?.companyId).toBe(companyBId);
|
||||
});
|
||||
|
||||
it("clears checkout_run_id on every sibling and preserves an in-flight retry's execution_run_id pointer", async () => {
|
||||
// Exercises the second of the two scoped bulk UPDATEs in
|
||||
// releaseIssueExecutionAndPromote, and the invariant that motivated
|
||||
// splitting it from the first one:
|
||||
//
|
||||
// - One sibling has BOTH execution_run_id and checkout_run_id pinned at
|
||||
// the finalizing run (normal "checked out and executing" shape) — both
|
||||
// columns must be cleared.
|
||||
//
|
||||
// - Another sibling is in the codex-transient / process-loss retry shape:
|
||||
// execution_run_id has already been moved to a *different* retry run,
|
||||
// while checkout_run_id is left pinned at the original failed run.
|
||||
// The checkout column must be cleared but the retry's execution_run_id
|
||||
// pointer must NOT be clobbered — otherwise the retry can never check
|
||||
// itself out and the bug re-surfaces under a different name.
|
||||
//
|
||||
// - A third sibling has checkout_run_id pinned at the finalizing run with
|
||||
// execution_run_id null — covers the post-status-change shape where the
|
||||
// issue's checkout pointer outlived its execution lock.
|
||||
const companyId = await seedCompany();
|
||||
const ceoAgentId = await seedAgent(companyId, "CEO");
|
||||
|
||||
const dualLockIssueId = await seedIssue(companyId);
|
||||
const retryIssueId = await seedIssue(companyId);
|
||||
const checkoutOnlyIssueId = await seedIssue(companyId);
|
||||
|
||||
const finalizingRunId = randomUUID();
|
||||
const retryRunId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values([
|
||||
{
|
||||
id: finalizingRunId,
|
||||
companyId,
|
||||
agentId: ceoAgentId,
|
||||
invocationSource: "assignment",
|
||||
status: "queued",
|
||||
contextSnapshot: { issueId: dualLockIssueId },
|
||||
},
|
||||
{
|
||||
// Marked `running` rather than `queued` so cancelRun(finalizingRunId)
|
||||
// does not also reconcile this queued retry into a cancelled state
|
||||
// (which would then release its own execution_run_id and defeat the
|
||||
// point of this test). In production, retries are typically already
|
||||
// running by the time the original failing run finalizes.
|
||||
id: retryRunId,
|
||||
companyId,
|
||||
agentId: ceoAgentId,
|
||||
invocationSource: "assignment",
|
||||
status: "running",
|
||||
contextSnapshot: { issueId: retryIssueId },
|
||||
},
|
||||
]);
|
||||
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: finalizingRunId,
|
||||
executionAgentNameKey: "ceo",
|
||||
executionLockedAt: new Date(),
|
||||
checkoutRunId: finalizingRunId,
|
||||
})
|
||||
.where(eq(issues.id, dualLockIssueId));
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: retryRunId,
|
||||
executionAgentNameKey: "ceo",
|
||||
executionLockedAt: new Date(),
|
||||
checkoutRunId: finalizingRunId,
|
||||
})
|
||||
.where(eq(issues.id, retryIssueId));
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
checkoutRunId: finalizingRunId,
|
||||
})
|
||||
.where(eq(issues.id, checkoutOnlyIssueId));
|
||||
|
||||
await heartbeatService(db).cancelRun(finalizingRunId);
|
||||
|
||||
const [dualAfter] = await db.select().from(issues).where(eq(issues.id, dualLockIssueId));
|
||||
const [retryAfter] = await db.select().from(issues).where(eq(issues.id, retryIssueId));
|
||||
const [checkoutOnlyAfter] = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(eq(issues.id, checkoutOnlyIssueId));
|
||||
|
||||
// Dual-lock sibling: both columns cleared.
|
||||
expect(dualAfter?.executionRunId).toBeNull();
|
||||
expect(dualAfter?.executionAgentNameKey).toBeNull();
|
||||
expect(dualAfter?.executionLockedAt).toBeNull();
|
||||
expect(dualAfter?.checkoutRunId).toBeNull();
|
||||
|
||||
// Retry sibling: checkout column cleared, retry's execution pointer preserved.
|
||||
expect(retryAfter?.checkoutRunId).toBeNull();
|
||||
expect(retryAfter?.executionRunId).toBe(retryRunId);
|
||||
expect(retryAfter?.executionAgentNameKey).toBe("ceo");
|
||||
expect(retryAfter?.executionLockedAt).not.toBeNull();
|
||||
|
||||
// Checkout-only sibling: column cleared, no execution side-effects.
|
||||
expect(checkoutOnlyAfter?.checkoutRunId).toBeNull();
|
||||
expect(checkoutOnlyAfter?.executionRunId).toBeNull();
|
||||
});
|
||||
|
||||
it("does not touch execution locks on issues owned by unrelated runs", async () => {
|
||||
const companyId = await seedCompany();
|
||||
const ceoAgentId = await seedAgent(companyId, "CEO");
|
||||
const seAgentId = await seedAgent(companyId, "SE1");
|
||||
|
||||
const finalizingRunId = randomUUID();
|
||||
const unrelatedRunId = randomUUID();
|
||||
|
||||
const contextIssueId = await seedIssue(companyId);
|
||||
const unrelatedIssueId = await seedIssue(companyId);
|
||||
|
||||
await db.insert(heartbeatRuns).values([
|
||||
{
|
||||
id: finalizingRunId,
|
||||
companyId,
|
||||
agentId: ceoAgentId,
|
||||
invocationSource: "assignment",
|
||||
status: "queued",
|
||||
contextSnapshot: { issueId: contextIssueId },
|
||||
},
|
||||
{
|
||||
id: unrelatedRunId,
|
||||
companyId,
|
||||
agentId: seAgentId,
|
||||
invocationSource: "assignment",
|
||||
status: "running",
|
||||
contextSnapshot: { issueId: unrelatedIssueId },
|
||||
},
|
||||
]);
|
||||
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: finalizingRunId,
|
||||
executionAgentNameKey: "ceo",
|
||||
executionLockedAt: new Date(),
|
||||
})
|
||||
.where(eq(issues.id, contextIssueId));
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: unrelatedRunId,
|
||||
executionAgentNameKey: "se1",
|
||||
executionLockedAt: new Date(),
|
||||
})
|
||||
.where(eq(issues.id, unrelatedIssueId));
|
||||
|
||||
await heartbeatService(db).cancelRun(finalizingRunId);
|
||||
|
||||
const [contextAfter] = await db.select().from(issues).where(eq(issues.id, contextIssueId));
|
||||
const [unrelatedAfter] = await db.select().from(issues).where(eq(issues.id, unrelatedIssueId));
|
||||
|
||||
expect(contextAfter?.executionRunId).toBeNull();
|
||||
expect(unrelatedAfter?.executionRunId).toBe(unrelatedRunId);
|
||||
expect(unrelatedAfter?.executionAgentNameKey).toBe("se1");
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -9524,52 +9524,109 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
const recoveryAgentNameKey = normalizeAgentNameKey(recoveryAgent?.name);
|
||||
|
||||
const promotionResult = await db.transaction(async (tx) => {
|
||||
if (contextIssueId) {
|
||||
await tx.execute(
|
||||
sql`select id from issues where company_id = ${run.companyId} and id = ${contextIssueId} for update`,
|
||||
);
|
||||
} else {
|
||||
await tx.execute(
|
||||
sql`select id from issues where company_id = ${run.companyId} and execution_run_id = ${run.id} for update`,
|
||||
);
|
||||
}
|
||||
// Lock the context issue (if any) AND every issue that still references this run.
|
||||
//
|
||||
// A single run can hold execution locks on multiple issues: the caller's context
|
||||
// issue (set via svc.checkout) plus any additional issues stamped by
|
||||
// enqueueWakeup's "legacy run" fallback when the run was the only queued/running
|
||||
// run matching their contextSnapshot.issueId. Historically this function only
|
||||
// resolved and cleared the lock on *one* issue (rows[0]), leaving the others
|
||||
// with an executionRunId pointing at a finalized run. Subsequent checkouts from
|
||||
// the assigned agent then failed with 409 and the issue stayed blocked forever.
|
||||
// `order by id` makes row-lock acquisition deterministic across concurrent
|
||||
// finalizations, which keeps deadlock risk independent of PostgreSQL's plan
|
||||
// choice when multiple issues match.
|
||||
await tx.execute(
|
||||
contextIssueId
|
||||
? sql`
|
||||
select id from issues
|
||||
where company_id = ${run.companyId}
|
||||
and (
|
||||
id = ${contextIssueId}
|
||||
or execution_run_id = ${run.id}
|
||||
or checkout_run_id = ${run.id}
|
||||
)
|
||||
order by id
|
||||
for update
|
||||
`
|
||||
: sql`
|
||||
select id from issues
|
||||
where company_id = ${run.companyId}
|
||||
and (execution_run_id = ${run.id} or checkout_run_id = ${run.id})
|
||||
order by id
|
||||
for update
|
||||
`,
|
||||
);
|
||||
|
||||
let issue = await tx
|
||||
const candidateIssues = await tx
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
eq(issues.companyId, run.companyId),
|
||||
contextIssueId ? eq(issues.id, contextIssueId) : eq(issues.executionRunId, run.id),
|
||||
contextIssueId
|
||||
? or(
|
||||
eq(issues.id, contextIssueId),
|
||||
eq(issues.executionRunId, run.id),
|
||||
eq(issues.checkoutRunId, run.id),
|
||||
)
|
||||
: or(eq(issues.executionRunId, run.id), eq(issues.checkoutRunId, run.id)),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
.orderBy(asc(issues.id));
|
||||
|
||||
// Clear orphaned execution-lock columns that still point at this finalizing
|
||||
// run, across every sibling issue in one statement so it scales with N
|
||||
// orphans without N round-trips. Rows are already held under FOR UPDATE from
|
||||
// the lock query above.
|
||||
//
|
||||
// The two columns are cleared in separate UPDATEs so we never clobber a
|
||||
// retry's executionRunId pointer: when a process-loss or codex-transient
|
||||
// retry is scheduled mid-finalization, it moves `executionRunId` from this
|
||||
// run to the retry run while leaving `checkoutRunId` pinned at this run.
|
||||
// Only the checkout column should be released in that case; the execution
|
||||
// column now belongs to the retry.
|
||||
const promotionUpdateTimestamp = new Date();
|
||||
await tx
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: null,
|
||||
executionAgentNameKey: null,
|
||||
executionLockedAt: null,
|
||||
updatedAt: promotionUpdateTimestamp,
|
||||
})
|
||||
.where(
|
||||
and(eq(issues.companyId, run.companyId), eq(issues.executionRunId, run.id)),
|
||||
);
|
||||
// `checkoutRunId` clear is symmetric to #6008's per-issue self-heal,
|
||||
// extended to all siblings: covers paths where the issue's assignee or
|
||||
// status changed between checkout and termination, which
|
||||
// adoptStaleCheckoutRun's narrow WHERE clause cannot reach.
|
||||
await tx
|
||||
.update(issues)
|
||||
.set({
|
||||
checkoutRunId: null,
|
||||
updatedAt: promotionUpdateTimestamp,
|
||||
})
|
||||
.where(
|
||||
and(eq(issues.companyId, run.companyId), eq(issues.checkoutRunId, run.id)),
|
||||
);
|
||||
|
||||
// Deferred-wake promotion is bound to a single primary issue: the run's context
|
||||
// issue when present, otherwise the first candidate we found (preserves the
|
||||
// legacy rows[0] selection for runs that were not tied to a specific issue).
|
||||
let issue =
|
||||
(contextIssueId
|
||||
? candidateIssues.find((candidate) => candidate.id === contextIssueId)
|
||||
: candidateIssues[0]) ?? null;
|
||||
|
||||
if (!issue) return null;
|
||||
if (issue.executionRunId && issue.executionRunId !== run.id) return null;
|
||||
|
||||
// Clear lock columns that point at the terminating run. checkoutRunId is
|
||||
// cleared here in addition to executionRunId so the issue self-heals even
|
||||
// if its assignee or status changed between checkout and termination —
|
||||
// adoptStaleCheckoutRun's narrow WHERE clause cannot cover those paths.
|
||||
if (issue.executionRunId === run.id || issue.checkoutRunId === run.id) {
|
||||
await tx
|
||||
.update(issues)
|
||||
.set({
|
||||
checkoutRunId: null,
|
||||
executionRunId: null,
|
||||
executionAgentNameKey: null,
|
||||
executionLockedAt: null,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(issues.id, issue.id),
|
||||
or(eq(issues.executionRunId, run.id), eq(issues.checkoutRunId, run.id)),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
// Workspace-validation recovery: if the finalizing run failed workspace
|
||||
// validation, surface the primary issue for the blocked-recovery comment path.
|
||||
// Sibling lock cleanup is already done above; only the primary issue carries
|
||||
// the recovery surface because the comment is attached to a single issue.
|
||||
if (
|
||||
isWorkspaceValidationFailedRun(run) &&
|
||||
(issue.status === "todo" || issue.status === "in_progress") &&
|
||||
@@ -9585,6 +9642,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
while (true) {
|
||||
const deferred = await tx
|
||||
.select()
|
||||
|
||||
Reference in New Issue
Block a user