From 8b83d69e3cf8575ed7102dcfc906fb4ee06ab287 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 2 Aug 2026 12:12:28 -0700 Subject: [PATCH] feat(heartbeat): serialize shared-workspace issue runs with bounded busy deferrals (#10699) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The heartbeat service dispatches agent runs, and issues in a project can share one project workspace (one working tree on disk). > - Two runs can execute in the same shared workspace at the same time. Each run mutates the same uncommitted files and branches, and the runs corrupt each other's state. > - Multi-agent projects hit this as soon as two issues in one project become active together, so the platform needs to serialize shared-workspace execution instead of relying on luck. > - This pull request adds a pre-dispatch gate: a run whose issue targets a busy shared workspace is deferred with a bounded scheduled retry instead of dispatched. > - The benefit is that concurrent issue runs in one project take turns in the shared working tree, while isolated-workspace runs and unrelated workspaces stay fully parallel. ## Linked Issues or Issue Description Fixes #10645 ## What Changed - `server/src/services/heartbeat.ts`: - New pre-dispatch gate in the run executor. Before adapter dispatch, when the run's issue has a `projectWorkspaceId` and the effective execution workspace mode is `shared_workspace`, the executor looks for a holder: another `running` run whose context issue shares the same project workspace. The gate covers every run shape that reaches adapter dispatch with issue context — assignee execution runs, comment/mention interaction wakes, and review-participant runs. - When a holder exists, the run throws `WorkspaceBusyDeferral` instead of dispatching. The outer catch recognizes the deferral: it cancels the run with `errorCode: "workspace_busy"` (contention is not a failure), cancels its wakeup, schedules a retry through the existing `scheduleBoundedRetryForRun` primitive (`workspace_busy` reason, 60–120 s jittered delay), and returns the agent to idle. The issue execution lock transfers to the scheduled retry run, so the issue keeps an active execution path and stranded-issue recovery does not fire. - An adapter never dispatches alongside a live holder: deferral has no attempt ceiling, so a deferred run keeps rescheduling until the workspace frees. Deadlock safety comes from holder liveness, not a counter — a holder silent past `ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS` (recovery's own "suspicious silence" bar, 1 h) stops counting as a holder, so a zombie run can only delay work, never park it forever, and recovery's silent-run escalation is already reaping it in parallel. If no retry can be scheduled (agent paused, issue reassigned), the deferral releases the issue execution lock so the issue does not strand. - Holder detection honors isolation: when the isolated-workspaces experiment is enabled, holders whose issue settings select `isolated_workspace` / `operator_branch` (or the legacy `isolated` alias) are not counted, because they never touch the shared tree. A NULL or `agent_default` mode counts as a holder — over-serializing is the safe direction. - Non-assignee deferrals survive replay: the deferral stamps `workspaceBusyDeferredWhileAssignee` into the run context (inherited by the scheduled retry), and both the retry promotion gate and the claim-time staleness check exempt a non-assignee `workspace_busy` retry from the reassignment cancellation — for such a retry an assignee mismatch is the expected state, not a reassignment race. An assignee run's retry keeps the full protection: if the issue is reassigned while the retry pends, it still cancels with `issue_reassigned`. - `server/src/__tests__/heartbeat-workspace-busy.test.ts` (new): embedded-Postgres coverage of the full lifecycle plus unit coverage of the delay window. ## Verification - `cd server && pnpm vitest run src/__tests__/heartbeat-workspace-busy.test.ts` — 10 tests: - a run whose issue targets a busy shared workspace is cancelled with `workspace_busy`, its adapter never executes, a `scheduled_retry` run exists with the 60–120 s window, the issue execution lock points at the retry run, the holder run is untouched, and the agent returns to idle; - after the holder finishes, `promoteDueScheduledRetries` + `resumeQueuedRuns` execute the retry run to success; - a non-assignee comment-mention wake defers, does not touch the issue execution lock, and its retry promotes, survives the claim-time staleness check, and executes despite the assignee mismatch; - an assignee retry is still cancelled with `issue_reassigned` when the issue is reassigned while the retry pends; - a holder issue with `executionWorkspaceSettings.mode = "isolated_workspace"` does not cause deferral; - a running run in a different project workspace does not cause deferral; - a holder silent past the staleness threshold does not cause deferral (the run executes); - a retry with ten prior deferrals still defers again — never dispatches — while the holder is live; - delay jitter stays inside the base-to-base-plus-jitter window and clamps out-of-range random sources. - `cd server && pnpm vitest run src/__tests__/heartbeat-` — full heartbeat suite sweep. - `cd server && pnpm run typecheck`. ## Risks - Behavioral shift: shared-workspace runs that used to start immediately now wait for the workspace to free. Against a long-running live holder the wait is unbounded by design — the alternative is dispatching into a held working tree, which is the corruption this PR removes. Every deferral is visible in the run timeline (lifecycle event with the holder run, issue, and attempt number), and the wake is parked, never dropped. - A zombie holder (a `running` row whose process died) delays contending runs by up to the 1 h staleness threshold before it stops counting. Recovery's silent-run escalation targets the same run on the same clock, so this window matches what the system already tolerates for silent active runs. - The holder check and the dispatch are not atomic; two runs that pass the gate in the same instant can still race. The gate closes the common window (a second run waking while the first is mid-execution); the pre-existing sync-conflict handling remains the backstop for the rare simultaneous start. - No schema change, no API change, no new configuration. ## Model Used Claude Fable 5 (`claude-fable-5`, Anthropic) via Claude Code — extended thinking, tool use, full repository access; implementation, tests, and verification runs. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --- .../heartbeat-workspace-busy.test.ts | 720 ++++++++++++++++++ server/src/services/heartbeat.ts | 330 +++++++- 2 files changed, 1037 insertions(+), 13 deletions(-) create mode 100644 server/src/__tests__/heartbeat-workspace-busy.test.ts diff --git a/server/src/__tests__/heartbeat-workspace-busy.test.ts b/server/src/__tests__/heartbeat-workspace-busy.test.ts new file mode 100644 index 0000000000..f757dc78fa --- /dev/null +++ b/server/src/__tests__/heartbeat-workspace-busy.test.ts @@ -0,0 +1,720 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { and, eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + agents, + agentRuntimeState, + agentWakeupRequests, + activityLog, + budgetPolicies, + companies, + companySkills, + createDb, + environmentLeases, + executionWorkspaces, + heartbeatRunEvents, + heartbeatRuns, + issueComments, + issueRelations, + issues, + projects, + projectWorkspaces, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; +import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.ts"; +import { + WORKSPACE_BUSY_ERROR_CODE, + WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, + WORKSPACE_BUSY_RETRY_BASE_DELAY_MS, + WORKSPACE_BUSY_RETRY_JITTER_MS, + WORKSPACE_BUSY_RETRY_REASON, + WORKSPACE_BUSY_RETRY_WAKE_REASON, + computeWorkspaceBusyRetryDelayMs, + heartbeatService, +} from "../services/heartbeat.ts"; +import { instanceSettingsService } from "../services/instance-settings.ts"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; +const WORKSPACE_BUSY_TEST_ADAPTER = "workspace_busy_test"; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres workspace-busy tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describe("computeWorkspaceBusyRetryDelayMs", () => { + it("stays within the base-delay-to-base-plus-jitter window", () => { + expect(computeWorkspaceBusyRetryDelayMs(() => 0)).toBe(WORKSPACE_BUSY_RETRY_BASE_DELAY_MS); + expect(computeWorkspaceBusyRetryDelayMs(() => 0.5)).toBe( + WORKSPACE_BUSY_RETRY_BASE_DELAY_MS + WORKSPACE_BUSY_RETRY_JITTER_MS / 2, + ); + expect(computeWorkspaceBusyRetryDelayMs(() => 1)).toBe( + WORKSPACE_BUSY_RETRY_BASE_DELAY_MS + WORKSPACE_BUSY_RETRY_JITTER_MS, + ); + }); + + it("clamps out-of-range random sources instead of leaving the window", () => { + expect(computeWorkspaceBusyRetryDelayMs(() => -5)).toBe(WORKSPACE_BUSY_RETRY_BASE_DELAY_MS); + expect(computeWorkspaceBusyRetryDelayMs(() => 7)).toBe( + WORKSPACE_BUSY_RETRY_BASE_DELAY_MS + WORKSPACE_BUSY_RETRY_JITTER_MS, + ); + }); +}); + +describeEmbeddedPostgres("shared-workspace run serialization", () => { + let db!: ReturnType; + let heartbeat!: ReturnType; + let tempDb: Awaited> | null = null; + let workspaceCwd!: string; + const executedRunIds: string[] = []; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-workspace-busy-"); + db = createDb(tempDb.connectionString); + heartbeat = heartbeatService(db); + workspaceCwd = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-workspace-busy-")); + registerServerAdapter({ + type: WORKSPACE_BUSY_TEST_ADAPTER, + execute: async (input: { runId?: string }) => { + executedRunIds.push(input.runId ?? "unknown"); + return { + exitCode: 0, + signal: null, + timedOut: false, + resultJson: {}, + }; + }, + testEnvironment: async () => ({ + adapterType: WORKSPACE_BUSY_TEST_ADAPTER, + status: "pass", + checks: [], + testedAt: new Date().toISOString(), + }), + }); + }, 20_000); + + afterEach(async () => { + // Seeded holder runs are synthetic "running" rows with no real execution + // behind them; cancel them first so the drain helper does not spin + // waiting for them to finish. + await db + .update(heartbeatRuns) + .set({ status: "cancelled", finishedAt: new Date() }) + .where(eq(heartbeatRuns.status, "running")); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await cleanupFixture(); + executedRunIds.length = 0; + }); + + afterAll(async () => { + unregisterServerAdapter(WORKSPACE_BUSY_TEST_ADAPTER); + if (workspaceCwd) await fs.rm(workspaceCwd, { recursive: true, force: true }); + await tempDb?.cleanup(); + }); + + async function cleanupFixture() { + for (let attempt = 0; attempt < 5; attempt += 1) { + try { + await cleanupFixtureOnce(); + return; + } catch (error) { + if (attempt === 4) throw error; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + } + } + + async function cleanupFixtureOnce() { + await db.delete(activityLog); + await db.delete(environmentLeases); + await db.delete(issueComments); + await db.delete(issueRelations); + await db.delete(issues); + await db.delete(executionWorkspaces); + await db.delete(projectWorkspaces); + await db.delete(projects); + await db.delete(heartbeatRunEvents); + await db.delete(activityLog); + await new Promise((resolve) => setTimeout(resolve, 25)); + await db.delete(heartbeatRunEvents); + await db.delete(activityLog); + await db.delete(heartbeatRuns); + await db.delete(agentWakeupRequests); + await db.delete(agentRuntimeState); + await db.delete(budgetPolicies); + await db.delete(agents); + await db.delete(companySkills); + await db.delete(companies); + } + + async function waitForRunToLeaveActiveStates(runId: string, timeoutMs = 10_000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const run = await heartbeat.getRun(runId); + if (run && !["queued", "running"].includes(run.status)) return run; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + return await heartbeat.getRun(runId); + } + + interface WorkspaceFixture { + companyId: string; + projectId: string; + projectWorkspaceId: string; + holderAgentId: string; + holderIssueId: string; + holderRunId: string; + agentId: string; + issueId: string; + nonAssigneeAgentId: string; + } + + async function seedWorkspaceFixture(input?: { + holderIssueWorkspaceSettings?: Record | null; + holderProjectWorkspaceId?: string; + holderActivityAt?: Date; + }): Promise { + const companyId = randomUUID(); + const projectId = randomUUID(); + const projectWorkspaceId = randomUUID(); + const holderAgentId = randomUUID(); + const holderIssueId = randomUUID(); + const holderRunId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const nonAssigneeAgentId = randomUUID(); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + const now = new Date(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + defaultResponsibleUserId: "responsible-user", + }); + + await db.insert(projects).values({ + id: projectId, + companyId, + name: "Workspace Busy Project", + }); + + await db.insert(projectWorkspaces).values({ + id: projectWorkspaceId, + companyId, + projectId, + name: "Primary workspace", + sourceType: "local_path", + cwd: workspaceCwd, + isPrimary: true, + }); + + const holderProjectWorkspaceId = input?.holderProjectWorkspaceId ?? projectWorkspaceId; + if (holderProjectWorkspaceId !== projectWorkspaceId) { + await db.insert(projectWorkspaces).values({ + id: holderProjectWorkspaceId, + companyId, + projectId, + name: "Secondary workspace", + sourceType: "local_path", + cwd: workspaceCwd, + }); + } + + for (const [id, name] of [ + [holderAgentId, "HolderCoder"], + [agentId, "DeferredCoder"], + [nonAssigneeAgentId, "CommenterCoder"], + ] as const) { + await db.insert(agents).values({ + id, + companyId, + name, + role: "engineer", + status: id === holderAgentId ? "running" : "idle", + adapterType: WORKSPACE_BUSY_TEST_ADAPTER, + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + }); + } + + const holderActivityAt = input?.holderActivityAt ?? now; + await db.insert(heartbeatRuns).values({ + id: holderRunId, + companyId, + agentId: holderAgentId, + invocationSource: "assignment", + triggerDetail: "system", + status: "running", + startedAt: holderActivityAt, + lastOutputAt: holderActivityAt, + contextSnapshot: { + issueId: holderIssueId, + wakeReason: "issue_assigned", + }, + createdAt: holderActivityAt, + updatedAt: holderActivityAt, + }); + + await db.insert(issues).values({ + id: holderIssueId, + companyId, + title: "Holder issue", + status: "in_progress", + priority: "medium", + responsibleUserId: "responsible-user", + assigneeAgentId: holderAgentId, + projectId, + projectWorkspaceId: holderProjectWorkspaceId, + executionRunId: holderRunId, + executionAgentNameKey: "holdercoder", + executionLockedAt: now, + issueNumber: 1, + identifier: `${issuePrefix}-1`, + ...(input?.holderIssueWorkspaceSettings !== undefined + ? { executionWorkspaceSettings: input.holderIssueWorkspaceSettings } + : {}), + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Deferred issue", + status: "in_progress", + priority: "medium", + responsibleUserId: "responsible-user", + assigneeAgentId: agentId, + projectId, + projectWorkspaceId, + issueNumber: 2, + identifier: `${issuePrefix}-2`, + }); + + return { + companyId, + projectId, + projectWorkspaceId, + holderAgentId, + holderIssueId, + holderRunId, + agentId, + issueId, + nonAssigneeAgentId, + }; + } + + it("defers a run whose issue targets a busy shared workspace and schedules a bounded retry", async () => { + const fixture = await seedWorkspaceFixture(); + + const run = await heartbeat.invoke( + fixture.agentId, + "assignment", + { issueId: fixture.issueId, wakeReason: "issue_assigned" }, + "system", + ); + expect(run).not.toBeNull(); + + const finishedRun = await waitForRunToLeaveActiveStates(run!.id); + expect(finishedRun?.status).toBe("cancelled"); + expect(finishedRun?.errorCode).toBe(WORKSPACE_BUSY_ERROR_CODE); + const workspaceBusy = (finishedRun?.resultJson as Record | null) + ?.workspaceBusy as Record | undefined; + expect(workspaceBusy).toMatchObject({ + projectWorkspaceId: fixture.projectWorkspaceId, + holderRunId: fixture.holderRunId, + holderIssueId: fixture.holderIssueId, + deferralAttempt: 0, + }); + + // The deferred run's adapter never executed — the whole point of the gate. + expect(executedRunIds).not.toContain(run!.id); + + const retryRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.retryOfRunId, run!.id)) + .then((rows) => rows[0] ?? null); + expect(retryRun).toMatchObject({ + status: "scheduled_retry", + scheduledRetryAttempt: 1, + scheduledRetryReason: WORKSPACE_BUSY_RETRY_REASON, + }); + expect((retryRun?.contextSnapshot as Record | null)?.wakeReason).toBe( + WORKSPACE_BUSY_RETRY_WAKE_REASON, + ); + + // The issue execution lock moved to the retry run, so the issue keeps an + // active execution path and stranded-issue recovery leaves it alone. + const issueRow = await db + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, fixture.issueId)) + .then((rows) => rows[0] ?? null); + expect(issueRow?.executionRunId).toBe(retryRun!.id); + + const finishedAtMs = finishedRun?.finishedAt ? new Date(finishedRun.finishedAt).getTime() : 0; + const dueAtMs = retryRun?.scheduledRetryAt ? new Date(retryRun.scheduledRetryAt).getTime() : 0; + expect(dueAtMs).toBeGreaterThanOrEqual(finishedAtMs + WORKSPACE_BUSY_RETRY_BASE_DELAY_MS); + expect(dueAtMs).toBeLessThanOrEqual( + finishedAtMs + WORKSPACE_BUSY_RETRY_BASE_DELAY_MS + WORKSPACE_BUSY_RETRY_JITTER_MS, + ); + + // The holder was left undisturbed and the agent returned to idle rather + // than error. + const holderRun = await heartbeat.getRun(fixture.holderRunId); + expect(holderRun?.status).toBe("running"); + await expect + .poll( + () => + db + .select({ status: agents.status, errorReason: agents.errorReason }) + .from(agents) + .where(eq(agents.id, fixture.agentId)) + .then((rows) => rows[0] ?? null), + { timeout: 5_000, interval: 50 }, + ) + .toEqual({ status: "idle", errorReason: null }); + + if (run!.wakeupRequestId) { + const wakeup = await db + .select({ status: agentWakeupRequests.status }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, run!.wakeupRequestId)) + .then((rows) => rows[0] ?? null); + expect(wakeup?.status).toBe("cancelled"); + } + }); + + it("executes the scheduled retry once the holder has finished", async () => { + const fixture = await seedWorkspaceFixture(); + + const run = await heartbeat.invoke( + fixture.agentId, + "assignment", + { issueId: fixture.issueId, wakeReason: "issue_assigned" }, + "system", + ); + expect(run).not.toBeNull(); + const deferred = await waitForRunToLeaveActiveStates(run!.id); + expect(deferred?.status).toBe("cancelled"); + + const retryRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.retryOfRunId, run!.id)) + .then((rows) => rows[0] ?? null); + expect(retryRun?.status).toBe("scheduled_retry"); + + // Holder finishes; the due retry promotes, queues, and executes. + await db + .update(heartbeatRuns) + .set({ status: "succeeded", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, fixture.holderRunId)); + + const afterDue = new Date(new Date(retryRun!.scheduledRetryAt!).getTime() + 1_000); + const promotion = await heartbeat.promoteDueScheduledRetries(afterDue); + expect(promotion.runIds).toContain(retryRun!.id); + + await heartbeat.resumeQueuedRuns(); + const finishedRetry = await waitForRunToLeaveActiveStates(retryRun!.id); + expect(finishedRetry?.status).toBe("succeeded"); + expect(executedRunIds).toContain(retryRun!.id); + }); + + it("defers a non-assignee run and executes its retry despite the assignee mismatch", async () => { + const fixture = await seedWorkspaceFixture(); + + // A comment-mention wake for an agent that is NOT the issue assignee — + // the interaction-wake shape that legitimately reaches adapter dispatch + // without assignee-ship. + const run = await heartbeat.invoke( + fixture.nonAssigneeAgentId, + "on_demand", + { + issueId: fixture.issueId, + wakeReason: "issue_comment_mentioned", + commentId: randomUUID(), + }, + "system", + ); + expect(run).not.toBeNull(); + + const deferred = await waitForRunToLeaveActiveStates(run!.id); + expect(deferred?.status).toBe("cancelled"); + expect(deferred?.errorCode).toBe(WORKSPACE_BUSY_ERROR_CODE); + expect(executedRunIds).not.toContain(run!.id); + + const retryRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.retryOfRunId, run!.id)) + .then((rows) => rows[0] ?? null); + expect(retryRun).toMatchObject({ + status: "scheduled_retry", + scheduledRetryReason: WORKSPACE_BUSY_RETRY_REASON, + }); + expect( + (retryRun?.contextSnapshot as Record | null)?.workspaceBusyDeferredWhileAssignee, + ).toBe(false); + + // The non-assignee run never held the issue execution lock, so the + // deferral must not have stolen or released it. + const issueRow = await db + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, fixture.issueId)) + .then((rows) => rows[0] ?? null); + expect(issueRow?.executionRunId).toBeNull(); + + // The holder finishes; the retry must survive the promotion gate even + // though the retry's agent is not the issue assignee. + await db + .update(heartbeatRuns) + .set({ status: "succeeded", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, fixture.holderRunId)); + + const afterDue = new Date(new Date(retryRun!.scheduledRetryAt!).getTime() + 1_000); + const promotion = await heartbeat.promoteDueScheduledRetries(afterDue); + expect(promotion.runIds).toContain(retryRun!.id); + + await heartbeat.resumeQueuedRuns(); + const finishedRetry = await waitForRunToLeaveActiveStates(retryRun!.id); + expect(finishedRetry?.status).toBe("succeeded"); + expect(executedRunIds).toContain(retryRun!.id); + }); + + it("still cancels an assignee retry at promotion when the issue is reassigned", async () => { + const fixture = await seedWorkspaceFixture(); + + const run = await heartbeat.invoke( + fixture.agentId, + "assignment", + { issueId: fixture.issueId, wakeReason: "issue_assigned" }, + "system", + ); + expect(run).not.toBeNull(); + const deferred = await waitForRunToLeaveActiveStates(run!.id); + expect(deferred?.status).toBe("cancelled"); + + const retryRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.retryOfRunId, run!.id)) + .then((rows) => rows[0] ?? null); + expect(retryRun?.status).toBe("scheduled_retry"); + expect( + (retryRun?.contextSnapshot as Record | null)?.workspaceBusyDeferredWhileAssignee, + ).toBe(true); + + // Ownership changes while the retry pends: the deferral was taken under + // assignee-ship, so the reassignment protection must still cancel it. + await db + .update(issues) + .set({ assigneeAgentId: fixture.nonAssigneeAgentId, executionRunId: null }) + .where(eq(issues.id, fixture.issueId)); + + const afterDue = new Date(new Date(retryRun!.scheduledRetryAt!).getTime() + 1_000); + const promotion = await heartbeat.promoteDueScheduledRetries(afterDue); + expect(promotion.runIds).not.toContain(retryRun!.id); + + const cancelledRetry = await heartbeat.getRun(retryRun!.id); + expect(cancelledRetry?.status).toBe("cancelled"); + expect(cancelledRetry?.errorCode).toBe("issue_reassigned"); + }); + + it("does not defer when the holder issue explicitly uses an isolated workspace", async () => { + await instanceSettingsService(db).updateExperimental({ enableIsolatedWorkspaces: true }); + try { + const fixture = await seedWorkspaceFixture({ + holderIssueWorkspaceSettings: { mode: "isolated_workspace" }, + }); + + const run = await heartbeat.invoke( + fixture.agentId, + "assignment", + { issueId: fixture.issueId, wakeReason: "issue_assigned" }, + "system", + ); + expect(run).not.toBeNull(); + + const finishedRun = await waitForRunToLeaveActiveStates(run!.id); + expect(finishedRun?.errorCode).not.toBe(WORKSPACE_BUSY_ERROR_CODE); + const retryRuns = await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, fixture.companyId), + eq(heartbeatRuns.scheduledRetryReason, WORKSPACE_BUSY_RETRY_REASON), + ), + ) + .then((rows) => rows.length); + expect(retryRuns).toBe(0); + } finally { + await instanceSettingsService(db).updateExperimental({ enableIsolatedWorkspaces: false }); + } + }); + + it("does not defer when the running run holds a different project workspace", async () => { + const fixture = await seedWorkspaceFixture({ holderProjectWorkspaceId: randomUUID() }); + + const run = await heartbeat.invoke( + fixture.agentId, + "assignment", + { issueId: fixture.issueId, wakeReason: "issue_assigned" }, + "system", + ); + expect(run).not.toBeNull(); + + const finishedRun = await waitForRunToLeaveActiveStates(run!.id); + expect(finishedRun?.errorCode).not.toBe(WORKSPACE_BUSY_ERROR_CODE); + const retryRuns = await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, fixture.companyId), + eq(heartbeatRuns.scheduledRetryReason, WORKSPACE_BUSY_RETRY_REASON), + ), + ) + .then((rows) => rows.length); + expect(retryRuns).toBe(0); + }); + + it("does not defer when the only holder has been silent past the staleness threshold", async () => { + const fixture = await seedWorkspaceFixture({ + holderActivityAt: new Date(Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS - 60_000), + }); + + const run = await heartbeat.invoke( + fixture.agentId, + "assignment", + { issueId: fixture.issueId, wakeReason: "issue_assigned" }, + "system", + ); + expect(run).not.toBeNull(); + + // The holder row is still "running", but it has been silent past the + // staleness bar, so it no longer blocks the workspace. + const finishedRun = await waitForRunToLeaveActiveStates(run!.id); + expect(finishedRun?.status).toBe("succeeded"); + expect(finishedRun?.errorCode).not.toBe(WORKSPACE_BUSY_ERROR_CODE); + expect(executedRunIds).toContain(run!.id); + }); + + it("keeps deferring past earlier attempts while the holder is still live", async () => { + const fixture = await seedWorkspaceFixture(); + + // Seed the promoted continuation of a run that has already been deferred + // many times while the holder is still live: it must defer again, not + // dispatch alongside the live holder. + const priorAttempts = 10; + const priorRunId = randomUUID(); + const wakeupId = randomUUID(); + const retryRunId = randomUUID(); + const now = new Date(); + const dueAt = new Date(now.getTime() - 1_000); + + await db.insert(heartbeatRuns).values({ + id: priorRunId, + companyId: fixture.companyId, + agentId: fixture.agentId, + invocationSource: "assignment", + triggerDetail: "system", + status: "cancelled", + errorCode: WORKSPACE_BUSY_ERROR_CODE, + finishedAt: now, + scheduledRetryAttempt: priorAttempts - 1, + scheduledRetryReason: WORKSPACE_BUSY_RETRY_REASON, + contextSnapshot: { + issueId: fixture.issueId, + wakeReason: WORKSPACE_BUSY_RETRY_WAKE_REASON, + }, + createdAt: now, + updatedAt: now, + }); + + await db.insert(agentWakeupRequests).values({ + id: wakeupId, + companyId: fixture.companyId, + agentId: fixture.agentId, + source: "automation", + triggerDetail: "system", + reason: WORKSPACE_BUSY_RETRY_WAKE_REASON, + payload: { + issueId: fixture.issueId, + retryOfRunId: priorRunId, + retryReason: WORKSPACE_BUSY_RETRY_REASON, + scheduledRetryAttempt: priorAttempts, + }, + status: "queued", + requestedByActorType: "system", + }); + + await db.insert(heartbeatRuns).values({ + id: retryRunId, + companyId: fixture.companyId, + agentId: fixture.agentId, + invocationSource: "automation", + triggerDetail: "system", + status: "scheduled_retry", + wakeupRequestId: wakeupId, + retryOfRunId: priorRunId, + scheduledRetryAt: dueAt, + scheduledRetryAttempt: priorAttempts, + scheduledRetryReason: WORKSPACE_BUSY_RETRY_REASON, + contextSnapshot: { + issueId: fixture.issueId, + wakeReason: WORKSPACE_BUSY_RETRY_WAKE_REASON, + retryReason: WORKSPACE_BUSY_RETRY_REASON, + workspaceBusyDeferredWhileAssignee: true, + }, + createdAt: now, + updatedAt: now, + }); + await db + .update(agentWakeupRequests) + .set({ runId: retryRunId }) + .where(eq(agentWakeupRequests.id, wakeupId)); + + const promotion = await heartbeat.promoteDueScheduledRetries(now); + expect(promotion.runIds).toContain(retryRunId); + + await heartbeat.resumeQueuedRuns(); + const finishedRun = await waitForRunToLeaveActiveStates(retryRunId); + + expect(finishedRun?.status).toBe("cancelled"); + expect(finishedRun?.errorCode).toBe(WORKSPACE_BUSY_ERROR_CODE); + expect(executedRunIds).not.toContain(retryRunId); + + const nextRetry = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.retryOfRunId, retryRunId)) + .then((rows) => rows[0] ?? null); + expect(nextRetry).toMatchObject({ + status: "scheduled_retry", + scheduledRetryAttempt: priorAttempts + 1, + scheduledRetryReason: WORKSPACE_BUSY_RETRY_REASON, + }); + + const holderRun = await heartbeat.getRun(fixture.holderRunId); + expect(holderRun?.status).toBe("running"); + }); +}); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 317e9dab00..3bb96329dd 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -3,7 +3,7 @@ import path from "node:path"; import { execFile as execFileCallback } from "node:child_process"; import { promisify } from "node:util"; import { createHash, randomUUID } from "node:crypto"; -import { and, asc, desc, eq, getTableColumns, gt, gte, inArray, isNull, lt, lte, notInArray, or, sql } from "drizzle-orm"; +import { and, asc, desc, eq, getTableColumns, gt, gte, inArray, isNull, lt, lte, ne, notInArray, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { AGENT_DEFAULT_MAX_CONCURRENT_RUNS, @@ -218,7 +218,7 @@ import { recoveryAssigneeAdapterOverrides, withRecoveryModelProfileHint, } from "./recovery/model-profile-hint.js"; -import { recoveryService } from "./recovery/service.js"; +import { ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS as RECOVERY_ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS, recoveryService } from "./recovery/service.js"; import { productivityReviewService } from "./productivity-review.js"; import { resolveRequiredSuccessfulRunHandoffOnValidPath } from "./successful-run-handoff-state.js"; import { taskWatchdogService } from "./task-watchdogs.js"; @@ -393,6 +393,27 @@ const MAX_TURN_CONTINUATION_MAX_ATTEMPTS_CAP = 10; const MAX_TURN_CONTINUATION_DEFAULT_DELAY_MS = 1_000; const MAX_TURN_CONTINUATION_MAX_DELAY_MS = 5 * 60 * 1000; const MAX_TURN_CONTINUATION_LIVE_RUN_STATUSES = ["scheduled_retry", "queued", "running"] as const; +export const WORKSPACE_BUSY_RETRY_REASON = "workspace_busy"; +export const WORKSPACE_BUSY_RETRY_WAKE_REASON = "workspace_busy_retry"; +export const WORKSPACE_BUSY_ERROR_CODE = "workspace_busy"; +export const WORKSPACE_BUSY_RETRY_BASE_DELAY_MS = 60 * 1000; +export const WORKSPACE_BUSY_RETRY_JITTER_MS = 60 * 1000; +// A running run stops counting as a shared-workspace holder once it has been +// silent this long. This is recovery's own "suspicious silence" bar for active +// runs (scanSilentActiveRuns escalates such runs), so a zombie holder cannot +// park other work on the workspace forever: it stops blocking here at the same +// moment the recovery machinery starts treating it as stuck. A LIVE holder, in +// contrast, never gets overtaken — a deferred run keeps rescheduling until the +// workspace frees, because dispatching alongside a live holder is exactly the +// concurrent-mutation failure this gate exists to prevent. +export const WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS = RECOVERY_ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS; +// Issue-level executionWorkspaceSettings.mode values that unambiguously opt an +// issue's runs out of the shared project workspace, and therefore out of +// shared-workspace serialization ("isolated" is the legacy alias +// parseIssueExecutionWorkspaceSettings normalizes to isolated_workspace). Any +// other value — including agent_default and an absent mode — may still resolve +// to the shared workspace and counts as a holder. +const ISOLATED_EXECUTION_WORKSPACE_MODES = ["isolated_workspace", "operator_branch", "isolated"] as const; type CodexTransientFallbackMode = | "same_session" | "safer_invocation" @@ -430,6 +451,68 @@ export class ConfigurationIncompleteFailure extends Error { } } +export interface SharedWorkspaceHolder { + runId: string; + agentId: string; + issueId: string; + issueIdentifier: string | null; +} + +// Pre-dispatch gate outcome: another running run currently holds the issue's +// shared project workspace. Not a failure — the run is parked as a bounded +// scheduled retry and re-attempted once the holder finishes, so two agents +// never mutate the same working tree concurrently. +export class WorkspaceBusyDeferral extends Error { + code = WORKSPACE_BUSY_ERROR_CODE; + holder: SharedWorkspaceHolder; + projectWorkspaceId: string; + deferralAttempt: number; + wasIssueAssignee: boolean; + + constructor(input: { + holder: SharedWorkspaceHolder; + projectWorkspaceId: string; + deferralAttempt: number; + wasIssueAssignee: boolean; + }) { + super( + `Shared project workspace is busy: run ${input.holder.runId} (issue ${ + input.holder.issueIdentifier ?? input.holder.issueId + }) is still running`, + ); + this.name = "WorkspaceBusyDeferral"; + this.holder = input.holder; + this.projectWorkspaceId = input.projectWorkspaceId; + this.deferralAttempt = input.deferralAttempt; + this.wasIssueAssignee = input.wasIssueAssignee; + } +} + +function isWorkspaceBusyDeferral(error: unknown): error is WorkspaceBusyDeferral { + return error instanceof WorkspaceBusyDeferral; +} + +export function computeWorkspaceBusyRetryDelayMs(random: () => number = Math.random) { + const jitter = Math.min(Math.max(random(), 0), 1); + return WORKSPACE_BUSY_RETRY_BASE_DELAY_MS + Math.floor(jitter * WORKSPACE_BUSY_RETRY_JITTER_MS); +} + +// True for the retry of a workspace-busy deferral whose original run did NOT +// execute under assignee-ship (a comment or review-participant wake). For such +// a retry an assignee mismatch is the expected state, so the reassignment +// protections in the promotion gate and the claim-time staleness check must +// not cancel it — cancelling would silently drop the wake the deferral +// promised to replay. +export function isNonAssigneeWorkspaceBusyRetry( + retryReason: string | null | undefined, + contextSnapshot: Record, +) { + return ( + retryReason === WORKSPACE_BUSY_RETRY_REASON && + contextSnapshot.workspaceBusyDeferredWhileAssignee === false + ); +} + function resolveCodexTransientFallbackMode(attempt: number): CodexTransientFallbackMode { if (attempt <= 1) return "same_session"; if (attempt === 2) return "safer_invocation"; @@ -10229,17 +10312,19 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } if (issue.assigneeAgentId !== run.agentId) { - return { - allowed: false, - reason: "Scheduled retry suppressed because issue ownership changed", - errorCode: "issue_reassigned", - issueId, - details: { + if (!isNonAssigneeWorkspaceBusyRetry(retryReason, contextSnapshot)) { + return { + allowed: false, + reason: "Scheduled retry suppressed because issue ownership changed", + errorCode: "issue_reassigned", issueId, - previousAssigneeAgentId: run.agentId, - currentAssigneeAgentId: issue.assigneeAgentId, - }, - }; + details: { + issueId, + previousAssigneeAgentId: run.agentId, + currentAssigneeAgentId: issue.assigneeAgentId, + }, + }; + } } if (issue.status === "cancelled" || issue.status === "done") { @@ -11176,6 +11261,175 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; } + // Finds a running heartbeat run (other than the caller's) whose context + // issue shares the same project workspace, i.e. the run that currently + // "holds" the shared working tree. Runs that have been silent past + // WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS do not count — a zombie holder must + // not park other work forever, and recovery's silent-run escalation is + // already reaping it. When isolated workspaces are enabled, holders whose + // issue explicitly opted into an isolated workspace never touch the shared + // tree, so they are excluded; a NULL/agent_default mode may resolve to the + // shared tree and counts as a holder (over-serializing is the safe + // direction). When the isolated-workspaces experiment is off, every run + // resolves to the shared tree, so no holder is excluded. + async function findSharedWorkspaceHolder(input: { + companyId: string; + projectWorkspaceId: string; + excludeIssueId: string; + excludeRunId: string; + honorIsolatedWorkspaceModes: boolean; + now?: Date; + }): Promise { + const staleCutoff = new Date( + (input.now ?? new Date()).getTime() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, + ); + return await db + .select({ + runId: heartbeatRuns.id, + agentId: heartbeatRuns.agentId, + issueId: sql`${issues.id}::text`, + issueIdentifier: issues.identifier, + }) + .from(heartbeatRuns) + .innerJoin( + issues, + and( + eq(issues.companyId, heartbeatRuns.companyId), + sql`${issues.id}::text = ${heartbeatRuns.contextSnapshot} ->> 'issueId'`, + ), + ) + .where( + and( + eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.status, "running"), + ne(heartbeatRuns.id, input.excludeRunId), + // Last observed activity: output beats start beats creation. A run + // that started recently but has not written output yet is live. + sql`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt}) >= ${staleCutoff.toISOString()}::timestamptz`, + eq(issues.projectWorkspaceId, input.projectWorkspaceId), + ne(sql`${issues.id}::text`, input.excludeIssueId), + ...(input.honorIsolatedWorkspaceModes + ? [ + or( + // Covers both a NULL settings blob and a blob without a mode + // key; either may still resolve to the shared workspace. + sql`${issues.executionWorkspaceSettings} ->> 'mode' is null`, + notInArray( + sql`${issues.executionWorkspaceSettings} ->> 'mode'`, + [...ISOLATED_EXECUTION_WORKSPACE_MODES], + ), + ), + ] + : []), + ), + ) + .orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + // Terminal handling for a WorkspaceBusyDeferral thrown by the pre-dispatch + // gate: cancel the run (contention is not a failure), schedule a + // workspace_busy retry, and leave the agent idle. The issue execution lock + // transfers to the scheduled retry run inside scheduleBoundedRetryForRun, so + // the issue keeps an active execution path and recovery leaves it alone. + // Deferral has no attempt ceiling — the retry keeps rescheduling while a + // live holder exists, and holder staleness (not a counter) is what prevents + // waiting on a zombie. If no retry could be scheduled (agent no longer + // invokable), the lock is released so the issue does not strand on a + // cancelled run. + async function finalizeWorkspaceBusyDeferral( + run: typeof heartbeatRuns.$inferSelect, + deferral: WorkspaceBusyDeferral, + ) { + const now = new Date(); + const cancelWrite = await setRunStatusIfRunning(run.id, "cancelled", { + error: deferral.message, + errorCode: WORKSPACE_BUSY_ERROR_CODE, + finishedAt: now, + resultJson: { + workspaceBusy: { + projectWorkspaceId: deferral.projectWorkspaceId, + holderRunId: deferral.holder.runId, + holderIssueId: deferral.holder.issueId, + deferralAttempt: deferral.deferralAttempt, + }, + }, + // Recorded on the run (and inherited by the scheduled retry's context) + // so the retry promotion gate can tell a non-assignee wake — where an + // assignee mismatch is the expected state — from a reassignment race. + contextSnapshot: { + ...parseObject(run.contextSnapshot), + workspaceBusyDeferredWhileAssignee: deferral.wasIssueAssignee, + }, + }); + if (!cancelWrite.updated) { + logger.info( + { runId: run.id, currentStatus: cancelWrite.run?.status ?? null }, + "skipping workspace-busy deferral finalization because the run already left running state", + ); + return; + } + await setWakeupStatus(run.wakeupRequestId, "cancelled", { + finishedAt: now, + error: deferral.message, + }).catch(() => undefined); + + const cancelledRun = cancelWrite.run ?? (await getRun(run.id).catch(() => null)); + const agentRow = await getAgent(run.agentId).catch(() => null); + let scheduleOutcome: string | null = null; + if (cancelledRun && agentRow) { + const scheduleResult = await scheduleBoundedRetryForRun(cancelledRun, agentRow, { + now, + retryReason: WORKSPACE_BUSY_RETRY_REASON, + wakeReason: WORKSPACE_BUSY_RETRY_WAKE_REASON, + // Always admit the next attempt: workspace-busy deferral is bounded by + // holder liveness, not by an attempt counter. + maxAttempts: (cancelledRun.scheduledRetryAttempt ?? 0) + 1, + delayMs: computeWorkspaceBusyRetryDelayMs(), + }).catch((scheduleErr) => { + logger.error( + { err: scheduleErr, runId: run.id }, + "failed to schedule workspace-busy retry after deferral", + ); + return null; + }); + scheduleOutcome = scheduleResult?.outcome ?? null; + } + + if (cancelledRun) { + await appendRunEvent(cancelledRun, await nextRunEventSeq(cancelledRun.id), { + eventType: "lifecycle", + stream: "system", + level: "info", + message: + scheduleOutcome === "scheduled" + ? `Deferred: ${deferral.message}. Retry ${deferral.deferralAttempt + 1} scheduled; the run waits for the workspace to free.` + : `Deferred: ${deferral.message}. No retry could be scheduled; releasing the issue for other runs.`, + payload: { + projectWorkspaceId: deferral.projectWorkspaceId, + holderRunId: deferral.holder.runId, + holderIssueId: deferral.holder.issueId, + deferralAttempt: deferral.deferralAttempt, + retryScheduled: scheduleOutcome === "scheduled", + }, + }).catch(() => undefined); + } + + if (cancelledRun && scheduleOutcome !== "scheduled") { + await releaseIssueExecutionAndPromote(cancelledRun).catch((releaseErr) => { + logger.error( + { err: releaseErr, runId: run.id }, + "failed to release issue execution after workspace-busy deferral", + ); + }); + } + + await finalizeAgentStatus(run.agentId, "cancelled", null, { + wasFirstHeartbeat: timerClaimWasFirstHeartbeat(run), + }).catch(() => undefined); + } + async function scheduleInteractionContinuationInfrastructureRetryIfEligible( run: typeof heartbeatRuns.$inferSelect, agent: typeof agents.$inferSelect, @@ -11973,7 +12227,12 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const isCurrentReviewParticipant = reviewParticipant?.type === "agent" && reviewParticipant.agentId === run.agentId; - if (issue.assigneeAgentId !== run.agentId && !isInteractionWake && !isCurrentReviewParticipant) { + if ( + issue.assigneeAgentId !== run.agentId && + !isInteractionWake && + !isCurrentReviewParticipant && + !isNonAssigneeWorkspaceBusyRetry(retryReason, context) + ) { return { stale: true, errorCode: "issue_assignee_changed", @@ -13232,6 +13491,38 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); const effectiveExecutionWorkspaceMode: ReturnType = requestedExecutionWorkspaceMode; + // Serialize shared-workspace execution: two runs mutating the same project + // working tree concurrently corrupt each other's uncommitted state, so a + // run whose issue targets a busy shared workspace is deferred (rescheduled + // retry) instead of dispatched, and keeps deferring until the workspace + // frees — an adapter never dispatches alongside a live holder. Deadlock + // safety comes from the holder query itself: a holder silent past + // WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS stops counting (recovery's + // silent-run escalation is already reaping it), so a zombie can only delay + // work, never park it forever. This covers non-assignee runs (comment and + // review wakes) too — their deferral records that the run never executed + // under assignee-ship, so the retry promotion gate does not cancel it as a + // reassignment. + if (issueRef?.projectWorkspaceId && effectiveExecutionWorkspaceMode === "shared_workspace") { + const workspaceHolder = await findSharedWorkspaceHolder({ + companyId: agent.companyId, + projectWorkspaceId: issueRef.projectWorkspaceId, + excludeIssueId: issueRef.id, + excludeRunId: run.id, + honorIsolatedWorkspaceModes: isolatedWorkspacesEnabled, + }); + if (workspaceHolder) { + throw new WorkspaceBusyDeferral({ + holder: workspaceHolder, + projectWorkspaceId: issueRef.projectWorkspaceId, + deferralAttempt: + run.scheduledRetryReason === WORKSPACE_BUSY_RETRY_REASON + ? (run.scheduledRetryAttempt ?? 0) + : 0, + wasIssueAssignee: issueContext?.assigneeAgentId === agent.id, + }); + } + } const executionPolicy = { executionMode: (await instanceSettings.getGeneral()).executionMode }; let selectedEnvironmentId = environmentResolution.environmentId; if (isExecutionForcedToKubernetes(executionPolicy)) { @@ -15339,6 +15630,18 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); } } catch (outerErr) { + if (isWorkspaceBusyDeferral(outerErr)) { + // Expected contention on a shared project workspace, not a + // failure: park the run as a bounded scheduled retry and leave the + // holder undisturbed. The finally block below still releases + // leases, runtime services, and scratch for this run. + await finalizeWorkspaceBusyDeferral(run, outerErr).catch((deferralErr) => { + logger.error( + { err: deferralErr, runId }, + "failed to finalize workspace-busy deferral", + ); + }); + } else { // Setup code before adapter.execute threw (e.g. ensureRuntimeState, resolveWorkspaceForRun). // The inner catch did not fire, so we must record the failure here. const message = redactCurrentUserText( @@ -15440,6 +15743,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) wasFirstHeartbeat: timerClaimWasFirstHeartbeat(run), }).catch(() => undefined); } + } } finally { const latestRun = await getRun(run.id).catch(() => null); await releaseEnvironmentLeasesForRun({