fix(server): adopt stale checkout run ownership (#5413)
## Thinking Path > - Paperclip is a control plane for autonomous AI-agent companies. > - Issue checkout ownership is part of the execution-control layer that prevents two runs from mutating the same task at the same time. > - The current lock model should preserve `409` conflicts for live competing owners, but it should not strand the rightful assignee behind a stale terminal run. > - A same-agent follow-up run can encounter an existing `checkoutRunId` from a failed, timed-out, succeeded, or missing heartbeat run. > - In that case, the new run should safely adopt ownership instead of failing with an ownership conflict. > - This pull request makes stale checkout adoption transactional and keeps live checkout owners protected. > - The benefit is safer run recovery without weakening single-owner checkout semantics. ## Linked Issues or Issue Description - Fixes #5350 - Closes #1508 - Closes #1970 - Closes #2083 - Closes #3158 - Closes #3190 - Related stale-lock PRs reviewed during dedup search: #7536, #6658, #5660, #5442, #6223, #7048, #6824, #6799 ## What Changed - Updated issue checkout ownership recovery so the current assignee can adopt a stale terminal or missing checkout run. - Added row locking around stale checkout adoption to avoid races while replacing `checkoutRunId` / `executionRunId`. - Preserved `409` behavior when a different live checkout owner is still active. - Prevented terminal actor runs from reclaiming an unowned checkout lock after the newer eager stale-checkout clear path. - Fixed the stale checkout test fixture so same-assignee cases do not insert duplicate agent rows. - Added/kept focused coverage for stale checkout adoption and live-owner conflict behavior. - Fixes #5350. ## Verification - Focused tests: ```sh pnpm exec vitest run server/src/__tests__/issues-service.test.ts server/src/__tests__/issue-stale-execution-lock-routes.test.ts ``` Result: ```text 2 passed, 84 tests passed ``` - Server typecheck: ```sh pnpm --filter @paperclipai/server typecheck ``` Result: ```text passed ``` - Live curl smoke confirmed same-agent stale checkout adoption returns `200` instead of `409`. ```text old_run_status=succeeded checkout_http=200 patch_http=200 ``` The PATCH response showed `checkoutRunId` and `executionRunId` updated to the new run id. ### Live curl smoke result <img width="1498" height="570" alt="Live curl smoke showing stale checkout adoption returned 200" src="https://github.com/user-attachments/assets/4bf834de-e3cd-4495-ac5a-74767b439eeb" /> ### Server request log <img width="631" height="131" alt="Server logs showing heartbeat, checkout, and patch requests succeeded" src="https://github.com/user-attachments/assets/ceaaa403-110e-44e8-bac8-5d8506e79cc3" /> ## Risks - Low to medium risk: this touches issue execution lock ownership. - The behavioral shift is intentionally narrow: only the current assignee can adopt stale terminal or missing checkout ownership. - Live checkout owners remain protected with `409`. - No database migration or API contract change. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI GPT-5.5 Codex coding agent with repository tool use, shell execution, code review, and local verification. ## 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 run tests locally and they pass - [x] I have added or updated tests where applicable - [ ] If this change affects the UI, I have included before/after screenshots - [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 ## Cross-references and status (maintainer) - Closes #1508 - Closes #1970 - Closes #2083 - Closes #3158 - Closes #3190 - Status: rebased onto current master; focused tests and server typecheck pass locally; all required CI is green; Greptile is 5/5; master drift verified. --------- Co-authored-by: Devin Foley <devin@paperclip.ing>
This commit is contained in:
Vasu Yadav
@@ -213,6 +213,83 @@ describeEmbeddedPostgres("stale issue execution lock routes", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("lets the current assignee recover a timed_out stale checkout owner during PATCH", async () => {
|
||||
const { companyId, agentId, currentRunId } = await seedCompanyAgentAndRuns();
|
||||
const timedOutRunId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: timedOutRunId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "timed_out",
|
||||
invocationSource: "manual",
|
||||
finishedAt: new Date(),
|
||||
});
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Stale checkout lock",
|
||||
status: "in_progress",
|
||||
priority: "high",
|
||||
assigneeAgentId: agentId,
|
||||
checkoutRunId: timedOutRunId,
|
||||
executionRunId: timedOutRunId,
|
||||
executionAgentNameKey: "codexcoder",
|
||||
executionLockedAt: new Date(),
|
||||
});
|
||||
|
||||
const res = await request(createApp(agentActor(companyId, agentId, currentRunId)))
|
||||
.patch(`/api/issues/${issueId}`)
|
||||
.send({ title: "Recovered stale checkout lock" });
|
||||
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(200);
|
||||
const row = await db
|
||||
.select({
|
||||
checkoutRunId: issues.checkoutRunId,
|
||||
executionRunId: issues.executionRunId,
|
||||
})
|
||||
.from(issues)
|
||||
.where(eq(issues.id, issueId))
|
||||
.then((rows) => rows[0]);
|
||||
expect(row).toEqual({
|
||||
checkoutRunId: currentRunId,
|
||||
executionRunId: currentRunId,
|
||||
});
|
||||
});
|
||||
|
||||
it("still returns 409 when a different live checkout owner is active", async () => {
|
||||
const { companyId, agentId, failedRunId } = await seedCompanyAgentAndRuns();
|
||||
const liveOwnerRunId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: liveOwnerRunId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "running",
|
||||
invocationSource: "manual",
|
||||
startedAt: new Date(),
|
||||
});
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Live checkout lock",
|
||||
status: "in_progress",
|
||||
priority: "high",
|
||||
assigneeAgentId: agentId,
|
||||
checkoutRunId: liveOwnerRunId,
|
||||
executionRunId: liveOwnerRunId,
|
||||
executionAgentNameKey: "codexcoder",
|
||||
executionLockedAt: new Date(),
|
||||
});
|
||||
|
||||
const res = await request(createApp(agentActor(companyId, agentId, failedRunId)))
|
||||
.patch(`/api/issues/${issueId}`)
|
||||
.send({ title: "Should fail" });
|
||||
|
||||
expect(res.status, JSON.stringify(res.body)).toBe(409);
|
||||
expect(res.body?.error).toBe("Issue run ownership conflict");
|
||||
});
|
||||
|
||||
it("restricts admin force-release to board users with company access and writes an audit event", async () => {
|
||||
const { companyId, agentId, failedRunId, currentRunId } = await seedCompanyAgentAndRuns();
|
||||
const issueId = randomUUID();
|
||||
|
||||
@@ -41,6 +41,16 @@ import { buildAgentMentionHref, buildProjectMentionHref, MAX_ISSUE_REQUEST_DEPTH
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
let reject!: (reason?: unknown) => void;
|
||||
const promise = new Promise<T>((promiseResolve, promiseReject) => {
|
||||
resolve = promiseResolve;
|
||||
reject = promiseReject;
|
||||
});
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
describe("issue list limit helpers", () => {
|
||||
it("clamps untrusted issue-list limits to the server maximum", () => {
|
||||
expect(clampIssueListLimit(0)).toBe(1);
|
||||
@@ -4952,3 +4962,229 @@ describeEmbeddedPostgres("accepted plan decomposition", () => {
|
||||
expect(record?.childIssues.every((child) => typeof child.title === "string")).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describeEmbeddedPostgres("issueService.assertCheckoutOwner stale checkout adoption", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let svc!: ReturnType<typeof issueService>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
|
||||
beforeAll(async () => {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issues-checkout-owner-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
svc = issueService(db);
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
await db.delete(issueComments);
|
||||
await db.delete(issueRelations);
|
||||
await db.delete(issueInboxArchives);
|
||||
await db.delete(activityLog);
|
||||
await db.delete(issues);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(executionWorkspaces);
|
||||
await db.delete(projectWorkspaces);
|
||||
await db.delete(projects);
|
||||
await db.delete(goals);
|
||||
await db.delete(agents);
|
||||
await db.delete(instanceSettings);
|
||||
await db.delete(companies);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await tempDb?.cleanup();
|
||||
});
|
||||
|
||||
async function seedOwnershipIssue(params: {
|
||||
checkoutStatus: "running" | "failed" | "timed_out";
|
||||
actorRunStatus?: "running" | "failed" | "timed_out" | "succeeded";
|
||||
assigneeMatchesActor?: boolean;
|
||||
}) {
|
||||
const companyId = randomUUID();
|
||||
const assigneeAgentId = randomUUID();
|
||||
const actorAgentId = params.assigneeMatchesActor === false ? randomUUID() : assigneeAgentId;
|
||||
const staleRunId = randomUUID();
|
||||
const actorRunId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
const actorRunStatus = params.actorRunStatus ?? "running";
|
||||
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
const agentRows = [
|
||||
{
|
||||
id: assigneeAgentId,
|
||||
companyId,
|
||||
name: "Assignee",
|
||||
role: "engineer" as const,
|
||||
status: "active" as const,
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
},
|
||||
];
|
||||
if (actorAgentId !== assigneeAgentId) {
|
||||
agentRows.push({
|
||||
id: actorAgentId,
|
||||
companyId,
|
||||
name: "Actor",
|
||||
role: "engineer",
|
||||
status: "active",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
}
|
||||
await db.insert(agents).values(agentRows);
|
||||
await db.insert(heartbeatRuns).values([
|
||||
{
|
||||
id: staleRunId,
|
||||
companyId,
|
||||
agentId: assigneeAgentId,
|
||||
status: params.checkoutStatus,
|
||||
invocationSource: "manual",
|
||||
finishedAt: params.checkoutStatus === "running" ? null : new Date(),
|
||||
},
|
||||
{
|
||||
id: actorRunId,
|
||||
companyId,
|
||||
agentId: actorAgentId,
|
||||
status: actorRunStatus,
|
||||
invocationSource: "manual",
|
||||
startedAt: actorRunStatus === "running" ? new Date() : null,
|
||||
finishedAt: actorRunStatus === "running" ? null : new Date(),
|
||||
},
|
||||
]);
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Checkout owner recovery",
|
||||
status: "in_progress",
|
||||
priority: "high",
|
||||
assigneeAgentId,
|
||||
checkoutRunId: staleRunId,
|
||||
executionRunId: staleRunId,
|
||||
executionLockedAt: new Date(),
|
||||
executionAgentNameKey: "assignee",
|
||||
});
|
||||
|
||||
return { issueId, assigneeAgentId, actorAgentId, staleRunId, actorRunId };
|
||||
}
|
||||
|
||||
it("lets the current assignee adopt a stale terminal checkout owner", async () => {
|
||||
const seeded = await seedOwnershipIssue({ checkoutStatus: "failed" });
|
||||
|
||||
const ownership = await svc.assertCheckoutOwner(seeded.issueId, seeded.actorAgentId, seeded.actorRunId);
|
||||
|
||||
expect(ownership.checkoutRunId).toBe(seeded.actorRunId);
|
||||
expect(ownership.executionRunId).toBe(seeded.actorRunId);
|
||||
expect(ownership.adoptedFromRunId).toBeNull();
|
||||
});
|
||||
|
||||
it("treats timed_out checkout owners as stale and recoverable", async () => {
|
||||
const seeded = await seedOwnershipIssue({ checkoutStatus: "timed_out" });
|
||||
|
||||
const ownership = await svc.assertCheckoutOwner(seeded.issueId, seeded.actorAgentId, seeded.actorRunId);
|
||||
|
||||
expect(ownership.checkoutRunId).toBe(seeded.actorRunId);
|
||||
expect(ownership.adoptedFromRunId).toBeNull();
|
||||
});
|
||||
|
||||
it("does not allow non-assignees to adopt stale checkout ownership", async () => {
|
||||
const seeded = await seedOwnershipIssue({ checkoutStatus: "failed", assigneeMatchesActor: false });
|
||||
|
||||
await expect(
|
||||
svc.assertCheckoutOwner(seeded.issueId, seeded.actorAgentId, seeded.actorRunId),
|
||||
).rejects.toMatchObject({ status: 409 });
|
||||
});
|
||||
|
||||
it("keeps live checkout owners protected with a 409 conflict", async () => {
|
||||
const seeded = await seedOwnershipIssue({ checkoutStatus: "running" });
|
||||
|
||||
await expect(
|
||||
svc.assertCheckoutOwner(seeded.issueId, seeded.actorAgentId, seeded.actorRunId),
|
||||
).rejects.toMatchObject({ status: 409 });
|
||||
});
|
||||
|
||||
it("does not let terminal actor runs adopt stale checkout ownership", async () => {
|
||||
const seeded = await seedOwnershipIssue({ checkoutStatus: "failed", actorRunStatus: "succeeded" });
|
||||
|
||||
await expect(
|
||||
svc.assertCheckoutOwner(seeded.issueId, seeded.actorAgentId, seeded.actorRunId),
|
||||
).rejects.toMatchObject({ status: 409 });
|
||||
|
||||
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: null,
|
||||
executionRunId: null,
|
||||
});
|
||||
});
|
||||
|
||||
it("adopts unowned checkout after a concurrent stale-checkout clear wins the lock race", async () => {
|
||||
const seeded = await seedOwnershipIssue({ checkoutStatus: "failed" });
|
||||
await db
|
||||
.update(issues)
|
||||
.set({
|
||||
executionRunId: seeded.actorRunId,
|
||||
executionLockedAt: new Date(),
|
||||
})
|
||||
.where(eq(issues.id, seeded.issueId));
|
||||
|
||||
const rowLocked = deferred<void>();
|
||||
const clearCanCommit = deferred<void>();
|
||||
|
||||
const concurrentClear = db.transaction(async (tx) => {
|
||||
await tx.execute(
|
||||
sql`select ${issues.id} from ${issues} where ${issues.id} = ${seeded.issueId} for update`,
|
||||
);
|
||||
rowLocked.resolve();
|
||||
await clearCanCommit.promise;
|
||||
await tx
|
||||
.update(issues)
|
||||
.set({
|
||||
checkoutRunId: null,
|
||||
executionRunId: seeded.actorRunId,
|
||||
executionLockedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(issues.id, seeded.issueId));
|
||||
});
|
||||
|
||||
await rowLocked.promise;
|
||||
|
||||
const ownershipPromise = svc.assertCheckoutOwner(seeded.issueId, seeded.actorAgentId, seeded.actorRunId);
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
clearCanCommit.resolve();
|
||||
await concurrentClear;
|
||||
|
||||
const ownership = await ownershipPromise;
|
||||
expect(ownership.checkoutRunId).toBe(seeded.actorRunId);
|
||||
expect(ownership.executionRunId).toBe(seeded.actorRunId);
|
||||
expect(ownership.adoptedFromRunId).toBeNull();
|
||||
|
||||
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,
|
||||
});
|
||||
});
|
||||
|
||||
});
|
||||
|
||||
+257
-117
@@ -3661,36 +3661,99 @@ export function issueService(db: Db) {
|
||||
actorRunId: string;
|
||||
expectedCheckoutRunId: string;
|
||||
}) {
|
||||
const stale = await isTerminalOrMissingHeartbeatRun(input.expectedCheckoutRunId);
|
||||
if (!stale) return null;
|
||||
return db.transaction(async (tx) => {
|
||||
const lockedIssue = await tx
|
||||
.select({
|
||||
id: issues.id,
|
||||
status: issues.status,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
checkoutRunId: issues.checkoutRunId,
|
||||
executionRunId: issues.executionRunId,
|
||||
})
|
||||
.from(issues)
|
||||
.where(eq(issues.id, input.issueId))
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!lockedIssue) {
|
||||
return { adopted: null, latest: null };
|
||||
}
|
||||
|
||||
const now = new Date();
|
||||
const adopted = await db
|
||||
.update(issues)
|
||||
.set({
|
||||
checkoutRunId: input.actorRunId,
|
||||
executionRunId: input.actorRunId,
|
||||
executionLockedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(issues.id, input.issueId),
|
||||
eq(issues.status, "in_progress"),
|
||||
eq(issues.assigneeAgentId, input.actorAgentId),
|
||||
eq(issues.checkoutRunId, input.expectedCheckoutRunId),
|
||||
if (
|
||||
lockedIssue.status !== "in_progress" ||
|
||||
lockedIssue.assigneeAgentId !== input.actorAgentId ||
|
||||
lockedIssue.checkoutRunId !== input.expectedCheckoutRunId
|
||||
) {
|
||||
return { adopted: null, latest: lockedIssue };
|
||||
}
|
||||
|
||||
await Promise.all([
|
||||
tx.execute(
|
||||
sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${input.expectedCheckoutRunId} for update`,
|
||||
),
|
||||
)
|
||||
.returning({
|
||||
id: issues.id,
|
||||
status: issues.status,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
checkoutRunId: issues.checkoutRunId,
|
||||
executionRunId: issues.executionRunId,
|
||||
})
|
||||
.then((rows) => rows[0] ?? null);
|
||||
tx.execute(
|
||||
sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${input.actorRunId} for update`,
|
||||
),
|
||||
]);
|
||||
const [existingRun, actorRun] = await Promise.all([
|
||||
tx
|
||||
.select({ status: heartbeatRuns.status })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, input.expectedCheckoutRunId))
|
||||
.then((rows) => rows[0] ?? null),
|
||||
tx
|
||||
.select({ status: heartbeatRuns.status })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, input.actorRunId))
|
||||
.then((rows) => rows[0] ?? null),
|
||||
]);
|
||||
const stale = !existingRun || TERMINAL_HEARTBEAT_RUN_STATUSES.has(existingRun.status);
|
||||
const actorLive = actorRun && !TERMINAL_HEARTBEAT_RUN_STATUSES.has(actorRun.status);
|
||||
if (!stale || !actorLive) {
|
||||
return { adopted: null, latest: lockedIssue };
|
||||
}
|
||||
|
||||
return adopted;
|
||||
const now = new Date();
|
||||
const adopted = await tx
|
||||
.update(issues)
|
||||
.set({
|
||||
checkoutRunId: input.actorRunId,
|
||||
executionRunId: input.actorRunId,
|
||||
executionLockedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(issues.id, input.issueId),
|
||||
eq(issues.status, "in_progress"),
|
||||
eq(issues.assigneeAgentId, input.actorAgentId),
|
||||
eq(issues.checkoutRunId, input.expectedCheckoutRunId),
|
||||
),
|
||||
)
|
||||
.returning({
|
||||
id: issues.id,
|
||||
status: issues.status,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
checkoutRunId: issues.checkoutRunId,
|
||||
executionRunId: issues.executionRunId,
|
||||
})
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (adopted) {
|
||||
return { adopted, latest: adopted };
|
||||
}
|
||||
|
||||
const latest = await tx
|
||||
.select({
|
||||
id: issues.id,
|
||||
status: issues.status,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
checkoutRunId: issues.checkoutRunId,
|
||||
executionRunId: issues.executionRunId,
|
||||
})
|
||||
.from(issues)
|
||||
.where(eq(issues.id, input.issueId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return { adopted: null, latest };
|
||||
});
|
||||
}
|
||||
|
||||
async function adoptUnownedCheckoutRun(input: {
|
||||
@@ -3698,34 +3761,46 @@ export function issueService(db: Db) {
|
||||
actorAgentId: string;
|
||||
actorRunId: string;
|
||||
}) {
|
||||
const now = new Date();
|
||||
const adopted = await db
|
||||
.update(issues)
|
||||
.set({
|
||||
checkoutRunId: input.actorRunId,
|
||||
executionRunId: input.actorRunId,
|
||||
executionLockedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(issues.id, input.issueId),
|
||||
eq(issues.status, "in_progress"),
|
||||
eq(issues.assigneeAgentId, input.actorAgentId),
|
||||
isNull(issues.checkoutRunId),
|
||||
or(isNull(issues.executionRunId), eq(issues.executionRunId, input.actorRunId)),
|
||||
),
|
||||
)
|
||||
.returning({
|
||||
id: issues.id,
|
||||
status: issues.status,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
checkoutRunId: issues.checkoutRunId,
|
||||
executionRunId: issues.executionRunId,
|
||||
})
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return db.transaction(async (tx) => {
|
||||
await tx.execute(
|
||||
sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${input.actorRunId} for update`,
|
||||
);
|
||||
const actorRun = await tx
|
||||
.select({ status: heartbeatRuns.status })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, input.actorRunId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!actorRun || TERMINAL_HEARTBEAT_RUN_STATUSES.has(actorRun.status)) return null;
|
||||
|
||||
return adopted;
|
||||
const now = new Date();
|
||||
const adopted = await tx
|
||||
.update(issues)
|
||||
.set({
|
||||
checkoutRunId: input.actorRunId,
|
||||
executionRunId: input.actorRunId,
|
||||
executionLockedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(issues.id, input.issueId),
|
||||
eq(issues.status, "in_progress"),
|
||||
eq(issues.assigneeAgentId, input.actorAgentId),
|
||||
isNull(issues.checkoutRunId),
|
||||
or(isNull(issues.executionRunId), eq(issues.executionRunId, input.actorRunId)),
|
||||
),
|
||||
)
|
||||
.returning({
|
||||
id: issues.id,
|
||||
status: issues.status,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
checkoutRunId: issues.checkoutRunId,
|
||||
executionRunId: issues.executionRunId,
|
||||
})
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
return adopted;
|
||||
});
|
||||
}
|
||||
|
||||
async function clearExecutionRunIfTerminal(issueId: string): Promise<boolean> {
|
||||
@@ -5480,13 +5555,13 @@ export function issueService(db: Db) {
|
||||
current.checkoutRunId &&
|
||||
current.checkoutRunId !== checkoutRunId
|
||||
) {
|
||||
const adopted = await adoptStaleCheckoutRun({
|
||||
const staleAdoption = await adoptStaleCheckoutRun({
|
||||
issueId: id,
|
||||
actorAgentId: agentId,
|
||||
actorRunId: checkoutRunId,
|
||||
expectedCheckoutRunId: current.checkoutRunId,
|
||||
});
|
||||
if (adopted) {
|
||||
if (staleAdoption.adopted) {
|
||||
const row = 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]);
|
||||
@@ -5562,77 +5637,142 @@ export function issueService(db: Db) {
|
||||
assertCheckoutOwner: async (id: string, actorAgentId: string, actorRunId: string | null) => {
|
||||
await clearExecutionRunIfTerminal(id);
|
||||
await clearCheckoutRunIfTerminal(id);
|
||||
const current = await db
|
||||
.select({
|
||||
id: issues.id,
|
||||
status: issues.status,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
checkoutRunId: issues.checkoutRunId,
|
||||
executionRunId: issues.executionRunId,
|
||||
})
|
||||
.from(issues)
|
||||
.where(eq(issues.id, id))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const loadCurrent = () =>
|
||||
db
|
||||
.select({
|
||||
id: issues.id,
|
||||
status: issues.status,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
checkoutRunId: issues.checkoutRunId,
|
||||
executionRunId: issues.executionRunId,
|
||||
})
|
||||
.from(issues)
|
||||
.where(eq(issues.id, id))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const current = await loadCurrent();
|
||||
|
||||
if (!current) throw notFound("Issue not found");
|
||||
|
||||
if (
|
||||
current.status === "in_progress" &&
|
||||
current.assigneeAgentId === actorAgentId &&
|
||||
sameRunLock(current.checkoutRunId, actorRunId)
|
||||
) {
|
||||
return { ...current, adoptedFromRunId: null as string | null };
|
||||
}
|
||||
const resolveSameRunOwnership = (candidate: {
|
||||
id: string;
|
||||
status: string;
|
||||
assigneeAgentId: string | null;
|
||||
checkoutRunId: string | null;
|
||||
executionRunId: string | null;
|
||||
}) => {
|
||||
if (
|
||||
candidate.status === "in_progress" &&
|
||||
candidate.assigneeAgentId === actorAgentId &&
|
||||
sameRunLock(candidate.checkoutRunId, actorRunId)
|
||||
) {
|
||||
return { ...candidate, adoptedFromRunId: null as string | null };
|
||||
}
|
||||
return null;
|
||||
};
|
||||
|
||||
if (
|
||||
actorRunId &&
|
||||
current.status === "in_progress" &&
|
||||
current.assigneeAgentId === actorAgentId &&
|
||||
current.checkoutRunId == null &&
|
||||
(current.executionRunId == null || current.executionRunId === actorRunId)
|
||||
) {
|
||||
const adopted = await adoptUnownedCheckoutRun({
|
||||
issueId: id,
|
||||
const canAdoptUnownedCheckout = (candidate: {
|
||||
status: string;
|
||||
assigneeAgentId: string | null;
|
||||
checkoutRunId: string | null;
|
||||
executionRunId: string | null;
|
||||
}) => (
|
||||
actorRunId
|
||||
&& candidate.status === "in_progress"
|
||||
&& candidate.assigneeAgentId === actorAgentId
|
||||
&& candidate.checkoutRunId == null
|
||||
&& (candidate.executionRunId == null || candidate.executionRunId === actorRunId)
|
||||
);
|
||||
|
||||
const resolveOwnership = async (
|
||||
candidate: {
|
||||
id: string;
|
||||
status: string;
|
||||
assigneeAgentId: string | null;
|
||||
checkoutRunId: string | null;
|
||||
executionRunId: string | null;
|
||||
},
|
||||
) => {
|
||||
const sameRunOwnership = resolveSameRunOwnership(candidate);
|
||||
if (sameRunOwnership) return { ownership: sameRunOwnership, latest: null };
|
||||
|
||||
if (canAdoptUnownedCheckout(candidate)) {
|
||||
const adopted = await adoptUnownedCheckoutRun({
|
||||
issueId: id,
|
||||
actorAgentId,
|
||||
actorRunId: actorRunId!,
|
||||
});
|
||||
|
||||
if (adopted) {
|
||||
return {
|
||||
ownership: {
|
||||
...adopted,
|
||||
adoptedFromRunId: null as string | null,
|
||||
},
|
||||
latest: null,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
if (
|
||||
actorRunId &&
|
||||
candidate.status === "in_progress" &&
|
||||
candidate.assigneeAgentId === actorAgentId &&
|
||||
candidate.checkoutRunId &&
|
||||
candidate.checkoutRunId !== actorRunId
|
||||
) {
|
||||
const previousCheckoutRunId = candidate.checkoutRunId;
|
||||
const staleAdoption = await adoptStaleCheckoutRun({
|
||||
issueId: id,
|
||||
actorAgentId,
|
||||
actorRunId,
|
||||
expectedCheckoutRunId: previousCheckoutRunId,
|
||||
});
|
||||
|
||||
if (staleAdoption.adopted) {
|
||||
return {
|
||||
ownership: {
|
||||
...staleAdoption.adopted,
|
||||
adoptedFromRunId: previousCheckoutRunId,
|
||||
},
|
||||
latest: null,
|
||||
};
|
||||
}
|
||||
|
||||
if (staleAdoption.latest) {
|
||||
const latestOwnership = resolveSameRunOwnership(staleAdoption.latest);
|
||||
if (latestOwnership) return { ownership: latestOwnership, latest: staleAdoption.latest };
|
||||
return { ownership: null, latest: staleAdoption.latest };
|
||||
}
|
||||
}
|
||||
|
||||
return { ownership: null, latest: null };
|
||||
};
|
||||
|
||||
const resolved = await resolveOwnership(current);
|
||||
if (resolved.ownership) return resolved.ownership;
|
||||
|
||||
const latest = resolved.latest ?? await loadCurrent();
|
||||
if (!latest) throw notFound("Issue not found");
|
||||
const resolvedLatest = await resolveOwnership(latest);
|
||||
if (resolvedLatest.ownership) return resolvedLatest.ownership;
|
||||
if (resolvedLatest.latest) {
|
||||
throw conflict("Issue run ownership conflict", {
|
||||
issueId: resolvedLatest.latest.id,
|
||||
status: resolvedLatest.latest.status,
|
||||
assigneeAgentId: resolvedLatest.latest.assigneeAgentId,
|
||||
checkoutRunId: resolvedLatest.latest.checkoutRunId,
|
||||
executionRunId: resolvedLatest.latest.executionRunId,
|
||||
actorAgentId,
|
||||
actorRunId,
|
||||
});
|
||||
|
||||
if (adopted) {
|
||||
return {
|
||||
...adopted,
|
||||
adoptedFromRunId: null as string | null,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
if (
|
||||
actorRunId &&
|
||||
current.status === "in_progress" &&
|
||||
current.assigneeAgentId === actorAgentId &&
|
||||
current.checkoutRunId &&
|
||||
current.checkoutRunId !== actorRunId
|
||||
) {
|
||||
const adopted = await adoptStaleCheckoutRun({
|
||||
issueId: id,
|
||||
actorAgentId,
|
||||
actorRunId,
|
||||
expectedCheckoutRunId: current.checkoutRunId,
|
||||
});
|
||||
|
||||
if (adopted) {
|
||||
return {
|
||||
...adopted,
|
||||
adoptedFromRunId: current.checkoutRunId,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
throw conflict("Issue run ownership conflict", {
|
||||
issueId: current.id,
|
||||
status: current.status,
|
||||
assigneeAgentId: current.assigneeAgentId,
|
||||
checkoutRunId: current.checkoutRunId,
|
||||
executionRunId: current.executionRunId,
|
||||
issueId: latest.id,
|
||||
status: latest.status,
|
||||
assigneeAgentId: latest.assigneeAgentId,
|
||||
checkoutRunId: latest.checkoutRunId,
|
||||
executionRunId: latest.executionRunId,
|
||||
actorAgentId,
|
||||
actorRunId,
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user