diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index a23524e0445..f25f6a28b94 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1265,6 +1265,12 @@ checks coordinator ownership in its terminal update, so a result recorded after the process check is also protected. Terminal task status remains authoritative; exhausted finalization retries do not keep an otherwise orphaned run alive. +Only one open task of a routine can hold an execution lock. A failed run that +waits for a scheduled resume keeps its task's lock, so the routine can queue a +newer task meanwhile. When that task's run is claimed while the slot is held, +the claim is rejected with a conflict, rolls back, and the run stays queued +until the slot is released. + A native failure that schedules a same-run resume marks the run `failed` but keeps the task's execution and checkout locks for that resume. The sweep leaves those locks in place and clears them only when the task is done or cancelled. diff --git a/server/src/__tests__/heartbeat-queued-run-claim-isolation.test.ts b/server/src/__tests__/heartbeat-queued-run-claim-isolation.test.ts index 8a6630feeff..ab100717603 100644 --- a/server/src/__tests__/heartbeat-queued-run-claim-isolation.test.ts +++ b/server/src/__tests__/heartbeat-queued-run-claim-isolation.test.ts @@ -4,10 +4,13 @@ import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest import { agents, agentWakeupRequests, + authUsers, companies, + companyMemberships, createDb, heartbeatRuns, externalAgentHolds, + issues, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, @@ -231,6 +234,49 @@ describeEmbeddedPostgres("heartbeat queued-run claim isolation", () => { expect(await runStatus(healthyRunId)).toMatchObject({ status: "succeeded" }); }); + it("keeps a routine run queued while another task holds the routine slot, then runs it once the slot is free", async () => { + const { companyId, agentId } = await insertAgent(); + const origin = { originKind: "routine_execution", originId: randomUUID(), originFingerprint: "scheduled" }; + // The earlier task keeps its lock while its failed run waits for a scheduled resume. + const holderIssueId = randomUUID(); + const holderRunId = randomUUID(); + await db.insert(issues).values({ id: holderIssueId, companyId, title: "Earlier routine task", + status: "in_progress", assigneeAgentId: agentId, ...origin }); + await db.insert(heartbeatRuns).values({ id: holderRunId, companyId, agentId, status: "failed", + finishedAt: new Date(), contextSnapshot: { issueId: holderIssueId } }); + await db.update(issues).set({ executionRunId: holderRunId }).where(sql`${issues.id} = ${holderIssueId}`); + const userId = `routine-owner-${randomUUID()}`; + await db.insert(authUsers).values({ id: userId, name: "Routine owner", email: `${userId}@paperclip.test`, + createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, + status: "active", membershipRole: "owner" }); + const routineIssueId = randomUUID(); + await db.insert(issues).values({ id: routineIssueId, companyId, title: "Current routine task", + status: "todo", assigneeAgentId: agentId, responsibleUserId: userId, ...origin }); + const { runId: routineRunId } = await insertQueuedRun(companyId, agentId, { + requestedByActorType: "user", requestedByActorId: userId, + payload: { manualUserWake: true, issueId: routineIssueId }, + }, new Date(Date.now() - 60_000)); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: routineIssueId } }) + .where(sql`${heartbeatRuns.id} = ${routineRunId}`); + const { runId: healthyRunId } = await insertClaimableRun(companyId, agentId); + + await expect(heartbeat.resumeQueuedRuns()).resolves.toBeUndefined(); + await heartbeat.drainActiveRunExecutions(); + expect(await runStatus(routineRunId)).toMatchObject({ status: "queued", errorCode: null }); + expect(await runStatus(healthyRunId)).toMatchObject({ status: "succeeded" }); + expect(mockAdapterExecute).toHaveBeenCalledOnce(); + + await db.update(issues).set({ executionRunId: null }).where(sql`${issues.id} = ${holderIssueId}`); + await expect(heartbeat.resumeQueuedRuns()).resolves.toBeUndefined(); + await heartbeat.drainActiveRunExecutions(); + expect(await runStatus(routineRunId)).toMatchObject({ status: "succeeded" }); + // Count executions per run: the finished task may queue its own follow-up run. + const executedRunIds = mockAdapterExecute.mock.calls.map(call => ((call as unknown[])[0] as { runId?: string }).runId); + expect(executedRunIds.filter(id => id === routineRunId)).toHaveLength(1); + expect(executedRunIds.filter(id => id === healthyRunId)).toHaveLength(1); + }); + it.each(["worker", "native-effects"])("retains a %s barrier across provider changes and resumes only after reconciliation", async barrier => { const { companyId, agentId } = await insertAgent(); const { runId } = await insertClaimableRun(companyId, agentId); diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index 1dd39dd5067..cf6e2f7bffb 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -693,6 +693,27 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { expect(deps.canStartNewExternalBotWork).not.toHaveBeenCalled(); }); + it("keeps a claim queued while another task of its routine holds the execution slot", async () => { + const { company, agent, issue, run } = await fixture("queued"); + const origin = { originKind: "routine_execution", originId: crypto.randomUUID(), originFingerprint: "scheduled" }; + await db.update(issues).set({ ...origin, executionRunId: null, executionAgentNameKey: null, executionLockedAt: null }) + .where(eq(issues.id, issue.id)); + // An earlier task of the same routine keeps its lock while its failed run waits for a scheduled resume. + const [holder] = await db.insert(issues).values({ companyId: company.id, title: "Earlier routine task", + status: "in_progress", assigneeAgentId: agent.id, ...origin }).returning(); + const [holderRun] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, + status: "failed", finishedAt: new Date(), contextSnapshot: { issueId: holder.id } }).returning(); + await db.update(issues).set({ executionRunId: holderRun.id }).where(eq(issues.id, holder.id)); + const deps = callbacks(db); + deps.getAgent.mockResolvedValue(agent); + await expect(createHeartbeatQueue(db, deps).claimQueuedRun({ ...run, responsibleUserId: "queue-test-user" })) + .rejects.toMatchObject({ status: 409 }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toMatchObject({ status: "queued" }); + expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0]).toMatchObject({ executionRunId: null }); + expect((await db.select().from(issues).where(eq(issues.id, holder.id)))[0]).toMatchObject({ executionRunId: holderRun.id }); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + it("cancels at the daily run cap before claiming or dispatching", async () => { const { run, agent } = await fixture("queued"); const cappedAgent = { ...agent, runtimeConfig: { heartbeat: { maxDailyRuns: 0 } } }; diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index b4a942b3506..dc84e83692c 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -112,6 +112,7 @@ import { notFound, } from "../../errors.js"; import { logger } from "../../middleware/logger.js"; +import { isUniqueViolation } from "../../db-errors.js"; import { publishLiveEvent } from "../live-events.js"; import { allocateHeartbeatRunEventSeq } from "../heartbeat-run-events.js"; import { @@ -1101,7 +1102,15 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende executionAgentNameKey: normalizeAgentNameKey(agent.name), executionLockedAt: claimedAt, updatedAt: claimedAt, - }).where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId))); + }).where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId))).catch((error: unknown) => { + // Another open task of the same routine holds its execution slot, for + // example a failed run that waits for its scheduled resume. Roll back + // this claim and keep the run queued until that slot is released. + if (isUniqueViolation(error, "issues_open_routine_execution_uq")) { + throw conflict("Another execution for this routine is already in progress"); + } + throw error; + }); } const nativeReviewContext = readNativeReviewAssignmentContext(context); const queuedCommentIds = queuedCommentIdsFromRunContext(context);