From 79ffbbaae03b453f80f00d903d59c6c9990c36a9 Mon Sep 17 00:00:00 2001 From: Chris Date: Fri, 11 Sep 2026 17:03:40 -0400 Subject: [PATCH 1/2] fix: preserve active runs through terminal issue writes Co-Authored-By: Paperclip --- doc/execution-semantics.md | 2 + .../issue-comment-reopen-routes.test.ts | 12 ++-- server/src/__tests__/issues-service.test.ts | 70 +++++++++++++++++++ .../recovery-stale-issue-lock-sweep.test.ts | 9 ++- server/src/routes/issues.ts | 27 ++----- server/src/services/issues.ts | 30 ++++++++ server/src/services/recovery/service.ts | 49 +++++++------ 7 files changed, 147 insertions(+), 52 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index e749807753..2a99d4204a 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -141,6 +141,8 @@ The active-lock lifecycle is part of the checkout contract: - process-loss retry handoff must not leave `checkoutRunId` pinned to the failed run when `executionRunId` moves to the retry run - checkout and checkout-owner checks may self-heal lock columns that point at terminal or missing runs before evaluating conflicts - the recovery sweeper may clear rows whose checkout and execution locks all point at terminal or missing runs +- a terminal issue status does not prove its agent process has stopped; recovery must not terminalize a run while its in-memory execution owner is still active +- a run-scoped checkout request is valid only while the requesting run is queued or running 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. diff --git a/server/src/__tests__/issue-comment-reopen-routes.test.ts b/server/src/__tests__/issue-comment-reopen-routes.test.ts index ee148f1356..782628b1eb 100644 --- a/server/src/__tests__/issue-comment-reopen-routes.test.ts +++ b/server/src/__tests__/issue-comment-reopen-routes.test.ts @@ -1770,11 +1770,11 @@ describe.sequential("issue comment reopen routes", () => { ); }); - it("still implicitly reopens done issues via POST comments when the comment runId differs from the issue's owning run", async () => { + it("does not implicitly reopen done issues via POST comments when finalization already cleared the comment run's lock", async () => { mockIssueService.getById.mockResolvedValue({ ...makeIssue("done"), - checkoutRunId: "run-owning", - executionRunId: "run-owning", + checkoutRunId: null, + executionRunId: null, }); mockIssueService.update.mockImplementation( async (_id: string, patch: Record) => ({ @@ -1794,12 +1794,12 @@ describe.sequential("issue comment reopen routes", () => { }), ) .post("/api/issues/11111111-1111-4111-8111-111111111111/comments") - .send({ body: "Real human follow-up — please reopen" }); + .send({ body: "Done — final note after the run lock was released" }); expect(res.status).toBe(201); - expect(mockIssueService.update).toHaveBeenCalledWith( + expect(mockIssueService.update).not.toHaveBeenCalledWith( "11111111-1111-4111-8111-111111111111", - { status: "todo" }, + expect.objectContaining({ status: "todo" }), ); }); diff --git a/server/src/__tests__/issues-service.test.ts b/server/src/__tests__/issues-service.test.ts index 3b91c8856c..cddd7cbaa2 100644 --- a/server/src/__tests__/issues-service.test.ts +++ b/server/src/__tests__/issues-service.test.ts @@ -6076,6 +6076,76 @@ describeEmbeddedPostgres("issueService.clearExecutionRunIfTerminal", () => { }); }); + it("checkout refuses a terminal actor run before it can reclaim an issue", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const terminalRunId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(heartbeatRuns).values({ + id: terminalRunId, + companyId, + agentId, + status: "succeeded", + invocationSource: "manual", + startedAt: new Date("2026-06-10T10:00:00.000Z"), + finishedAt: new Date("2026-06-10T10:01:00.000Z"), + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Issue reopened before a terminal run tried to reclaim it", + status: "todo", + priority: "high", + assigneeAgentId: agentId, + }); + + await expect( + svc.checkout(issueId, agentId, ["todo", "in_progress"], terminalRunId), + ).rejects.toMatchObject({ + status: 409, + details: { + code: "issue_checkout_run_not_live", + checkoutRunId: terminalRunId, + runStatus: "succeeded", + }, + }); + + const row = await db + .select({ + status: issues.status, + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + startedAt: issues.startedAt, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(row).toEqual({ + status: "todo", + checkoutRunId: null, + executionRunId: null, + startedAt: null, + }); + }); + it("checkout adoption of a stale checkoutRunId preserves the issue's assigneeUserId", async () => { // Regression for PR #2482 checkout-adoption review finding: any adoption // helper that re-locks an existing in_progress issue (e.g. when the prior diff --git a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts index 2cc53380fb..c68ee4595d 100644 --- a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts +++ b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts @@ -379,14 +379,14 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { .resolves.toEqual([{ checkoutRunId: runningRunId, executionRunId: runningRunId }]); }); - it("preserves a process-less run while its in-process execution is still finalizing", async () => { + it("preserves a terminal issue's run while its in-process execution is still finalizing", async () => { const { companyId, agentId, runningRunId } = await seed(); const issueId = randomUUID(); await db.insert(issues).values({ id: issueId, companyId, - title: "Native finalization remains live", - status: "in_progress", + title: "Terminal issue while executor remains live", + status: "done", priority: "high", assigneeAgentId: agentId, checkoutRunId: runningRunId, @@ -396,8 +396,7 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { await db .update(heartbeatRuns) .set({ - runtimeMode: "native", - processPid: 2_000_000_000, + processPid: process.pid, }) .where(eq(heartbeatRuns.id, runningRunId)); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index e587b51a2e..2829dcb989 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -2276,10 +2276,7 @@ function shouldImplicitlyMoveCommentedIssueToTodo(input: { issueStatus: string | null | undefined; assigneeAgentId: string | null | undefined; actorType: "agent" | "user"; - actorId: string; actorRunId: string | null | undefined; - checkoutRunId: string | null | undefined; - executionRunId: string | null | undefined; requestAddsExplicitBlockers?: boolean; }) { // A request that wires a non-empty blockedByIssueIds list is declaring that @@ -2288,18 +2285,12 @@ function shouldImplicitlyMoveCommentedIssueToTodo(input: { // edits — flipping to todo here would contradict the caller's stated intent // in the same request. if (input.requestAddsExplicitBlockers) return false; - // Local-CLI agents post comments under user auth, so the actor.type is "user" - // even though the comment originates from the same heartbeat run that owns - // the issue lock. Without this guard, an agent that closes its own issue and - // then posts a follow-up comment in the same run silently reopens it. - // Suppress the implicit move whenever the comment's source run matches the - // issue's checkout/execution run. - if ( - typeof input.actorRunId === "string" && - input.actorRunId.length > 0 && - (input.actorRunId === input.checkoutRunId || - input.actorRunId === input.executionRunId) - ) { + // Local-CLI agents post comments under user auth, so actor.type alone cannot + // distinguish a human comment from a run-originated one. Run finalization can + // clear the issue lock before the agent posts its final comment, so equality + // with the current lock is not a reliable discriminator. Any non-empty run id + // means the request is run-originated and must require an explicit resume. + if (typeof input.actorRunId === "string" && input.actorRunId.length > 0) { return false; } // Only human comments should implicitly reopen finished work. @@ -12816,10 +12807,7 @@ export function issueRoutes( issueStatus: existing.status, assigneeAgentId: requestedAssigneeAgentId, actorType: actor.actorType, - actorId: actor.actorId, actorRunId: actor.runId, - checkoutRunId: existing.checkoutRunId, - executionRunId: existing.executionRunId, requestAddsExplicitBlockers: Array.isArray(req.body.blockedByIssueIds) && req.body.blockedByIssueIds.length > 0, @@ -17053,10 +17041,7 @@ export function issueRoutes( issueStatus: issue.status, assigneeAgentId: issue.assigneeAgentId, actorType: actor.actorType, - actorId: actor.actorId, actorRunId: actor.runId, - checkoutRunId: issue.checkoutRunId, - executionRunId: issue.executionRunId, }) || shouldResumeInProgressScheduledRetry); const hasUnresolvedFirstClassBlockers = diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 1197e3fbae..cf582398e3 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -11356,6 +11356,36 @@ export function issueService(db: Db) { }); } + if (checkoutRunId) { + const checkoutRun = await db + .select({ + status: heartbeatRuns.status, + agentId: heartbeatRuns.agentId, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.id, checkoutRunId), + eq(heartbeatRuns.companyId, issueCompany.companyId), + ), + ) + .then((rows) => rows[0] ?? null); + if ( + !checkoutRun || + checkoutRun.agentId !== agentId || + !ACTIVE_RUN_STATUSES.includes(checkoutRun.status) + ) { + throw conflict("Issue checkout requires a live owning run", { + code: "issue_checkout_run_not_live", + issueId: id, + actorAgentId: agentId, + checkoutRunId, + runStatus: + checkoutRun?.agentId === agentId ? checkoutRun.status : null, + }); + } + } + await clearExecutionRunIfTerminal(id); await clearCheckoutRunIfTerminal(id); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 0361621ec1..09f2332461 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -5445,16 +5445,17 @@ export function recoveryService( // state is auditable. It never overwrites a status that another path already // made terminal. // - // Two independent authorities terminalize the run. Either one is enough: + // Two independent authorities can terminalize the run after its in-memory + // execution owner is gone: // // - Issue-terminal authority: the run's issue already reached a terminal // status (done or cancelled), but the run row is still "running". A healthy // run always terminalizes its own row before or just after the issue reaches // a terminal status, so a lasting "running" row under a terminal issue is - // orphaned. This authority does not depend on process death. It is the only - // authority that catches the reuse-lease path: the release stops the sandbox - // but keeps the server process alive, so the in-memory handle and the - // recorded pid can both persist. + // orphaned. This authority does not depend on recorded process death. It + // catches the reuse-lease path after its in-memory execution owner is gone: + // the release stops the sandbox but keeps the server process alive, so the + // recorded pid can persist. // - Process-death authority: the run has no in-memory handle and its recorded // process and process group are both gone. This catches a hard server crash // that skipped the graceful teardown, even when the issue is not terminal. @@ -5484,11 +5485,23 @@ export function recoveryService( if (isNativeRunnerOwnershipHeld(run)) return { terminalized: false, status: run.status }; + // A live in-memory execution is the strongest ownership signal. The agent + // can set its issue to a terminal status before the enclosing heartbeat + // finishes its output, telemetry, and run finalization. Terminalizing here + // would race that still-running executor, release its checkout lock, and + // reject its remaining run-scoped writes as ownership conflicts. + const hasLiveExecution = + deps.liveRunExecutions?.has(run.id) ?? runningProcesses.has(run.id); + if (hasLiveExecution) { + return { terminalized: false, status: run.status }; + } + const pid = run.processPid ?? null; const processGroupId = run.processGroupId ?? null; - // Issue-terminal authority. When the run's issue is terminal, the run row is - // orphaned regardless of process or handle state. Prefer the referencing + // Issue-terminal authority. When the run's issue is terminal and no live + // execution owns it, the run row is orphaned regardless of recorded process + // state. Prefer the referencing // issue status that the caller passed, because a lock column is the direct // link from the stuck "Live" issue to this run. Fall back to the issue id in // the run context snapshot when the caller passed nothing. Skip the fallback @@ -5525,24 +5538,20 @@ export function recoveryService( // group. Require recorded process metadata, so this authority never fires // on a run that has not yet stored its pid. let processGone = false; - const hasLiveExecution = - deps.liveRunExecutions?.has(run.id) ?? runningProcesses.has(run.id); - if (!hasLiveExecution) { - if (typeof pid === "number" || typeof processGroupId === "number") { - const processAlive = - (typeof pid === "number" && isPidAlive(pid)) || - (typeof processGroupId === "number" && - isProcessGroupAlive(processGroupId)); - processGone = !processAlive; - } + if (typeof pid === "number" || typeof processGroupId === "number") { + const processAlive = + (typeof pid === "number" && isPidAlive(pid)) || + (typeof processGroupId === "number" && + isProcessGroupAlive(processGroupId)); + processGone = !processAlive; } // A result-less native run may intentionally have no live provider process // while the native finalization coordinator waits to resume the same // provider session. That coordinator, rather than this generic - // process-death backstop, owns retryable/resumed attempts. Preserve issue - // terminality as the stronger authority, but never interrupt coordinator- - // owned recovery merely because the provider process has exited. + // process-death backstop, owns retryable/resumed attempts. In the absence of + // a terminal issue, never interrupt coordinator-owned recovery merely + // because the provider process has exited. if (!issueTerminalStatus && processGone && run.runtimeMode === "native") { const coordinator = await db .select({ From bac7f2a0e5992ea184b0b215656ce239fc40631f Mon Sep 17 00:00:00 2001 From: Chris Date: Fri, 11 Sep 2026 17:36:17 -0400 Subject: [PATCH 2/2] fix: make checkout run validation atomic Co-Authored-By: Paperclip --- .../issue-stale-execution-lock-routes.test.ts | 4 + server/src/__tests__/issues-service.test.ts | 157 ++++++++++- server/src/services/issues.ts | 258 +++++++++++------- 3 files changed, 326 insertions(+), 93 deletions(-) diff --git a/server/src/__tests__/issue-stale-execution-lock-routes.test.ts b/server/src/__tests__/issue-stale-execution-lock-routes.test.ts index ed2e39ed79..86f91e302d 100644 --- a/server/src/__tests__/issue-stale-execution-lock-routes.test.ts +++ b/server/src/__tests__/issue-stale-execution-lock-routes.test.ts @@ -500,6 +500,10 @@ describeEmbeddedPostgres("stale issue execution lock routes", () => { runtimeConfig: {}, permissions: {}, }); + await db + .update(heartbeatRuns) + .set({ agentId: otherAgentId }) + .where(eq(heartbeatRuns.id, currentRunId)); await db.insert(issues).values({ id: issueId, companyId, diff --git a/server/src/__tests__/issues-service.test.ts b/server/src/__tests__/issues-service.test.ts index cddd7cbaa2..1add36c3d6 100644 --- a/server/src/__tests__/issues-service.test.ts +++ b/server/src/__tests__/issues-service.test.ts @@ -6146,6 +6146,96 @@ describeEmbeddedPostgres("issueService.clearExecutionRunIfTerminal", () => { }); }); + it("rejects checkout when the owning run becomes terminal before the issue mutation", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const checkoutRunId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(heartbeatRuns).values({ + id: checkoutRunId, + companyId, + agentId, + status: "running", + invocationSource: "manual", + startedAt: new Date("2026-08-26T11:16:18.000Z"), + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Checkout racing run completion", + status: "todo", + priority: "critical", + assigneeAgentId: agentId, + }); + + const terminalWriteReady = deferred(); + const allowTerminalCommit = deferred(); + const terminalWrite = db.transaction(async (tx) => { + await tx + .update(heartbeatRuns) + .set({ + status: "succeeded", + finishedAt: new Date("2026-08-26T11:16:19.000Z"), + }) + .where(eq(heartbeatRuns.id, checkoutRunId)); + terminalWriteReady.resolve(); + await allowTerminalCommit.promise; + }); + await terminalWriteReady.promise; + + const checkout = svc.checkout( + issueId, + agentId, + ["todo"], + checkoutRunId, + ); + const checkoutAssertion = expect(checkout).rejects.toMatchObject({ + status: 409, + details: { + code: "issue_checkout_run_not_live", + checkoutRunId, + runStatus: "succeeded", + }, + }); + await new Promise((resolve) => setImmediate(resolve)); + allowTerminalCommit.resolve(); + await terminalWrite; + await checkoutAssertion; + + const row = await db + .select({ + status: issues.status, + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(row).toEqual({ + status: "todo", + checkoutRunId: null, + executionRunId: null, + }); + }); + it("checkout adoption of a stale checkoutRunId preserves the issue's assigneeUserId", async () => { // Regression for PR #2482 checkout-adoption review finding: any adoption // helper that re-locks an existing in_progress issue (e.g. when the prior @@ -6973,7 +7063,12 @@ describeEmbeddedPostgres("issueService.assertCheckoutOwner stale checkout adopti async function seedOwnershipIssue(params: { checkoutStatus: "running" | "failed" | "timed_out"; - actorRunStatus?: "running" | "failed" | "timed_out" | "succeeded"; + actorRunStatus?: + | "scheduled_retry" + | "running" + | "failed" + | "timed_out" + | "succeeded"; assigneeMatchesActor?: boolean; }) { const companyId = randomUUID(); @@ -7108,6 +7203,21 @@ describeEmbeddedPostgres("issueService.assertCheckoutOwner stale checkout adopti }); }); + it("does not let scheduled-retry runs adopt checkout ownership", async () => { + const seeded = await seedOwnershipIssue({ + checkoutStatus: "failed", + actorRunStatus: "scheduled_retry", + }); + + await expect( + svc.assertCheckoutOwner( + seeded.issueId, + seeded.actorAgentId, + seeded.actorRunId, + ), + ).rejects.toMatchObject({ status: 409 }); + }); + it("adopts unowned checkout after a concurrent stale-checkout clear wins the lock race", async () => { const seeded = await seedOwnershipIssue({ checkoutStatus: "failed" }); await db @@ -7164,6 +7274,51 @@ describeEmbeddedPostgres("issueService.assertCheckoutOwner stale checkout adopti }); }); + it("serializes concurrent checkout and unowned ownership assertion without a deadlock", async () => { + const seeded = await seedOwnershipIssue({ checkoutStatus: "failed" }); + await db + .update(issues) + .set({ + checkoutRunId: null, + executionRunId: null, + executionLockedAt: null, + executionAgentNameKey: null, + }) + .where(eq(issues.id, seeded.issueId)); + + const [checkedOut, ownership] = await Promise.all([ + svc.checkout( + seeded.issueId, + seeded.actorAgentId, + ["in_progress"], + seeded.actorRunId, + ), + svc.assertCheckoutOwner( + seeded.issueId, + seeded.actorAgentId, + seeded.actorRunId, + ), + ]); + + expect(checkedOut.checkoutRunId).toBe(seeded.actorRunId); + expect(checkedOut.executionRunId).toBe(seeded.actorRunId); + expect(ownership.checkoutRunId).toBe(seeded.actorRunId); + expect(ownership.executionRunId).toBe(seeded.actorRunId); + + const row = await db + .select({ + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where(eq(issues.id, seeded.issueId)) + .then((rows) => rows[0]); + expect(row).toEqual({ + checkoutRunId: seeded.actorRunId, + executionRunId: seeded.actorRunId, + }); + }); + }); describeEmbeddedPostgres("issueService.addComment createdByRunId", () => { diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index cf582398e3..f639287f1d 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1856,6 +1856,7 @@ type IssueUserContextInput = { type ProjectGoalReader = Pick; type DbReader = Pick; type DbTransaction = Parameters[0]>[0]; +type DbOrTransaction = Db | DbTransaction; type IssueCreateInput = Omit & { labelIds?: string[]; blockedByIssueIds?: string[]; @@ -7482,6 +7483,49 @@ export function issueService(db: Db) { return heartbeatRunIsTerminalOrMissing(dbOrTx, runId); } + async function withActiveCheckoutRun(input: { + issueId: string; + companyId: string; + agentId: string; + checkoutRunId: string; + operation: (tx: DbTransaction) => Promise; + }): Promise { + return db.transaction(async (tx) => { + // Keep checkout's lock order aligned with stale-lock cleanup and run + // finalization. Holding both rows through the mutation makes run + // liveness and issue ownership one atomic decision. + await tx.execute( + sql`select ${issues.id} from ${issues} where ${issues.id} = ${input.issueId} for update`, + ); + await tx.execute( + sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${input.checkoutRunId} for update`, + ); + const checkoutRun = await tx + .select({ + status: heartbeatRuns.status, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, input.checkoutRunId)) + .then((rows) => rows[0] ?? null); + const isOwnedRun = + checkoutRun && + checkoutRun.companyId === input.companyId && + checkoutRun.agentId === input.agentId; + if (!isOwnedRun || !ACTIVE_RUN_STATUSES.includes(checkoutRun.status)) { + throw conflict("Issue checkout requires a live owning run", { + code: "issue_checkout_run_not_live", + issueId: input.issueId, + actorAgentId: input.agentId, + checkoutRunId: input.checkoutRunId, + runStatus: isOwnedRun ? checkoutRun.status : null, + }); + } + return input.operation(tx); + }); + } + async function adoptStaleCheckoutRun(input: { issueId: string; actorAgentId: string; @@ -7536,7 +7580,7 @@ export function issueService(db: Db) { const stale = !existingRun || TERMINAL_HEARTBEAT_RUN_STATUSES.has(existingRun.status); const actorLive = - actorRun && !TERMINAL_HEARTBEAT_RUN_STATUSES.has(actorRun.status); + actorRun && ACTIVE_RUN_STATUSES.includes(actorRun.status); if (!stale || !actorLive) { return { adopted: null, latest: lockedIssue }; } @@ -7591,6 +7635,11 @@ export function issueService(db: Db) { actorRunId: string; }) { return db.transaction(async (tx) => { + // Match checkout's issue -> heartbeat lock order to avoid a deadlock when + // an ownership assertion races a checkout for the same issue and run. + await tx.execute( + sql`select ${issues.id} from ${issues} where ${issues.id} = ${input.issueId} for update`, + ); await tx.execute( sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${input.actorRunId} for update`, ); @@ -7599,7 +7648,7 @@ export function issueService(db: Db) { .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.actorRunId)) .then((rows) => rows[0] ?? null); - if (!actorRun || TERMINAL_HEARTBEAT_RUN_STATUSES.has(actorRun.status)) + if (!actorRun || !ACTIVE_RUN_STATUSES.includes(actorRun.status)) return null; const now = new Date(); @@ -11357,33 +11406,13 @@ export function issueService(db: Db) { } if (checkoutRunId) { - const checkoutRun = await db - .select({ - status: heartbeatRuns.status, - agentId: heartbeatRuns.agentId, - }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.id, checkoutRunId), - eq(heartbeatRuns.companyId, issueCompany.companyId), - ), - ) - .then((rows) => rows[0] ?? null); - if ( - !checkoutRun || - checkoutRun.agentId !== agentId || - !ACTIVE_RUN_STATUSES.includes(checkoutRun.status) - ) { - throw conflict("Issue checkout requires a live owning run", { - code: "issue_checkout_run_not_live", - issueId: id, - actorAgentId: agentId, - checkoutRunId, - runStatus: - checkoutRun?.agentId === agentId ? checkoutRun.status : null, - }); - } + await withActiveCheckoutRun({ + issueId: id, + companyId: issueCompany.companyId, + agentId, + checkoutRunId, + operation: async () => undefined, + }); } await clearExecutionRunIfTerminal(id); @@ -11428,27 +11457,37 @@ export function issueService(db: Db) { eq(issues.executionRunId, checkoutRunId), ) : isNull(issues.executionRunId); - const updated = await db - .update(issues) - .set({ - assigneeAgentId: agentId, - assigneeUserId: null, - checkoutRunId, - executionRunId: checkoutRunId, - status: "in_progress", - startedAt: now, - updatedAt: now, - }) - .where( - and( - eq(issues.id, id), - inArray(issues.status, expectedStatuses), - or(isNull(issues.assigneeAgentId), sameRunAssigneeCondition), - executionLockCondition, - ), - ) - .returning() - .then((rows) => rows[0] ?? null); + const updateIssue = (dbOrTx: DbOrTransaction) => + dbOrTx + .update(issues) + .set({ + assigneeAgentId: agentId, + assigneeUserId: null, + checkoutRunId, + executionRunId: checkoutRunId, + status: "in_progress", + startedAt: now, + updatedAt: now, + }) + .where( + and( + eq(issues.id, id), + inArray(issues.status, expectedStatuses), + or(isNull(issues.assigneeAgentId), sameRunAssigneeCondition), + executionLockCondition, + ), + ) + .returning() + .then((rows) => rows[0] ?? null); + const updated = checkoutRunId + ? await withActiveCheckoutRun({ + issueId: id, + companyId: issueCompany.companyId, + agentId, + checkoutRunId, + operation: updateIssue, + }) + : await updateIssue(db); if (updated) { const [enriched] = await withIssueLabels(db, [updated]); @@ -11477,27 +11516,34 @@ export function issueService(db: Db) { current.executionRunId === checkoutRunId) && checkoutRunId ) { - const adopted = await db - .update(issues) - .set({ - checkoutRunId, - executionRunId: checkoutRunId, - updatedAt: new Date(), - }) - .where( - and( - eq(issues.id, id), - eq(issues.status, "in_progress"), - eq(issues.assigneeAgentId, agentId), - isNull(issues.checkoutRunId), - or( - isNull(issues.executionRunId), - eq(issues.executionRunId, checkoutRunId), - ), - ), - ) - .returning() - .then((rows) => rows[0] ?? null); + const adopted = await withActiveCheckoutRun({ + issueId: id, + companyId: issueCompany.companyId, + agentId, + checkoutRunId, + operation: (tx) => + tx + .update(issues) + .set({ + checkoutRunId, + executionRunId: checkoutRunId, + updatedAt: new Date(), + }) + .where( + and( + eq(issues.id, id), + eq(issues.status, "in_progress"), + eq(issues.assigneeAgentId, agentId), + isNull(issues.checkoutRunId), + or( + isNull(issues.executionRunId), + eq(issues.executionRunId, checkoutRunId), + ), + ), + ) + .returning() + .then((rows) => rows[0] ?? null), + }); if (adopted) return adopted; } @@ -11539,6 +11585,7 @@ export function issueService(db: Db) { current.executionRunId, ); if (stale) { + const previousExecutionRunId = current.executionRunId; const now = new Date(); const adoptionSet: Record = { assigneeAgentId: agentId, @@ -11552,22 +11599,29 @@ export function issueService(db: Db) { if (current.status !== "in_progress") { adoptionSet.startedAt = now; } - const adopted = await db - .update(issues) - .set(adoptionSet) - .where( - and( - eq(issues.id, id), - inArray(issues.status, expectedStatuses), - eq(issues.executionRunId, current.executionRunId), - or( - isNull(issues.assigneeAgentId), - eq(issues.assigneeAgentId, agentId), - ), - ), - ) - .returning() - .then((rows) => rows[0] ?? null); + const adopted = await withActiveCheckoutRun({ + issueId: id, + companyId: issueCompany.companyId, + agentId, + checkoutRunId, + operation: (tx) => + tx + .update(issues) + .set(adoptionSet) + .where( + and( + eq(issues.id, id), + inArray(issues.status, expectedStatuses), + eq(issues.executionRunId, previousExecutionRunId), + or( + isNull(issues.assigneeAgentId), + eq(issues.assigneeAgentId, agentId), + ), + ), + ) + .returning() + .then((rows) => rows[0] ?? null), + }); if (adopted) { const [enriched] = await withIssueLabels(db, [adopted]); return enriched; @@ -11581,11 +11635,31 @@ export function issueService(db: Db) { current.status === "in_progress" && sameRunLock(current.checkoutRunId, checkoutRunId) ) { - const row = await db - .select() - .from(issues) - .where(eq(issues.id, id)) - .then((rows) => rows[0] ?? null); + const row = checkoutRunId + ? await withActiveCheckoutRun({ + issueId: id, + companyId: issueCompany.companyId, + agentId, + checkoutRunId, + operation: (tx) => + tx + .select() + .from(issues) + .where( + and( + eq(issues.id, id), + eq(issues.status, "in_progress"), + eq(issues.assigneeAgentId, agentId), + eq(issues.checkoutRunId, checkoutRunId), + ), + ) + .then((rows) => rows[0] ?? null), + }) + : await db + .select() + .from(issues) + .where(eq(issues.id, id)) + .then((rows) => rows[0] ?? null); if (!row) throw notFound("Issue not found"); const [enriched] = await withIssueLabels(db, [row]); return enriched;