From 34f71f7d57166e7936b9c30c7e32d52f80649830 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 11 Nov 2026 13:56:45 -0600 Subject: [PATCH] fix(server): defer a run that meets a busy agent directory instead of failing it (#20829) 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 > - Each run of an agent with managed agent files gets a private copy of the agent directory. Remote copies serialize their prepare, checkpoint, collect, and release operations through one Postgres advisory lock per agent, taken with `pg_try_advisory_xact_lock` > - When the lock is busy, the operation fails at once with "Agent directory operation is already active" > - On staging, a deploy restarted the controller while three runs of the same agent were active. When one run resumed, another run of that agent was checkpointing its directory. The resume failed, used its last attempt, and the task became blocked with "Retry or send a message" > - A run must not fail because another run of the same agent is briefly using the agent directory > - This pull request keeps the fail-fast lock and, when it is busy during preparation, puts the same native run back on the existing same-run retry schedule a few seconds later > - The benefit is that lock contention delays a run by seconds instead of ending it as failed ## Linked Issues or Issue Description **What happened?** A task's run failed with `native_continuation_requires_reconciliation` and the error "Agent directory operation is already active". The task was blocked and asked the user to retry. The same error ended at least two other runs of the same agent within one day. **Steps to reproduce** 1. Use an agent with managed agent files in a remote sandbox. 2. Let one run of the agent checkpoint or collect its agent directory. 3. At the same time, let another run of the same agent prepare (for example, a same-run resume after a controller restart). 4. The second run fails immediately with "Agent directory operation is already active". **Expected behavior** The second run waits until the first operation is done, then continues. It does not end as failed. ## What Changed - `server/src/services/agent-directory-working-copies.ts`: The busy lock now throws `AgentDirectoryOperationBusyError`, still a 409 with the same message. Lock behavior is unchanged from `master`: a fail-fast `pg_try_advisory_xact_lock` held for the life of the holder's transaction. - `server/src/services/heartbeat/agent-directory-busy.ts`: `prepareWithAgentDirectoryDeferral` wraps only instruction preparation and turns the busy error into `AgentDirectoryPreparationDeferral` ("Another run of this agent is still using its files"). A busy lock met while collecting files after a turn is not wrapped and keeps its existing path. - **Before native admission** (a brand-new run, such as the first run of a task): `server/src/services/heartbeat/retries.ts` `finalizeAgentDirectoryBusyDeferral` cancels the run as an expected control-plane wait (`providerWorkStarted: false`) and schedules an `agent_directory_busy` retry 5–15 seconds later. The task keeps its execution lock on the retry. The lane spends no failure retries (`execution-recovery-attempt.ts`), its count is the retry run's `scheduledRetryAttempt`, and it stops after 30 attempts. `pre-dispatch-execution-evidence.ts` classifies this wait like the existing workspace and AI-connection waits, so retry scheduling does not ask for reconciliation. - **A resumed native run**: `server/src/services/native-runtime/native-controller-recovery.ts` `deferNativeRunForBusyAgentDirectory` keeps the same run on the existing same-run retry schedule without spending an attempt. It uses the same guards as `deferNativeControllerRecovery`: it locks the task, requires the task's execution run to be this run, requires the current lease owner, and refuses after a Stop, a startup cancellation or a pending native cancellation, so those settle normally. Its count is stored in the run's `contextSnapshot`, which a resume claim does not clear, and it stops after 30. - `server/src/services/heartbeat.ts`: routes a preparation deferral to the same-run path or, before admission, to the scheduled-retry path. After the bound it follows the normal failure path with the plain message. - No schema or migration. Postgres releases a dead holder's advisory lock with its session. The staging evidence showed a live holder (another run's checkpoint), not a stale one. - `doc/DEVELOPING.md`: Documented both paths and the bound. Earlier revisions of this PR waited on the lock and added idle-session expiry, keepalive, ownership fences, fenced remote removal and attempt slots. This revision removes all of them. ### How the directory-wait retry keeps the original wake The retry no longer replaces the wake. `scheduleBoundedRetryForRun` takes `preserveWakeContext` for this wait: the retry keeps the original `wakeReason`, `retryReason` and the whole wake context (interaction, comment, continuation and admission fields), and its wake request keeps the original reason. The wait is recorded separately as `preparationWait: { kind: "agent_directory_busy", attempt, nextAttemptAt }`, next to the carried `executionRetryAccounting`. Gates that key on the wake reason therefore treat the retry exactly as they treated the original wake. | # | Site | Treatment for the directory wait | Why | | --- | --- | --- | --- | | 1 | `heartbeat/retries.ts` `scheduleBoundedRetryForRun` | Changed: `preserveWakeContext` keeps the original wake context and reason and adds `preparationWait` | Root cause: the retry overwrote the wake reason | | 2 | `run-dispatch/domain/wake-context.ts` `isResolvedInteractionContinuationWakeContext` / `PREPARATION_WAIT_RETRY_REASONS` | Special case removed | The original wake context is kept, so the continuation is classified from it | | 3 | `heartbeat/retries.ts` `interactionRunnerWait` | Special case removed | The interaction fields are kept in the context | | 4 | `run-dispatch/adapters/postgres.ts` scheduled-retry dependency facts | Changed: a preparation-wait retry whose original wake was admitted as a dependency-blocked interaction (`dependencyBlockedInteraction`, recorded at wake time) is admitted the same way | Without the wait the original wake would have run; this gate is only met because of the wait | | 5 | `run-dispatch/domain/wake-context.ts` `isNonAssigneeWorkspaceBusyRetry` + `heartbeat/retries.ts` `finalizeAgentDirectoryBusyDeferral` | Kept: `preparationWaitDeferredWhileAssignee`, derived only from the recorded admission authority | The scheduled-retry ownership gate has no wake-reason path; the workspace and AI waits use the same recorded flag | | 6 | `heartbeat/retries.ts` explicit user continuation admission | Kept: admits the `agent_directory_busy` scheduling reason | The successor needs its own authorization receipt | | 7 | `execution-recovery-attempt.ts` uncharged lane and `agentDirectoryWaits` | Kept | Cap and accounting | | 8 | `pre-dispatch-execution-evidence.ts` (`isPreDispatchResourceWait`, `isAgentDirectoryPreparationWait`), used by `legacy-execution-recovery.ts`, `cancelled-native-startup.ts`, `explicit-native-continuation.ts` | Kept | Describe the cancelled pre-provider parent run, not the retry's wake | | 9 | `explicit-native-continuation.ts` `undeliveredLegacyUserCommentIds`, `run-dispatch/adapters/postgres.ts` `unstartedUserTurnOrigin` | Kept | The cancelled parent did not consume the user's comment | | 10 | `execution-projection.ts` label and saved accounting, `task-interruption-presentation.ts`, UI toast and chat suppression | Kept | Presentation of the wait | | 11 | `run-dispatch/domain/policy.ts` AI-wait lock requirement; AI-subscription lock recheck; AI wait detection (`execution-configuration.ts`, `provider-limit-wait.ts`, `heartbeat.ts`, `ai-connections.ts`, `ai-connection-runtime.ts`, `local-ai-login.ts`) | Not applicable | AI-subscription specific | | 12 | `workspace-busy-wait.ts`, `environment-selection.ts`, `workspace-preparation.ts`, UI workspace-holder banners | Not applicable | Workspace holder parking; the directory wait is a timed retry | | 13 | `cancelled-native-startup.ts` paused workspace-retry repair; `failureRetriesBefore*Wait` | Not applicable | Historical workspace and AI records only; every directory wait saves its ledger | | 14 | `failedRunRetryBudgetSpent`; storage-wait exhaustion escalation; `status-relevance.ts` `isAdmissionWait` | Not applicable | A directory wait never fails the run, stops at its own bound, and its run has started | ## Verification - Original wake kinds after a directory wait (`heartbeat-agent-directory-busy.test.ts`, real heartbeat): an assignment runs; a plain user comment on a dependency-blocked task is answered with `dependencyBlockedInteraction`; an accepted interaction continuation keeps its parked-task and reassignment gates; a user message continuation runs with its comment; an accepted non-assignee mention runs; and a reassignment during preparation does not let the previous assignee restart. Each retry keeps the original `wakeReason` and records `preparationWait`. With `preserveWakeContext` off, five of these fail. Restart recovery keeps its attempt credit (`native-runner-restart-recovery.integration.test.ts`). - Non-assignee wake: an accepted mention for an agent other than the assignee meets a busy directory; its retry records `preparationWaitDeferredWhileAssignee: false`, survives promotion, runs and succeeds once the directory is free. Without the shared clause in `isNonAssigneeWorkspaceBusyRetry`, promotion cancels it. - Interaction continuations: `heartbeat-agent-directory-busy.test.ts` runs an accepted interaction continuation into a busy directory. The retry keeps `mutation`, `interactionId`, `interactionKind` and `interactionStatus` and is still classified as a continuation. If the task is moved to todo, or reassigned, while the retry waits, the retry does not run and the task stays parked or with its new owner. With the old explicit storage/Runner lists, both cases fail. - Red/green, new run: `server/src/__tests__/heartbeat-agent-directory-busy.test.ts` runs a real heartbeat for a new assignment whose first instruction preparation meets a busy lock. Without the preparation wrapper, the run ends failed (red). With it, the first run is cancelled as a wait, an `agent_directory_busy` retry is scheduled with the task still locked to it, the retry runs and succeeds, and no run of the task is failed (green). - Red/green, same-run: with the deferral as a no-op, the same-run deferral tests failed (`expected false to be true`); they now pass. - `native-controller-recovery.integration.test.ts` covers repeated claim-and-defer cycles of a resumed run: each cycle keeps the attempt count, cannot be claimed before the delay, stores the count on the run, and the fourth deferral past a bound of 3 is refused. It also covers Stop after the busy error, a foreign lease owner and a moved task (each refused, nothing changed), a run before admission, and that a raw busy error (as from collection) is never deferred. - Focused suites: 232/232 passed (`agent-directory-working-copies`, `instruction-working-copy`, `heartbeat-agent-directory-busy`, `native-controller-recovery.integration`, `heartbeat-retry-scheduling`). `tsc --noEmit` for `server` reports no errors in the changed files. - Staging evidence (read-only): the failed run had a graceful restart at 07:25Z, then same-run resumes. Its third attempt failed at 07:31:08Z. Another run of the same agent recorded a warm checkpoint receipt at 07:31:37Z. A second run of the same agent failed the same way at 07:59:46Z. ## Risks - A deferred run waits 5 to 15 seconds per deferral, up to 30 times. If the holder is still busy after that, the run fails as before with a plain message. - Before admission, deferral cancels the run and creates a retry run, like the existing workspace-busy wait. ## Model Used - Claude Opus 5.5 (claude-opus-5-5), Anthropic, with tool use and code execution in Claude Code. ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/DEVELOPING.md | 9 + doc/run-log-events.md | 22 ++ .../heartbeat-agent-directory-busy.test.ts | 331 ++++++++++++++++++ .../modules/run-dispatch/adapters/postgres.ts | 11 +- .../run-dispatch/domain/wake-context.ts | 13 +- server/src/modules/run-dispatch/index.ts | 3 + .../agent-directory-working-copies.ts | 5 +- .../execution-projection-recovery.test.ts | 13 + server/src/services/execution-projection.ts | 7 +- .../execution-recovery-attempt.test.ts | 24 +- .../services/execution-recovery-attempt.ts | 15 +- .../services/explicit-native-continuation.ts | 4 +- server/src/services/heartbeat.ts | 22 +- .../heartbeat/agent-directory-busy.ts | 55 +++ server/src/services/heartbeat/retries.ts | 73 +++- ...ve-controller-recovery.integration.test.ts | 81 ++++- .../native-controller-recovery.ts | 51 +++ ...unner-restart-recovery.integration.test.ts | 46 +++ .../pre-dispatch-execution-evidence.ts | 19 + .../task-interruption-presentation.ts | 4 +- ui/src/components/TaskChatThread.tsx | 2 +- ui/src/context/LiveUpdatesProvider.tsx | 2 +- 22 files changed, 789 insertions(+), 23 deletions(-) create mode 100644 server/src/__tests__/heartbeat-agent-directory-busy.test.ts create mode 100644 server/src/services/heartbeat/agent-directory-busy.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 123e6bf63de..59b19543503 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -5158,6 +5158,15 @@ lease and save baseline remain unchanged; current canonical files are never uploaded over the retained execution. Remote collection and cleanup serialize through Postgres across controller homes. A missing remote collector retains a pending save instead of treating an empty local cache as the agent's files. +A run whose instruction preparation finds another operation on the same +agent's directory holding that lock does not fail. Before native admission it +is cancelled as an expected wait and a `agent_directory_busy` retry is +scheduled 5 to 15 seconds later; this lane spends no failure retries and stops +after 30 attempts. A native run resuming on its own saved state instead stays +on the same run and is rescheduled through the same-run retry schedule; its +count is kept on the run's context so a resume claim does not reset it, and a +Stop, a lost lease or a moved task leave it to normal settlement. A busy lock +met while collecting files after a turn is not deferred this way. ## Cloud startup configuration diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 417088809f0..ecca5d5a430 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -954,6 +954,28 @@ This pre-provider wait records `ai_connection_busy` on the cancelled run and doe not consume the provider-failure retry allowance. The event contains no credentials and creates no Telemetry or OpenTelemetry export. +## Agent directory waits + +A run whose instruction preparation finds another run of the same agent using +its files writes one informational `lifecycle` event to the local run log, with +`stream: "system"`. It records no file contents, paths or credentials, and +creates no Telemetry or OpenTelemetry export. + +- A resumed native run that stays on its own run writes the message "Another + run of this agent is still using its files; this run continues when it + finishes". Its payload has exactly two fields: `retryReasonCode`, always + `agent_directory_operation_busy`, and `nextAttemptAt`, the ISO time when the + same run is scheduled to try again. +- A run that has not yet started native execution is cancelled with error code + `agent_directory_busy` and writes "Another run of this agent is still using + its files. This task will continue automatically." when an + `agent_directory_busy` retry was scheduled, or "Another run of this agent is + still using its files, and this task could not be rescheduled." when it was + not. Its payload contains only `retryScheduled`, a boolean. + +Neither wait consumes the provider-failure retry allowance. Each stops after 30 +deferrals, after which the run follows the normal failure path. + ## Managed Agent File Save Receipts The server writes `instruction_save` after managed file collection or a warm diff --git a/server/src/__tests__/heartbeat-agent-directory-busy.test.ts b/server/src/__tests__/heartbeat-agent-directory-busy.test.ts new file mode 100644 index 00000000000..3c6838d0a58 --- /dev/null +++ b/server/src/__tests__/heartbeat-agent-directory-busy.test.ts @@ -0,0 +1,331 @@ +import { createHash, randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agentWakeupRequests, + agents, + authUsers, + companies, + companyMemberships, + createDb, + closeRegisteredClients, + heartbeatRuns, + issueComments, + issueRecoveryActions, + issueRelations, + issues, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.ts"; +import { heartbeatService } from "../services/heartbeat.ts"; +import { AgentDirectoryOperationBusyError } from "../services/agent-directory-working-copies.js"; +import { AGENT_DIRECTORY_BUSY_ERROR_CODE, AGENT_DIRECTORY_BUSY_RETRY_REASON } from "../services/heartbeat/agent-directory-busy.js"; +import { undeliveredLegacyUserCommentIds } from "../services/explicit-native-continuation.js"; +import { isResolvedInteractionContinuationWakeContext } from "../modules/run-dispatch/index.js"; + +const busyPrepares = vi.hoisted(() => ({ remaining: 0, calls: 0, beforeBusy: null as null | (() => Promise) })); +vi.mock("../services/agent-instruction-working-copies.js", async importOriginal => { + const actual = await importOriginal(); + return { + ...actual, + agentInstructionWorkingCopyService: (...args: Parameters) => { + const service = actual.agentInstructionWorkingCopyService(...args); + return { + ...service, + prepare: async (input: Parameters[0]) => { + busyPrepares.calls++; + if (busyPrepares.remaining > 0) { + busyPrepares.remaining--; + await busyPrepares.beforeBusy?.(); + throw new AgentDirectoryOperationBusyError(); + } + return service.prepare(input); + }, + }; + }, + }; +}); + +const externalTestDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL?.trim(); +const support = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = support.supported ? describe : describe.skip; +const ADAPTER = "agent_directory_busy_test"; + +describeEmbeddedPostgres("a new run that meets a busy agent directory", () => { + let db!: ReturnType; + let heartbeat!: ReturnType; + let tempDb: Awaited> | null = null; + const executedRunIds: string[] = []; + const executedContexts = new Map>(); + + beforeAll(async () => { + if (externalTestDatabaseUrl) db = createDb(externalTestDatabaseUrl); + else { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-agent-directory-busy-"); + db = createDb(tempDb.connectionString); + } + heartbeat = heartbeatService(db); + registerServerAdapter({ + type: ADAPTER, + supportsInstructionsBundle: true, + execute: async input => { + executedRunIds.push(input.runId); + executedContexts.set(input.runId, input.context as Record); + return { exitCode: 0, signal: null, timedOut: false, resultJson: {} }; + }, + testEnvironment: async () => ({ adapterType: ADAPTER, status: "pass", checks: [], testedAt: new Date().toISOString() }), + }); + }, 20_000); + + afterAll(async () => { + unregisterServerAdapter(ADAPTER); + if (externalTestDatabaseUrl) await closeRegisteredClients(externalTestDatabaseUrl); + await db.$client.end(); + await tempDb?.cleanup(); + }); + + async function seedCompanyAndAgent() { + const companyId = randomUUID(), agentId = randomUUID(); + const now = new Date(); + await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: `B${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "busy-directory-user" }); + await db.insert(authUsers).values({ id: "busy-directory-user", name: "Responsible user", email: "busy-directory@example.test", + createdAt: now, updatedAt: now }).onConflictDoNothing(); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "busy-directory-user", + status: "active", membershipRole: "member" }); + await db.insert(agents).values({ id: agentId, companyId, name: "BusyCoder", role: "engineer", status: "idle", adapterType: ADAPTER, + adapterConfig: {}, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, permissions: {} }); + return { companyId, agentId }; + } + + it("keeps a user's continuation when its run waits for a busy agent directory", async () => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const issueId = randomUUID(), runId = randomUUID(), wakeupRequestId = randomUUID(), sourceRunId = randomUUID(), commentId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId, agentId, status: "failed", + createdAt: new Date("2020-01-01T00:00:00Z"), finishedAt: new Date("2020-01-01T00:00:01Z"), contextSnapshot: { issueId } }); + await db.insert(agentWakeupRequests).values({ id: wakeupRequestId, companyId, agentId, source: "automation", triggerDetail: "system", + reason: "issue_commented", payload: { issueId, commentId }, status: "queued", runId, requestedByActorType: "user", + requestedByActorId: "busy-directory-user" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "automation", triggerDetail: "system", + status: "queued", wakeupRequestId, contextSnapshot: { issueId, taskId: issueId, wakeReason: "issue_commented", wakeCommentId: commentId, + previousRunId: sourceRunId, explicitUserContinuation: { previousRunId: sourceRunId, commentId } } }); + await db.insert(issues).values({ id: issueId, companyId, title: "Continue repair", status: "in_progress", priority: "medium", + responsibleUserId: "busy-directory-user", assigneeAgentId: agentId, checkoutRunId: runId, executionRunId: runId, issueNumber: 2 }); + await db.update(heartbeatRuns).set({ scopeKind: "issue", issueId }).where(eq(heartbeatRuns.id, runId)); + const [comment] = await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorType: "user", + authorUserId: "busy-directory-user", body: "Continue the task." }).returning(); + await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "active_run_watchdog", + cause: "legacy_execution_requires_reconciliation", fingerprint: sourceRunId, status: "resolved", outcome: "cancelled", + nextAction: "User continued.", evidence: { explicitUserContinuation: { runId, previousRunId: sourceRunId, commentId, + actorId: "busy-directory-user", recordedAt: new Date().toISOString(), commentUpdatedAt: comment!.updatedAt.toISOString(), + commentBodyHash: createHash("sha256").update(comment!.body).digest("hex") } } }); + busyPrepares.remaining = 1; + busyPrepares.calls = 0; + + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + + expect(await heartbeat.getRun(runId)).toMatchObject({ status: "cancelled", errorCode: AGENT_DIRECTORY_BUSY_ERROR_CODE }); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, runId)); + expect(retry).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: AGENT_DIRECTORY_BUSY_RETRY_REASON, + contextSnapshot: { wakeReason: "issue_commented", previousRunId: runId, wakeCommentId: commentId, + explicitUserContinuation: { previousRunId: runId, commentId } } }); + const afterDue = new Date(new Date(retry!.scheduledRetryAt!).getTime() + 1_000); + expect((await heartbeat.promoteDueScheduledRetries(afterDue)).runIds).toContain(retry!.id); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + + expect((await heartbeat.getRun(retry!.id))?.status).toBe("succeeded"); + expect(executedContexts.get(retry!.id)).toMatchObject({ wakeCommentId: commentId, + explicitUserContinuation: { previousRunId: runId, commentId } }); + }); + + it("keeps a user comment undelivered when the run that carried it only waited for agent files", async () => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const issueId = randomUUID(), runId = randomUUID(), commentId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Message during a wait", status: "in_progress", priority: "medium", + responsibleUserId: "busy-directory-user", assigneeAgentId: agentId, issueNumber: 3 }); + await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorType: "user", + authorUserId: "busy-directory-user", body: "Please also check the logs." }); + const now = new Date(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "cancelled", runtimeMode: "legacy", + startedAt: now, finishedAt: now, errorCode: AGENT_DIRECTORY_BUSY_ERROR_CODE, + contextSnapshot: { issueId, wakeReason: "issue_commented", wakeCommentId: commentId }, + resultJson: { executionRecovery: { kind: "agent_directory_wait", providerWorkStarted: false }, + cancellation: { source: "control_plane", expected: true, initiator: { type: "system" }, + reason: "Waiting for another run of this agent to finish with its files", recordedAt: now.toISOString() } } }); + expect(await undeliveredLegacyUserCommentIds(db, companyId, issueId, agentId, [commentId])).toEqual([commentId]); + await db.update(heartbeatRuns).set({ errorCode: "operator_cancelled", + resultJson: { cancellation: { source: "operator", expected: true } } }).where(eq(heartbeatRuns.id, runId)); + expect(await undeliveredLegacyUserCommentIds(db, companyId, issueId, agentId, [commentId])).toEqual([]); + }); + + it.each(["parked", "reassigned"] as const)("keeps an accepted interaction continuation's gates across a directory wait: %s", async change => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const otherAgentId = randomUUID(); + await db.insert(agents).values({ id: otherAgentId, companyId, name: "OtherCoder", role: "engineer", status: "idle", adapterType: ADAPTER, + adapterConfig: {}, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, permissions: {} }); + const issueId = randomUUID(), runId = randomUUID(), wakeupRequestId = randomUUID(), interactionId = randomUUID(); + const context = { issueId, taskId: issueId, wakeReason: "issue_commented", mutation: "interaction", interactionId, + interactionKind: "request_confirmation", interactionStatus: "accepted" }; + await db.insert(agentWakeupRequests).values({ id: wakeupRequestId, companyId, agentId, source: "automation", triggerDetail: "system", + reason: "issue_commented", payload: { issueId }, status: "queued", runId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "automation", triggerDetail: "system", + status: "queued", wakeupRequestId, contextSnapshot: context }); + await db.insert(issues).values({ id: issueId, companyId, title: "Approved plan", status: "in_progress", priority: "medium", + responsibleUserId: "busy-directory-user", assigneeAgentId: agentId, executionRunId: runId, issueNumber: 4 }); + await db.update(heartbeatRuns).set({ scopeKind: "issue", issueId }).where(eq(heartbeatRuns.id, runId)); + busyPrepares.remaining = 1; + busyPrepares.calls = 0; + + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + + expect(await heartbeat.getRun(runId)).toMatchObject({ status: "cancelled", errorCode: AGENT_DIRECTORY_BUSY_ERROR_CODE }); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, runId)); + expect(retry).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: AGENT_DIRECTORY_BUSY_RETRY_REASON, + contextSnapshot: { mutation: "interaction", interactionId, interactionKind: "request_confirmation", interactionStatus: "accepted" } }); + expect(isResolvedInteractionContinuationWakeContext(retry!.contextSnapshot)).toBe(true); + + if (change === "parked") await db.update(issues).set({ status: "todo" }).where(eq(issues.id, issueId)); + else await db.update(issues).set({ assigneeAgentId: otherAgentId }).where(eq(issues.id, issueId)); + const afterDue = new Date(new Date(retry!.scheduledRetryAt!).getTime() + 1_000); + await heartbeat.promoteDueScheduledRetries(afterDue); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + + expect((await heartbeat.getRun(retry!.id))?.status).not.toBe("succeeded"); + expect(executedRunIds).not.toContain(retry!.id); + const [task] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(task!.status).toBe(change === "parked" ? "todo" : "in_progress"); + expect(task!.assigneeAgentId).toBe(change === "parked" ? agentId : otherAgentId); + }); + + it("keeps an accepted non-assignee mention through a directory wait and runs it once files are free", async () => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const assigneeId = randomUUID(), issueId = randomUUID(); + await db.insert(agents).values({ id: assigneeId, companyId, name: "AssigneeCoder", role: "engineer", status: "idle", adapterType: ADAPTER, + adapterConfig: {}, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, permissions: {} }); + await db.insert(issues).values({ id: issueId, companyId, title: "Shared review", status: "in_progress", priority: "medium", + responsibleUserId: "busy-directory-user", assigneeAgentId: assigneeId, issueNumber: 5 }); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, invocationSource: "on_demand", status: "queued", + responsibleUserId: "busy-directory-user", + contextSnapshot: { issueId, wakeReason: "issue_comment_mentioned", commentId: randomUUID() } }).returning(); + busyPrepares.remaining = 1; + busyPrepares.calls = 0; + + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + + expect(await heartbeat.getRun(run!.id)).toMatchObject({ status: "cancelled", errorCode: AGENT_DIRECTORY_BUSY_ERROR_CODE }); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run!.id)); + expect(retry).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: AGENT_DIRECTORY_BUSY_RETRY_REASON, + contextSnapshot: { wakeReason: "issue_comment_mentioned", preparationWait: { kind: AGENT_DIRECTORY_BUSY_RETRY_REASON, attempt: 1 } } }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]!.assigneeAgentId).toBe(assigneeId); + + const afterDue = new Date(new Date(retry!.scheduledRetryAt!).getTime() + 1_000); + expect((await heartbeat.promoteDueScheduledRetries(afterDue)).runIds).toContain(retry!.id); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + + expect((await heartbeat.getRun(retry!.id))?.status).toBe("succeeded"); + expect(executedRunIds).toContain(retry!.id); + }); + + it("does not let the previous assignee restart after the task is reassigned during preparation", async () => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const newOwnerId = randomUUID(), issueId = randomUUID(); + await db.insert(agents).values({ id: newOwnerId, companyId, name: "NewOwner", role: "engineer", status: "idle", adapterType: ADAPTER, + adapterConfig: {}, runtimeConfig: { heartbeat: { wakeOnDemand: false, maxConcurrentRuns: 1 } }, permissions: {} }); + await db.insert(issues).values({ id: issueId, companyId, title: "Handed over", status: "in_progress", priority: "medium", + responsibleUserId: "busy-directory-user", assigneeAgentId: agentId, issueNumber: 6 }); + busyPrepares.remaining = 1; + busyPrepares.calls = 0; + busyPrepares.beforeBusy = async () => { + await db.update(issues).set({ assigneeAgentId: newOwnerId }).where(eq(issues.id, issueId)); + }; + try { + const run = await heartbeat.invoke(agentId, "assignment", { issueId, wakeReason: "issue_assigned" }, "system"); + await heartbeat.drainActiveRunExecutions(); + expect(await heartbeat.getRun(run!.id)).toMatchObject({ status: "cancelled", errorCode: AGENT_DIRECTORY_BUSY_ERROR_CODE }); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run!.id)); + if (retry) { + expect(retry.contextSnapshot).toMatchObject({ wakeReason: "issue_assigned", preparationWait: { kind: AGENT_DIRECTORY_BUSY_RETRY_REASON } }); + const afterDue = new Date(new Date(retry.scheduledRetryAt!).getTime() + 1_000); + await heartbeat.promoteDueScheduledRetries(afterDue); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + expect((await heartbeat.getRun(retry.id))?.status).toBe("cancelled"); + expect(executedRunIds).not.toContain(retry.id); + } + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]!.assigneeAgentId).toBe(newOwnerId); + } finally { + busyPrepares.beforeBusy = null; + } + }); + + it("answers a plain comment on a dependency-blocked task after a directory wait, as it would without one", async () => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const blockerId = randomUUID(), issueId = randomUUID(), commentId = randomUUID(); + await db.insert(issues).values([ + { id: blockerId, companyId, title: "Prerequisite", status: "todo", priority: "medium", responsibleUserId: "busy-directory-user", issueNumber: 7 }, + { id: issueId, companyId, title: "Waits on prerequisite", status: "todo", priority: "medium", responsibleUserId: "busy-directory-user", + assigneeAgentId: agentId, issueNumber: 8 }, + ]); + await db.insert(issueRelations).values({ companyId, issueId: blockerId, relatedIssueId: issueId, type: "blocks" }); + await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorType: "user", authorUserId: "busy-directory-user", + body: "What is this waiting on?" }); + busyPrepares.remaining = 1; + busyPrepares.calls = 0; + + const run = await heartbeat.wakeup(agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId, commentId }, contextSnapshot: { issueId, wakeReason: "issue_commented" } }); + expect(run).not.toBeNull(); + await heartbeat.drainActiveRunExecutions(); + + expect(await heartbeat.getRun(run!.id)).toMatchObject({ status: "cancelled", errorCode: AGENT_DIRECTORY_BUSY_ERROR_CODE }); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run!.id)); + expect(retry).toMatchObject({ status: "scheduled_retry", contextSnapshot: { wakeReason: "issue_commented", + preparationWait: { kind: AGENT_DIRECTORY_BUSY_RETRY_REASON, attempt: 1 } } }); + const afterDue = new Date(new Date(retry!.scheduledRetryAt!).getTime() + 1_000); + expect((await heartbeat.promoteDueScheduledRetries(afterDue)).runIds).toContain(retry!.id); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + + expect(await heartbeat.getRun(retry!.id)).toMatchObject({ status: "succeeded", + contextSnapshot: { dependencyBlockedInteraction: true, unresolvedBlockerIssueIds: [blockerId] } }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]!.status).toBe("todo"); + }); + + it("defers the first run of a task instead of failing it, then runs it", async () => { + const { companyId, agentId } = await seedCompanyAndAgent(); + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Daily repair", status: "in_progress", priority: "medium", + responsibleUserId: "busy-directory-user", assigneeAgentId: agentId, issueNumber: 1 }); + busyPrepares.remaining = 1; + busyPrepares.calls = 0; + + const run = await heartbeat.invoke(agentId, "assignment", { issueId, wakeReason: "issue_assigned" }, "system"); + expect(run).not.toBeNull(); + await heartbeat.drainActiveRunExecutions(); + + const first = await heartbeat.getRun(run!.id); + expect(first).toMatchObject({ status: "cancelled", errorCode: AGENT_DIRECTORY_BUSY_ERROR_CODE }); + expect(busyPrepares.calls).toBe(1); + expect(executedRunIds).not.toContain(run!.id); + const [retry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run!.id)); + expect(retry).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: AGENT_DIRECTORY_BUSY_RETRY_REASON, scheduledRetryAttempt: 1, + contextSnapshot: { wakeReason: "issue_assigned", preparationWait: { kind: AGENT_DIRECTORY_BUSY_RETRY_REASON, attempt: 1 } } }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toMatchObject({ status: "in_progress", executionRunId: retry!.id }); + + const afterDue = new Date(new Date(retry!.scheduledRetryAt!).getTime() + 1_000); + expect((await heartbeat.promoteDueScheduledRetries(afterDue)).runIds).toContain(retry!.id); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + + expect((await heartbeat.getRun(retry!.id))?.status).toBe("succeeded"); + expect(executedRunIds).toContain(retry!.id); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.issueId, issueId))) + .filter(row => row.status === "failed")).toEqual([]); + }); +}); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 13615065a5b..84dbbff7769 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -3,6 +3,7 @@ import { notifyDeferredExecutionWork } from "../../../services/deferred-executio import { hasRequiredWorkspaceRecovery } from "../../../services/workspace-restore-recovery-state.js"; import { readNativeWorkspaceSyncReference } from "../../../services/native-runtime/native-workspace-sync.js"; import { isPreDispatchExecutionWaitVerified } from "../../../services/pre-dispatch-execution-wait.js"; +import { isAgentDirectoryPreparationWait } from "../../../services/pre-dispatch-execution-evidence.js"; import { isUserCommentContinuationReason } from "../../../services/explicit-native-continuation.js"; import { queuedCommentIdsFromRunContext, withQueuedCommentIdsInRunContext, withQueuedCommentIdsInWakePayload } from "../../../services/issue-queued-comment-queue.js"; import { notifyExternalAgentSettlement } from "../../external-agents/index.js"; @@ -504,8 +505,13 @@ export function createPostgresRunDispatchAdapter( && input.contextSnapshot.dependencyBlockedInteraction === true && await isVerifiedDependencyInteractionResponse(dbOrTx, { companyId: input.companyId, issueId, agentId: input.agentId, context: input.contextSnapshot }); + const preparationWait = input.contextSnapshot.preparationWait; + const preparationWaitAdmitted = Boolean(readiness && !readiness.isDependencyReady + && input.contextSnapshot.dependencyBlockedInteraction === true + && preparationWait && typeof preparationWait === "object" && !Array.isArray(preparationWait) + && typeof (preparationWait as Record).kind === "string"); facts.dependenciesBlocked = - readiness && !readiness.isDependencyReady && !stallAdmitted && !responseAdmitted + readiness && !readiness.isDependencyReady && !stallAdmitted && !responseAdmitted && !preparationWaitAdmitted ? { unresolvedBlockerIssueIds: readiness.unresolvedBlockerIssueIds, unresolvedBlockerCount: readiness.unresolvedBlockerCount, @@ -1028,7 +1034,8 @@ export function createPostgresRunDispatchAdapter( const recovery = parseObject(parent.resultJson?.executionRecovery); const unstarted = parent.errorCode === WORKSPACE_BUSY_RETRY_REASON ? recovery.kind === "workspace_wait" && recovery.providerWorkStarted === false - : parent.errorCode === "execution_reconciliation_required" && await isPreDispatchExecutionWaitVerified(tx, parent); + : isAgentDirectoryPreparationWait(parent) + || (parent.errorCode === "execution_reconciliation_required" && await isPreDispatchExecutionWaitVerified(tx, parent)); if (!unstarted) return null; const [wake] = await tx.select().from(agentWakeupRequests).where(and( eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), diff --git a/server/src/modules/run-dispatch/domain/wake-context.ts b/server/src/modules/run-dispatch/domain/wake-context.ts index 754fd0f1712..18c7716359e 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.ts @@ -21,6 +21,13 @@ export const AI_CONNECTION_POOL_WAIT_RETRY_REASON = "ai_connection_pool_wait"; export const AI_CONNECTION_BUSY_RETRY_REASON = "ai_connection_busy"; export const RUNNER_REVISION_WAIT_RETRY_REASON = "runner_revision_wait"; export const STORAGE_UNAVAILABLE_WAIT_RETRY_REASON = "storage_unavailable_wait"; +export const AGENT_DIRECTORY_BUSY_RETRY_REASON = "agent_directory_busy"; +export const PREPARATION_WAIT_RETRY_REASONS: ReadonlySet = new Set([ + RUNNER_REVISION_WAIT_RETRY_REASON, STORAGE_UNAVAILABLE_WAIT_RETRY_REASON, +]); +export function isPreparationWaitRetryReason(reason: string | null | undefined): boolean { + return typeof reason === "string" && PREPARATION_WAIT_RETRY_REASONS.has(reason); +} export const INTERACTION_CONTINUATION_INFRA_RETRY_REASON = "interaction_continuation_infra_retry"; export const INTERACTION_CONTINUATION_INFRA_WAKE_REASON = "interaction_continuation_infra_retry"; export const WAKE_COMMENT_IDS_KEY = "wakeCommentIds"; @@ -106,7 +113,8 @@ export function isNonAssigneeWorkspaceBusyRetry( (retryReason === WORKSPACE_BUSY_RETRY_REASON && contextSnapshot.workspaceBusyDeferredWhileAssignee === false) || ((retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON) && - contextSnapshot.aiConnectionBusyDeferredWhileAssignee === false) + contextSnapshot.aiConnectionBusyDeferredWhileAssignee === false) || + (retryReason === AGENT_DIRECTORY_BUSY_RETRY_REASON && contextSnapshot.preparationWaitDeferredWhileAssignee === false) ); } @@ -205,8 +213,7 @@ export function isResolvedInteractionContinuationWakeContext(contextSnapshot: un const retryReason = readNonEmptyString(context.retryReason); return ( (mutation === "interaction" && wakeReason === "issue_commented") || - (mutation === "interaction" && (retryReason === RUNNER_REVISION_WAIT_RETRY_REASON - || retryReason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON)) || + (mutation === "interaction" && isPreparationWaitRetryReason(retryReason)) || (mutation === "interaction" && context.dependencyBlockedInteraction === true && retryReason !== null && RESOLVED_INTERACTION_RESOURCE_WAIT_REASONS.has(retryReason)) || wakeReason === INTERACTION_CONTINUATION_INFRA_WAKE_REASON || diff --git a/server/src/modules/run-dispatch/index.ts b/server/src/modules/run-dispatch/index.ts index 96523837323..6d7678b2cd0 100644 --- a/server/src/modules/run-dispatch/index.ts +++ b/server/src/modules/run-dispatch/index.ts @@ -16,6 +16,9 @@ export { AI_CONNECTION_POOL_WAIT_RETRY_REASON, RUNNER_REVISION_WAIT_RETRY_REASON, STORAGE_UNAVAILABLE_WAIT_RETRY_REASON, + AGENT_DIRECTORY_BUSY_RETRY_REASON, + PREPARATION_WAIT_RETRY_REASONS, + isPreparationWaitRetryReason, INTERACTION_CONTINUATION_INFRA_RETRY_REASON, INTERACTION_CONTINUATION_INFRA_WAKE_REASON, WAKE_COMMENT_IDS_KEY, diff --git a/server/src/services/agent-directory-working-copies.ts b/server/src/services/agent-directory-working-copies.ts index 3e277345440..8109e1ef45b 100644 --- a/server/src/services/agent-directory-working-copies.ts +++ b/server/src/services/agent-directory-working-copies.ts @@ -29,6 +29,9 @@ function adoptionAuthority(row: Copy) { } export class AgentDirectoryReuseInvalidatedError extends Error {} +export class AgentDirectoryOperationBusyError extends HttpError { + constructor() { super(409, "Agent directory operation is already active"); } +} const completed = new Set(["saved", "unchanged", "resolved", "unavailable"]); const transports = new Map(); const key = (row: Pick) => `${row.companyId}:${row.runId}`; @@ -660,7 +663,7 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string return withDedicatedDbConnection(db, dedicated => dedicated.transaction(async tx => { const [lock] = await tx.execute(sql`select pg_try_advisory_xact_lock(hashtextextended(${`agent-directory:${row.companyId}:${row.agentId}`}, 0)) as acquired, (select setting::bigint from pg_settings where name = 'idle_in_transaction_session_timeout') as idle_timeout_ms`); - if (lock?.acquired !== true) throw conflict("Agent directory operation is already active"); + if (lock?.acquired !== true) throw new AgentDirectoryOperationBusyError(); return withWorkspaceLockKeepalive(tx as unknown as Db, Number(lock.idle_timeout_ms), run); })); } diff --git a/server/src/services/execution-projection-recovery.test.ts b/server/src/services/execution-projection-recovery.test.ts index d03c165b996..c6c6aec0f37 100644 --- a/server/src/services/execution-projection-recovery.test.ts +++ b/server/src/services/execution-projection-recovery.test.ts @@ -15,6 +15,19 @@ describe.skipIf(!support.supported)("persisted stopped-workspace projection", () }, 60_000); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + it("does not count agent directory waits as failed attempts in the batched projection", async () => { + const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Directory wait", issuePrefix: "DW" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Coder", adapterType: "codex_local" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "scheduled_retry", runtimeMode: "legacy", + scheduledRetryReason: "agent_directory_busy", scheduledRetryAttempt: 30, scheduledRetryAt: new Date(Date.now() + 10_000), + contextSnapshot: { retryReason: "agent_directory_busy", + executionRetryAccounting: { version: 1, failureRetries: 0, maxTurnContinuations: 0, agentDirectoryWaits: 30 } } }); + expect((await executionProjectionsForRuns(db, companyId, [runId])).get(runId)).toMatchObject({ + phase: "retry_scheduled", label: "Waiting for agent files", attempt: 1, + }); + }); + it("names failed setup for an unstarted queued run and removes the label after readiness", async () => { const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Setup wait", issuePrefix: "SW" }); diff --git a/server/src/services/execution-projection.ts b/server/src/services/execution-projection.ts index cd51b3c73b1..7140a84c6ec 100644 --- a/server/src/services/execution-projection.ts +++ b/server/src/services/execution-projection.ts @@ -46,7 +46,8 @@ const executionRunColumns = { 'issueId', ${heartbeatRuns.contextSnapshot}->'issueId', 'wakeReason', ${heartbeatRuns.contextSnapshot}->'wakeReason', 'failureRetriesBeforeAiConnectionWait', ${heartbeatRuns.contextSnapshot}->'failureRetriesBeforeAiConnectionWait', - 'failureRetriesBeforeWorkspaceWait', ${heartbeatRuns.contextSnapshot}->'failureRetriesBeforeWorkspaceWait')`, + 'failureRetriesBeforeWorkspaceWait', ${heartbeatRuns.contextSnapshot}->'failureRetriesBeforeWorkspaceWait', + 'executionRetryAccounting', ${heartbeatRuns.contextSnapshot}->'executionRetryAccounting')`, }; type Run = Pick; type Coordinator = typeof nativeRunFinalizations.$inferSelect; @@ -399,6 +400,10 @@ export function projectExecution( projection.nextAction = `Waiting for the agent's provider usage limit to reset; the work starts at ${resetAt}. To continue sooner, restore the provider allowance or change the agent's AI connection, then start a manual run.`; return set("retry_scheduled", "Waiting for usage limit reset"); } + if (run.status === "scheduled_retry" && run.scheduledRetryReason === "agent_directory_busy") { + projection.nextAction = "Waiting for another run of this agent to finish with its files; this task continues automatically."; + return set("retry_scheduled", "Waiting for agent files"); + } if (run.status === "scheduled_retry" && run.scheduledRetryReason === "workspace_busy") { projection.nextAction = "Waiting for the live workspace holder to finish; the scheduled check will revalidate ownership."; return set("retry_scheduled", "Waiting for workspace"); diff --git a/server/src/services/execution-recovery-attempt.test.ts b/server/src/services/execution-recovery-attempt.test.ts index 1940a71358e..128cccbe3a9 100644 --- a/server/src/services/execution-recovery-attempt.test.ts +++ b/server/src/services/execution-recovery-attempt.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "vitest"; -import { accountingForScheduledRetry, executionFailureRetryCount, executionRetryAttemptCount, INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS, INTERRUPTED_PREPARATION_RETRY_REASON } from "./execution-recovery-attempt.js"; +import { accountingForScheduledRetry, AGENT_DIRECTORY_WAIT_RETRY_REASON, executionFailureRetryCount, executionRetryAttemptCount, INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS, INTERRUPTED_PREPARATION_RETRY_REASON } from "./execution-recovery-attempt.js"; +import { AGENT_DIRECTORY_BUSY_MAX_DEFERRALS } from "./heartbeat/agent-directory-busy.js"; import { RUNNER_REVISION_WAIT_RETRY_REASON } from "./runner-revision-mismatch.js"; import { RUNNER_REVISION_WAIT_RETRY_REASON as DISPATCH_RUNNER_REVISION_WAIT_RETRY_REASON } from "../modules/run-dispatch/index.js"; @@ -121,3 +122,24 @@ describe("storage outage waits", () => { expect(executionFailureRetryCount(afterWorkspaceWait)).toBe(1); }); }); + +describe("agent directory waits", () => { + it("keep their count through alternating workspace waits and stop at the bound", () => { + let run: { scheduledRetryReason: string; scheduledRetryAttempt: number; contextSnapshot: Record } = { + scheduledRetryReason: "transient_failure", scheduledRetryAttempt: 1, + contextSnapshot: { executionRetryAccounting: { version: 1, failureRetries: 1, maxTurnContinuations: 0 } } }; + let directoryWaits = 0; + while (executionRetryAttemptCount(run, AGENT_DIRECTORY_WAIT_RETRY_REASON) < AGENT_DIRECTORY_BUSY_MAX_DEFERRALS) { + const attempt = executionRetryAttemptCount(run, AGENT_DIRECTORY_WAIT_RETRY_REASON) + 1; + run = { scheduledRetryReason: AGENT_DIRECTORY_WAIT_RETRY_REASON, scheduledRetryAttempt: attempt, + contextSnapshot: { executionRetryAccounting: accountingForScheduledRetry(run, AGENT_DIRECTORY_WAIT_RETRY_REASON, attempt) } }; + directoryWaits++; + const workspaceAttempt = executionRetryAttemptCount(run, "workspace_busy") + 1; + run = { scheduledRetryReason: "workspace_busy", scheduledRetryAttempt: workspaceAttempt, + contextSnapshot: { executionRetryAccounting: accountingForScheduledRetry(run, "workspace_busy", workspaceAttempt) } }; + expect(executionRetryAttemptCount(run, AGENT_DIRECTORY_WAIT_RETRY_REASON)).toBe(directoryWaits); + } + expect(directoryWaits).toBe(AGENT_DIRECTORY_BUSY_MAX_DEFERRALS); + expect(executionFailureRetryCount(run)).toBe(1); + }); +}); diff --git a/server/src/services/execution-recovery-attempt.ts b/server/src/services/execution-recovery-attempt.ts index bff18d9aa6c..6905c14b486 100644 --- a/server/src/services/execution-recovery-attempt.ts +++ b/server/src/services/execution-recovery-attempt.ts @@ -1,6 +1,7 @@ import { RUNNER_REVISION_WAIT_RETRY_REASON } from "./runner-revision-mismatch.js"; import { STORAGE_UNAVAILABLE_WAIT_RETRY_REASON } from "../storage/storage-unavailable.js"; import { PROVIDER_ATTEMPT_CREDIT_GENERATION_LIMIT } from "./native-runtime/provider-attempt-outcome.js"; +import { AGENT_DIRECTORY_BUSY_RETRY_REASON } from "../modules/run-dispatch/index.js"; /** * A preparation that a restart interrupted before any provider work started @@ -11,10 +12,11 @@ import { PROVIDER_ATTEMPT_CREDIT_GENERATION_LIMIT } from "./native-runtime/provi */ export const INTERRUPTED_PREPARATION_RETRY_REASON = "interrupted_preparation"; export const INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS = PROVIDER_ATTEMPT_CREDIT_GENERATION_LIMIT; +export const AGENT_DIRECTORY_WAIT_RETRY_REASON = AGENT_DIRECTORY_BUSY_RETRY_REASON; /** Retry lanes that keep their own counter and never spend failures. */ const UNCHARGED_RETRY_LANES: ReadonlySet = new Set([ - "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait", RUNNER_REVISION_WAIT_RETRY_REASON, + "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait", AGENT_DIRECTORY_WAIT_RETRY_REASON, RUNNER_REVISION_WAIT_RETRY_REASON, INTERRUPTED_PREPARATION_RETRY_REASON, STORAGE_UNAVAILABLE_WAIT_RETRY_REASON, ]); @@ -37,6 +39,8 @@ export interface ExecutionRetryAccounting { interruptedPreparations?: number; /** Storage-outage waits in this chain, kept across intervening resource waits. Omitted while zero. */ storageWaits?: number; + /** Agent-directory waits in this chain, kept across intervening resource waits. Omitted while zero. */ + agentDirectoryWaits?: number; } function count(value: unknown): number | null { @@ -52,9 +56,11 @@ function savedAccounting(run: RetryRun): ExecutionRetryAccounting | null { if (saved.version !== 1 || failureRetries === null || maxTurnContinuations === null) return null; const interruptedPreparations = count(saved.interruptedPreparations) ?? 0; const storageWaits = count(saved.storageWaits) ?? 0; + const agentDirectoryWaits = count(saved.agentDirectoryWaits) ?? 0; return { version: 1, failureRetries, maxTurnContinuations, ...(interruptedPreparations > 0 ? { interruptedPreparations } : {}), - ...(storageWaits > 0 ? { storageWaits } : {}) }; + ...(storageWaits > 0 ? { storageWaits } : {}), + ...(agentDirectoryWaits > 0 ? { agentDirectoryWaits } : {}) }; } function historicalFailureCount(run: RetryRun): number { @@ -79,6 +85,8 @@ export function executionRetryAccounting(run: RetryRun): ExecutionRetryAccountin reason === INTERRUPTED_PREPARATION_RETRY_REASON ? count(run.scheduledRetryAttempt) ?? 0 : 0); const storageWaits = Math.max(saved?.storageWaits ?? 0, reason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON ? count(run.scheduledRetryAttempt) ?? 0 : 0); + const agentDirectoryWaits = Math.max(saved?.agentDirectoryWaits ?? 0, + reason === AGENT_DIRECTORY_WAIT_RETRY_REASON ? count(run.scheduledRetryAttempt) ?? 0 : 0); return { version: 1, failureRetries: Math.max(saved?.failureRetries ?? 0, saved && nonFailureLane ? 0 : historicalFailureCount(run)), @@ -86,6 +94,7 @@ export function executionRetryAccounting(run: RetryRun): ExecutionRetryAccountin run.scheduledRetryReason === "max_turns_continuation" ? count(run.scheduledRetryAttempt) ?? 0 : 0), ...(interruptedPreparations > 0 ? { interruptedPreparations } : {}), ...(storageWaits > 0 ? { storageWaits } : {}), + ...(agentDirectoryWaits > 0 ? { agentDirectoryWaits } : {}), }; } @@ -97,6 +106,7 @@ export function executionFailureRetryCount(run: RetryRun): number { export function executionRetryAttemptCount(run: RetryRun, reason: string): number { if (reason === INTERRUPTED_PREPARATION_RETRY_REASON) return executionRetryAccounting(run).interruptedPreparations ?? 0; if (reason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON) return executionRetryAccounting(run).storageWaits ?? 0; + if (reason === AGENT_DIRECTORY_WAIT_RETRY_REASON) return executionRetryAccounting(run).agentDirectoryWaits ?? 0; if (UNCHARGED_RETRY_LANES.has(reason)) { return run.scheduledRetryReason === reason ? count(run.scheduledRetryAttempt) ?? 0 : 0; } @@ -109,6 +119,7 @@ export function accountingForScheduledRetry(run: RetryRun, reason: string, attem if (reason === "max_turns_continuation") accounting.maxTurnContinuations = attempt; else if (reason === INTERRUPTED_PREPARATION_RETRY_REASON) accounting.interruptedPreparations = attempt; else if (reason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON) accounting.storageWaits = attempt; + else if (reason === AGENT_DIRECTORY_WAIT_RETRY_REASON) accounting.agentDirectoryWaits = attempt; else if (!UNCHARGED_RETRY_LANES.has(reason)) accounting.failureRetries = attempt; return accounting; } diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 460374426f1..dd6b167655e 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -1,4 +1,5 @@ import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; +import { agentDirectoryPreparationWaitPredicate, isAgentDirectoryPreparationWait } from "./pre-dispatch-execution-evidence.js"; import { assertDurableChatWakeupRequest, assertDurableChatWakeupReceipt, type DurableChatWakeupRequest } from "./durable-chat-wakeup.js"; import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { canSelectFreshWorkspace, retainedIssueWorkspaceForRetry, type ContinuationReadiness } from "./fresh-workspace-continuation.js"; @@ -112,6 +113,7 @@ export async function undeliveredLegacyUserCommentIds( sql`not coalesce(${heartbeatRuns.status} = 'cancelled' and ${heartbeatRuns.errorCode} = 'workspace_busy' and ${heartbeatRuns.resultJson}#>>'{executionRecovery,kind}' = 'workspace_wait' and ${heartbeatRuns.resultJson}#>'{executionRecovery,providerWorkStarted}' = 'false'::jsonb, false)`, + sql`not coalesce(${agentDirectoryPreparationWaitPredicate()}, false)`, or(...commentIds.map(id => or( sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${id}`, sql`${heartbeatRuns.contextSnapshot}->'wakeCommentIds' @> ${JSON.stringify([id])}::jsonb`, @@ -799,7 +801,7 @@ export async function admitExplicitContinuationRetry(input: { )).for("update"); if (!task || task.assigneeAgentId !== agentId || task.executionRunId !== input.parentRunId || ["done", "cancelled"].includes(task.status) || !parent || - !["failed", "timed_out"].includes(parent.status) || !parent.finishedAt || + (!["failed", "timed_out"].includes(parent.status) && !isAgentDirectoryPreparationWait(parent)) || !parent.finishedAt || parent.runtimeMode !== "legacy" || parent.contextSnapshot?.issueId !== issueId || (parent.nativeIssueId !== null && parent.nativeIssueId !== issueId) || adapterExecutionControls.has(parent.id) || diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 7dd4ab69b46..7846c8d53f8 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -43,6 +43,7 @@ import { dotRunnerBroker } from "./dot-runner-broker.js"; import { withRunExecutionTiming, measureRunPhase, measureWorkspaceSyncPhase, isWorkspaceSyncDetailPhase } from "./execution-preparation-timing.js"; import { createHeartbeatPreparationTiming } from "./heartbeat/preparation-timing.js"; import { createHeartbeatInstructionCopyRelease, createHeartbeatInstructionWorkingCopy } from "./heartbeat/instruction-working-copy.js"; +import { AgentDirectoryPreparationDeferral, deferRunForBusyAgentDirectory, prepareWithAgentDirectoryDeferral } from "./heartbeat/agent-directory-busy.js"; import { preparePredecessorRetirementFallback } from "./heartbeat/predecessor-retirement-fallback.js"; import { prepareHeartbeatRuntimeServiceIntents, startHeartbeatRuntimeServices, recordHeartbeatAdapterRuntimeServices } from "./heartbeat/runtime-services.js"; import { bindHeartbeatRunnerHomeMode, prepareHeartbeatExecutionConfiguration } from "./heartbeat/execution-configuration.js"; @@ -1041,6 +1042,7 @@ export function heartbeatService( findExecutionWorkspaceHolder, finalizeAiConnectionBusyDeferral, finalizeWorkspaceBusyDeferral, + finalizeAgentDirectoryBusyDeferral, promoteDueScheduledRetries, nextScheduledRetryAt, retryScheduledRetryNow, @@ -3533,7 +3535,7 @@ export function heartbeatService( }, { measure: timePreparation, prepareInstructions: async nativeLifecyclePolicy => { - runtimeConfig = await preparedInstructions.prepare({ runtimeConfig, nativeRuntimeResolution, hasInstructionFilesystem, nativeLifecyclePolicy }); + runtimeConfig = await prepareWithAgentDirectoryDeferral(() => preparedInstructions.prepare({ runtimeConfig, nativeRuntimeResolution, hasInstructionFilesystem, nativeLifecyclePolicy })); return { runtimeConfig, instructionCopy: preparedInstructions.instructionCopy }; }, }); @@ -3713,6 +3715,18 @@ export function heartbeatService( })); } catch (err) { await persistUsageCaptureFailure?.(); + const agentDirectoryDeferral = await deferRunForBusyAgentDirectory(db, { + run, error: err, leaseOwner: runOptions.nativeLeaseOwner, restartRecovery: runOptions.nativeRestartRecovery, + schedule: (runId, retryAt, mode) => scheduleNativeSessionResumeDispatch(runId, retryAt, mode), + record: retryAt => appendRunEvent(run, { eventType: "lifecycle", stream: "system", level: "info", + message: "Another run of this agent is still using its files; this run continues when it finishes", + payload: { retryReasonCode: "agent_directory_operation_busy", nextAttemptAt: retryAt.toISOString() } }), + }); + if (agentDirectoryDeferral === "deferred") { + nativeSessionResumeScheduled = true; + return; + } + if (agentDirectoryDeferral === "pre_admission") throw err; if (err instanceof NativeStartupCleanupPendingError) { await requestStartupCleanup(err); return; @@ -3917,6 +3931,12 @@ export function heartbeatService( eventMessage: "stale execution continuation cancelled before dispatch", suppressImmediateRecovery: true, }); + } else if (outerErr instanceof AgentDirectoryPreparationDeferral && await finalizeAgentDirectoryBusyDeferral(run, outerErr) + .catch(deferralErr => { + logger.error({ err: deferralErr, runId }, "failed to finalize agent-directory deferral"); + return false; + })) { + logger.info({ runId }, "run deferred while another run of the agent uses its files"); } else if (isWorkspaceBusyDeferral(outerErr)) { // Expected contention on a shared project workspace, not a // failure: park the run as a bounded scheduled retry and leave the diff --git a/server/src/services/heartbeat/agent-directory-busy.ts b/server/src/services/heartbeat/agent-directory-busy.ts new file mode 100644 index 00000000000..67bf108af94 --- /dev/null +++ b/server/src/services/heartbeat/agent-directory-busy.ts @@ -0,0 +1,55 @@ +import type { Db } from "@paperclipai/db"; +import { AgentDirectoryOperationBusyError } from "../agent-directory-working-copies.js"; +import { deferNativeRunForBusyAgentDirectory } from "../native-runtime/native-controller-recovery.js"; +import { AGENT_DIRECTORY_WAIT_RETRY_REASON } from "../execution-recovery-attempt.js"; + +export const AGENT_DIRECTORY_BUSY_RETRY_REASON = AGENT_DIRECTORY_WAIT_RETRY_REASON; +export const AGENT_DIRECTORY_BUSY_RETRY_WAKE_REASON = "agent_directory_busy_retry"; +export const AGENT_DIRECTORY_BUSY_ERROR_CODE = "agent_directory_busy"; +export const AGENT_DIRECTORY_BUSY_MAX_DEFERRALS = 30; +export const AGENT_DIRECTORY_BUSY_RETRY_MIN_MS = 5_000; +export const AGENT_DIRECTORY_BUSY_RETRY_JITTER_MS = 10_000; + +export class AgentDirectoryPreparationDeferral extends Error { + readonly code = AGENT_DIRECTORY_BUSY_ERROR_CODE; + constructor(cause: AgentDirectoryOperationBusyError) { + super("Another run of this agent is still using its files", { cause }); + this.name = "AgentDirectoryPreparationDeferral"; + } +} + +export async function prepareWithAgentDirectoryDeferral(prepare: () => Promise): Promise { + try { + return await prepare(); + } catch (error) { + if (error instanceof AgentDirectoryOperationBusyError) throw new AgentDirectoryPreparationDeferral(error); + throw error; + } +} + +export function agentDirectoryBusyRetryDelayMs(random: () => number = Math.random): number { + return AGENT_DIRECTORY_BUSY_RETRY_MIN_MS + Math.floor(Math.min(Math.max(random(), 0), 0.999) * AGENT_DIRECTORY_BUSY_RETRY_JITTER_MS); +} + +export type AgentDirectoryDeferralOutcome = "not_applicable" | "deferred" | "pre_admission" | "refused"; + +export async function deferRunForBusyAgentDirectory(db: Db, input: { + run: { id: string; companyId: string }; + error: unknown; + leaseOwner?: string; + restartRecovery?: { leaseOwner: string }; + schedule: (runId: string, retryAt: Date, mode?: "restart") => void; + record: (retryAt: Date) => Promise; + random?: () => number; +}): Promise { + if (!(input.error instanceof AgentDirectoryPreparationDeferral)) return "not_applicable"; + const retryAt = new Date(Date.now() + agentDirectoryBusyRetryDelayMs(input.random)); + const outcome = await deferNativeRunForBusyAgentDirectory(db, { + companyId: input.run.companyId, runId: input.run.id, leaseOwner: input.leaseOwner ?? input.restartRecovery?.leaseOwner, + retryAt, maxDeferrals: AGENT_DIRECTORY_BUSY_MAX_DEFERRALS, + }); + if (outcome !== "deferred") return outcome; + await input.record(retryAt).catch(() => undefined); + input.schedule(input.run.id, retryAt, input.restartRecovery ? "restart" : undefined); + return "deferred"; +} diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 6bc3ad55266..a3e6ac42970 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -1,4 +1,5 @@ import { runOwnerFenceHeldSql } from "../remote-workspace-writer-hold.js"; +import { AGENT_DIRECTORY_BUSY_ERROR_CODE, AGENT_DIRECTORY_BUSY_MAX_DEFERRALS, AGENT_DIRECTORY_BUSY_RETRY_REASON, AGENT_DIRECTORY_BUSY_RETRY_WAKE_REASON, agentDirectoryBusyRetryDelayMs, type AgentDirectoryPreparationDeferral } from "./agent-directory-busy.js"; import { isInterruptedNativePreparation } from "../cancelled-native-startup.js"; import { isPreDispatchExecutionWaitVerified } from "../pre-dispatch-execution-wait.js"; import { notifyAttentionRetentionWork } from "../retention-work-notifications.js"; @@ -78,6 +79,7 @@ import { type WorkspaceBusyWait, isNonAssigneeWorkspaceBusyRetry, isResolvedInteractionContinuationWakeContext, + isPreparationWaitRetryReason, } from "../../modules/run-dispatch/index.js"; import type { AppendHeartbeatRunEventInput } from "../heartbeat-run-events.js"; @@ -469,6 +471,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen wakeReason?: string; maxAttempts?: number; delayMs?: number; + preserveWakeContext?: boolean; }, ) { const preparation = parseObject(run.resultJson?.executionRecovery); @@ -696,8 +699,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen } const taskKey = deriveTaskKeyWithHeartbeatFallback(contextSnapshot, null); const sessionBefore = await resolveSessionBeforeForWakeup(agent, taskKey); - const interactionRunnerWait = (retryReason === RUNNER_REVISION_WAIT_RETRY_REASON - || retryReason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON) + const interactionRunnerWait = isPreparationWaitRetryReason(retryReason) && isResolvedInteractionContinuationWakeContext(run.contextSnapshot); const interactionContinuationPayload = retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON || interactionRunnerWait @@ -728,8 +730,9 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen ...contextSnapshot, executionRetryAccounting: accountingForScheduledRetry(run, retryReason, schedule.attempt), retryOfRunId: run.id, - wakeReason, - retryReason, + ...(opts?.preserveWakeContext + ? { preparationWait: { kind: retryReason, attempt: schedule.attempt, nextAttemptAt: schedule.dueAt.toISOString() } } + : { wakeReason, retryReason }), ...(interactionRunnerWait ? { mutation: "interaction" } : {}), ...(retryReason === WORKSPACE_BUSY_RETRY_REASON ? { @@ -1087,8 +1090,9 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen const scheduledRunId = randomUUID(); if (contextSnapshot.explicitUserContinuation) { - const continuation = issueId && (retryReason === "transient_failure" || retryReason === RUNNER_REVISION_WAIT_RETRY_REASON - || retryReason === STORAGE_UNAVAILABLE_WAIT_RETRY_REASON) ? await admitExplicitContinuationRetry({ + const continuation = issueId && (retryReason === "transient_failure" || isPreparationWaitRetryReason(retryReason) + || retryReason === AGENT_DIRECTORY_BUSY_RETRY_REASON) + ? await admitExplicitContinuationRetry({ db: tx as unknown as Db, companyId: run.companyId, issueId, agentId: run.agentId, parentRunId: run.id, successorRunId: scheduledRunId, now, }) : null; @@ -1111,7 +1115,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen agentId: run.agentId, source: "automation", triggerDetail: "system", - reason: wakeReason, + reason: opts?.preserveWakeContext ? readNonEmptyString(contextSnapshot.wakeReason) ?? wakeReason : wakeReason, payload: withRecoveryContext( { ...(issueId ? { issueId } : {}), @@ -1594,6 +1598,60 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen } } + async function finalizeAgentDirectoryBusyDeferral( + run: typeof heartbeatRuns.$inferSelect, + deferral: AgentDirectoryPreparationDeferral, + ): Promise { + if (executionRetryAttemptCount(run, AGENT_DIRECTORY_BUSY_RETRY_REASON) >= AGENT_DIRECTORY_BUSY_MAX_DEFERRALS) return false; + const now = new Date(); + const runContext = parseObject(run.contextSnapshot); + const admittedNonAssignee = parseObject(run.runnerProfileJson).aiConnectionNonAssigneeCommentWake === true + || isNonAssigneeWorkspaceBusyRetry(run.scheduledRetryReason, runContext); + const cancelled = await setRunStatusIfRunning(run.id, "cancelled", { + error: deferral.message, errorCode: AGENT_DIRECTORY_BUSY_ERROR_CODE, finishedAt: now, + resultJson: { + executionRecovery: { kind: "agent_directory_wait", providerWorkStarted: false }, + cancellation: { + source: "control_plane", + expected: true, + initiator: { type: "system" }, + reason: "Waiting for another run of this agent to finish with its files", + recordedAt: now.toISOString(), + }, + }, + contextSnapshot: { ...runContext, preparationWaitDeferredWhileAssignee: !admittedNonAssignee }, + }); + if (!cancelled.updated) return true; + await setWakeupStatus(run.wakeupRequestId, "cancelled", { finishedAt: now, error: deferral.message }).catch(() => undefined); + const cancelledRun = cancelled.run ?? await getRun(run.id); + const agent = await getAgent(run.agentId); + let scheduled = false; + try { + if (cancelledRun && agent) { + const retry = await scheduleBoundedRetryForRun(cancelledRun, agent, { + now, retryReason: AGENT_DIRECTORY_BUSY_RETRY_REASON, wakeReason: AGENT_DIRECTORY_BUSY_RETRY_WAKE_REASON, + maxAttempts: AGENT_DIRECTORY_BUSY_MAX_DEFERRALS, + delayMs: agentDirectoryBusyRetryDelayMs(), + preserveWakeContext: true, + }); + scheduled = retry.outcome === "scheduled"; + await appendRunEvent(cancelledRun, { + eventType: "lifecycle", stream: "system", level: "info", + message: scheduled ? "Another run of this agent is still using its files. This task will continue automatically." + : "Another run of this agent is still using its files, and this task could not be rescheduled.", + payload: { retryScheduled: scheduled }, + }); + } + } finally { + try { + if (cancelledRun && !scheduled) await releaseIssueExecutionAndPromote(cancelledRun); + } finally { + await finalizeAgentStatus(run.agentId, "cancelled", null, { wasFirstHeartbeat: timerClaimWasFirstHeartbeat(run) }); + } + } + return true; + } + // The wait applies only while the task keeps the workspace binding it had // when it deferred; run dispatch compares the same binding before parking. async function readWorkspaceBusyWaitForRetry( @@ -2057,6 +2115,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen findExecutionWorkspaceHolder, finalizeAiConnectionBusyDeferral, finalizeWorkspaceBusyDeferral, + finalizeAgentDirectoryBusyDeferral, promoteDueScheduledRetries, nextScheduledRetryAt, retryScheduledRetryNow, diff --git a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts index 9129ea6374f..ebdd6c7547f 100644 --- a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts @@ -15,9 +15,11 @@ import { startRecoveryPostgres, recoveryPostgresSupport } from "../../__tests__/ import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js"; import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; -import { assertNativeRetainedSandboxBinding, createNativeControllerRecovery, deferNativeControllerRecovery, importNativeControllerRecovery, NativeControllerRecoveryWaitingError, NATIVE_RETAINED_OWNER_WAIT, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; +import { assertNativeRetainedSandboxBinding, createNativeControllerRecovery, deferNativeControllerRecovery, deferNativeRunForBusyAgentDirectory, NATIVE_AGENT_DIRECTORY_BUSY_WAIT, importNativeControllerRecovery, NativeControllerRecoveryWaitingError, NATIVE_RETAINED_OWNER_WAIT, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; import { claimNativeRestartRecoveries } from "./native-restart-recovery.js"; +import { AgentDirectoryPreparationDeferral, deferRunForBusyAgentDirectory, prepareWithAgentDirectoryDeferral } from "../heartbeat/agent-directory-busy.js"; +import { AgentDirectoryOperationBusyError } from "../agent-directory-working-copies.js"; import { readNativeControllerRecoveryInventory } from "./native-controller-recovery-inventory.js"; import { inspectRemoteWarmSessionRecovery } from "./remote-warm-session-recovery.js"; @@ -272,6 +274,83 @@ describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable d } }); + async function seedBusyDirectoryRun(input: { checkpoint?: boolean; resultJson?: Record } = {}) { + const issueId = randomUUID(), runId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Busy agent directory", assigneeAgentId: agentId, status: "in_progress" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, status: "running", runtimeMode: "native", + resultJson: input.resultJson ?? null, + runnerProfileJson: { nativeExecutionInput: { binding: { companyId, agentId, issueId, runId } }, + ...(input.checkpoint ? { sessionCheckpoint: { backendKind: "mock", sessionId: randomUUID(), providerSessionId: "provider-session", + identity: { companyId, agentId, issueId, runId } } } : {}) } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "retryable_failure", attempt: 3, + failureDetail: { recoveryMode: "exact_checkpoint_resume", providerSessionEstablished: true, providerEventsExist: true } }); + const read = async () => (await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0]!; + const readRun = async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]!; + return { issueId, runId, read, readRun }; + } + + it("defers a resumed run whose agent directory is busy across repeated claims and stops at the bound", async () => { + const { runId, read, readRun } = await seedBusyDirectoryRun({ checkpoint: true }); + let now = new Date(); + for (let cycle = 1; cycle <= 3; cycle++) { + const [claim] = await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [runId], now }); + expect(claim).toBeDefined(); + expect((await read()).failureDetail?.agentDirectoryBusyDeferrals).toBeUndefined(); + const retryAt = new Date(now.getTime() + 10_000); + expect(await deferNativeRunForBusyAgentDirectory(db, { companyId, runId, leaseOwner: claim!.leaseOwner, retryAt, maxDeferrals: 3 })).toBe("deferred"); + expect(await read()).toMatchObject({ phase: "retryable_failure", attempt: 3, leaseOwner: null, nextAttemptAt: retryAt, + failureCode: NATIVE_AGENT_DIRECTORY_BUSY_WAIT, failureDetail: { recoveryMode: "same_session_resume" } }); + expect(await readRun()).toMatchObject({ status: "running", errorCode: null, contextSnapshot: { agentDirectoryBusyDeferrals: cycle } }); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [runId], now })).toEqual([]); + now = retryAt; + } + const [claim] = await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [runId], now }); + expect(await deferNativeRunForBusyAgentDirectory(db, { companyId, runId, leaseOwner: claim!.leaseOwner, + retryAt: new Date(now.getTime() + 10_000), maxDeferrals: 3 })).toBe("refused"); + expect(await read()).toMatchObject({ phase: "observed", leaseOwner: claim!.leaseOwner }); + }); + + it("leaves a busy-directory run to normal settlement after Stop, a lost lease or a moved task", async () => { + const retryAt = new Date(Date.now() + 10_000); + const stopped = await seedBusyDirectoryRun({ checkpoint: true }); + const [stoppedClaim] = await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [stopped.runId] }); + await db.update(heartbeatRuns).set({ resultJson: { cancellation: { source: "operator", expected: true, initiator: { type: "user", userId: "board" }, + reason: "Stopped", recordedAt: new Date().toISOString() } } }).where(eq(heartbeatRuns.id, stopped.runId)); + expect(await deferNativeRunForBusyAgentDirectory(db, { companyId, runId: stopped.runId, leaseOwner: stoppedClaim!.leaseOwner, retryAt, maxDeferrals: 30 })).toBe("refused"); + expect((await stopped.read()).nextAttemptAt).toBeNull(); + const foreign = await seedBusyDirectoryRun({ checkpoint: true }); + await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [foreign.runId] }); + expect(await deferNativeRunForBusyAgentDirectory(db, { companyId, runId: foreign.runId, leaseOwner: "another-controller", retryAt, maxDeferrals: 30 })).toBe("refused"); + const moved = await seedBusyDirectoryRun({ checkpoint: true }); + const [movedClaim] = await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [moved.runId] }); + await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, moved.issueId)); + expect(await deferNativeRunForBusyAgentDirectory(db, { companyId, runId: moved.runId, leaseOwner: movedClaim!.leaseOwner, retryAt, maxDeferrals: 30 })).toBe("refused"); + const fresh = randomUUID(); + await db.insert(heartbeatRuns).values({ id: fresh, companyId, agentId, status: "running", runtimeMode: "native" }); + expect(await deferNativeRunForBusyAgentDirectory(db, { companyId, runId: fresh, retryAt, maxDeferrals: 30 })).toBe("pre_admission"); + }); + + it("defers only a busy agent directory met during preparation, never a busy collection after a turn", async () => { + const { runId, read } = await seedBusyDirectoryRun({ checkpoint: true }); + const [claim] = await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [runId] }); + const schedule = vi.fn(), record = vi.fn(async () => undefined); + const run = { id: runId, companyId }; + expect(await deferRunForBusyAgentDirectory(db, { run, error: new AgentDirectoryOperationBusyError(), leaseOwner: claim!.leaseOwner, schedule, record })).toBe("not_applicable"); + expect(await deferRunForBusyAgentDirectory(db, { run, error: new Error("unrelated"), leaseOwner: claim!.leaseOwner, schedule, record })).toBe("not_applicable"); + expect(schedule).not.toHaveBeenCalled(); + expect((await read()).phase).toBe("observed"); + const deferral = await prepareWithAgentDirectoryDeferral(async () => { throw new AgentDirectoryOperationBusyError(); }).catch(error => error); + expect(deferral).toBeInstanceOf(AgentDirectoryPreparationDeferral); + expect(deferral.message).toBe("Another run of this agent is still using its files"); + const started = Date.now(); + expect(await deferRunForBusyAgentDirectory(db, { run, error: deferral, leaseOwner: claim!.leaseOwner, schedule, record, random: () => 0.5 })).toBe("deferred"); + const [, retryAt] = schedule.mock.calls[0]! as [string, Date]; + expect(record).toHaveBeenCalledWith(retryAt); + expect(retryAt.getTime() - started).toBeGreaterThanOrEqual(10_000); + expect(retryAt.getTime() - started).toBeLessThan(11_000); + }); + it("keeps an unstarted follow-up on the same run until the retained owner's deadline", async () => { const fixture = await seed(), runId = randomUUID(); const scope = { ...fixture.scope, runId }; diff --git a/server/src/services/native-runtime/native-controller-recovery.ts b/server/src/services/native-runtime/native-controller-recovery.ts index 91cad55c06f..859c7c837b7 100644 --- a/server/src/services/native-runtime/native-controller-recovery.ts +++ b/server/src/services/native-runtime/native-controller-recovery.ts @@ -122,6 +122,57 @@ export async function deferNativeControllerRecovery(db: Db, input: { }); } +export const NATIVE_AGENT_DIRECTORY_BUSY_WAIT = "agent_directory_operation_busy"; + +export async function deferNativeRunForBusyAgentDirectory(db: Db, input: { + companyId: string; runId: string; leaseOwner?: string; retryAt: Date; maxDeferrals: number; +}): Promise<"deferred" | "pre_admission" | "refused"> { + return db.transaction(async tx => { + const [current] = await tx.select({ issueId: heartbeatRuns.nativeIssueId }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))); + const [existing] = await tx.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId))); + if (!current?.issueId || !existing) return "pre_admission"; + const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues) + .where(and(eq(issues.companyId, input.companyId), eq(issues.id, current.issueId))).for("update"); + const [owner] = await tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.issueId, current.issueId), + eq(nativeRunFinalizations.runId, input.runId))).for("update"); + const [run] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.nativeIssueId, current.issueId))).for("update"); + const cancellation = record(run?.resultJson?.nativeCancellation); + const stop = readRunCancellation(run?.resultJson); + const context = record(run?.contextSnapshot); + const deferrals = typeof context.agentDirectoryBusyDeferrals === "number" ? context.agentDirectoryBusyDeferrals : 0; + const [clock] = await tx.execute<{ now: Date }>(sql`select clock_timestamp() as now`); + if (!run || run.runtimeMode !== "native" || run.status !== "running" || task?.executionRunId !== input.runId + || !owner || owner.resultId || owner.phase !== "observed" + || owner.leaseOwner !== (input.leaseOwner ?? null) + || (input.leaseOwner && (!owner.leaseExpiresAt || owner.leaseExpiresAt <= new Date(clock.now))) + || run.resultJson?.startupCancellation || stop + || (["run", "issue"].includes(String(cancellation.scope)) && ["pending", "acknowledged"].includes(String(cancellation.dispatchState))) + || deferrals >= input.maxDeferrals) return "refused"; + await notifyNativeFinalizationWork(tx, { companyId: input.companyId, issueId: current.issueId, runId: input.runId }); + await tx.update(heartbeatRuns).set({ contextSnapshot: { ...context, agentDirectoryBusyDeferrals: deferrals + 1 }, updatedAt: new Date() }) + .where(eq(heartbeatRuns.id, input.runId)); + const detail = record(owner.failureDetail); + await tx.update(nativeRunFinalizations).set({ + phase: "retryable_failure", leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, + nextAttemptAt: input.retryAt, recoveryState: "resuming_session", failureCode: NATIVE_AGENT_DIRECTORY_BUSY_WAIT, + failureDetail: { + recoveryMode: record(run.runnerProfileJson).sessionCheckpoint ? "same_session_resume" : "bootstrap_retry", + providerSessionEstablished: false, providerEventsExist: false, + ...detail, + originalFailureCode: detail.originalFailureCode ?? NATIVE_AGENT_DIRECTORY_BUSY_WAIT, + message: "Waiting for another run of this agent to finish with its files.", + }, + updatedAt: new Date(), + }).where(eq(nativeRunFinalizations.runId, input.runId)); + return "deferred"; + }); +} + /** The retained Runner keeps its state under the directory its own lease * recorded. A workspace can move to a new allocation of the SAME sandbox: its * prior epoch was abandoned, so the last save was restored into a fresh root diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index f58fa27e351..b67a49363e6 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -2,6 +2,8 @@ import { currentControllerProcess, currentProcessSpaceHash } from "../controller import { renewNativeSessionExecutionLease, startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "./native-execution-lease.js"; import { enableRunFailureReporting } from "../../__tests__/helpers/run-failure-reporting.js"; import { NativeControllerRecoveryWaitingError } from "./native-controller-recovery.js"; +import { deferRunForBusyAgentDirectory, prepareWithAgentDirectoryDeferral } from "../heartbeat/agent-directory-busy.js"; +import { AgentDirectoryOperationBusyError } from "../agent-directory-working-copies.js"; import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; import { hasNativeLocalProcessStop } from "../native-local-process-stop.js"; import { createHash, randomUUID } from "node:crypto"; @@ -1570,6 +1572,50 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); expect(run?.processPid).toBe(remotePid); }); + it("keeps a restart recovery's attempt credit while the agent directory is busy, then resumes at attempt 3", async () => { + const fixture = await seedRun("remote-busy-directory"); + await fixture.db.update(heartbeatRuns).set({ + processPid: null, + processStartedAt: new Date("2020-01-01T00:00:00Z"), + runnerProfileJson: { + nativeWorkspaceSync: { + schema: "paperclip.native-workspace-sync/v1", state: "prepared", + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), + workspaceId: "workspace", leaseId: "lease", providerLeaseId: "sandbox", + remoteCwd: "/workspace", + }, + sessionCheckpoint: { + identity: { companyId, agentId, issueId: fixture.issueId, runId: fixture.runId, + sessionId: fixture.native.normalizedSessionId }, + providerSessionId: "provider-session", + process: { providerPid: process.pid }, + }, + }, + }).where(eq(heartbeatRuns.id, fixture.runId)); + await fixture.db.update(nativeRunFinalizations).set({ attempt: 3 }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "graceful", runIds: [fixture.runId] }); + expect(claim).toMatchObject({ kind: "reattach_remote_runner", providerAttempt: 3 }); + if (claim?.kind !== "reattach_remote_runner") throw new Error("restart recovery did not reattach"); + const schedule = vi.fn(); + const deferral = await prepareWithAgentDirectoryDeferral(async () => { throw new AgentDirectoryOperationBusyError(); }).catch(error => error); + expect(await deferRunForBusyAgentDirectory(fixture.db, { run: { id: fixture.runId, companyId }, error: deferral, + restartRecovery: claim, schedule, record: async () => undefined, random: () => 0 })).toBe("deferred"); + const [, retryAt, mode] = schedule.mock.calls[0]! as [string, Date, string | undefined]; + expect(mode).toBe("restart"); + expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, restartKind: "graceful", + runIds: [fixture.runId], now: new Date(retryAt.getTime() - 1_000) })).toEqual([]); + const [resumed] = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "graceful", runIds: [fixture.runId], now: retryAt }); + expect(resumed).toMatchObject({ kind: "reattach_remote_runner", runId: fixture.runId, providerAttempt: 3 }); + if (resumed?.kind !== "reattach_remote_runner") throw new Error("deferred restart recovery did not reattach"); + expect(nextNativeProviderAttempt(3, resumed.kind)).toBe(3); + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(coordinator).toMatchObject({ attempt: 3, phase: "observed", leaseOwner: resumed.leaseOwner }); + const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(run).toMatchObject({ status: "running", errorCode: null }); + }); + it.each(["running", "interrupted-stop", "interrupted-no-intent", "interrupted-foreign-intent", "interrupted-missing-audit", "interrupted-foreign-owner", "interrupted-terminal", "interrupted-no-checkpoint"])("gives only eligible recovery a full lease after an old batch observation (%s)", async mode => { const fixture = await seedRun(`LATE-BATCH-CLAIM-${mode}`); const binding = { companyId, issueId: fixture.issueId, runId: fixture.runId }; diff --git a/server/src/services/pre-dispatch-execution-evidence.ts b/server/src/services/pre-dispatch-execution-evidence.ts index a6f712dc950..6854ffea3e6 100644 --- a/server/src/services/pre-dispatch-execution-evidence.ts +++ b/server/src/services/pre-dispatch-execution-evidence.ts @@ -8,10 +8,29 @@ export function isPreDispatchResourceWait(run: Pick | undefined; return run.runtimeMode === "legacy" && run.status === "cancelled" && evidence?.providerWorkStarted === false && (run.errorCode === "workspace_busy" && evidence.kind === "workspace_wait" || + run.errorCode === "agent_directory_busy" && evidence.kind === "agent_directory_wait" || ["ai_connection_busy", "ai_connection_pool_exhausted", "provider_limit_wait"].includes(run.errorCode ?? "") && evidence.kind === "ai_connection_wait"); } +export function isAgentDirectoryPreparationWait(run: Pick): boolean { + const evidence = run.resultJson?.executionRecovery as Record | undefined; + const cancellation = run.resultJson?.cancellation as Record | undefined; + return run.runtimeMode === "legacy" && run.status === "cancelled" && run.errorCode === "agent_directory_busy" + && evidence?.kind === "agent_directory_wait" && evidence.providerWorkStarted === false + && cancellation?.source === "control_plane" && cancellation.expected === true; +} + +export function agentDirectoryPreparationWaitPredicate() { + return sql`(${heartbeatRuns.runtimeMode} = 'legacy' and ${heartbeatRuns.status} = 'cancelled' + and ${heartbeatRuns.errorCode} = 'agent_directory_busy' + and ${heartbeatRuns.resultJson}#>>'{executionRecovery,kind}' = 'agent_directory_wait' + and ${heartbeatRuns.resultJson}#>'{executionRecovery,providerWorkStarted}' = 'false'::jsonb + and ${heartbeatRuns.resultJson}#>>'{cancellation,source}' = 'control_plane' + and ${heartbeatRuns.resultJson}#>'{cancellation,expected}' = 'true'::jsonb)`; +} + export function executionStartedEventPredicate() { return or(isNotNull(heartbeatRunEvents.sourceEventId), inArray(heartbeatRunEvents.eventType, ["adapter.invoke", PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, "harness.ready", "session.started", diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index 99bab822e93..1e8c2f59f15 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -1,4 +1,5 @@ import { STORAGE_UNAVAILABLE_WAIT_RETRY_REASON } from "../storage/storage-unavailable.js"; +import { AGENT_DIRECTORY_WAIT_RETRY_REASON } from "./execution-recovery-attempt.js"; import { boundSetupCommandFailure, readSetupCommandFailure } from "./workspace-setup-command-failure.js"; import { and, desc, eq, inArray, isNull, sql, type SQL } from "drizzle-orm"; import { unidentifiedLocalLaunchRefusals } from "./native-local-process-stop.js"; @@ -30,7 +31,8 @@ import { AI_CONNECTION_BUSY_RETRY_REASON, AI_CONNECTION_POOL_WAIT_RETRY_REASON, // attempt. Their dispatcher re-checks the resource, so the run can stay // scheduled past its retry time. Retry cannot free the resource either. const resourceWaitRetryReasons = new Set([WORKSPACE_BUSY_RETRY_REASON, AI_CONNECTION_BUSY_RETRY_REASON, - AI_CONNECTION_POOL_WAIT_RETRY_REASON, RUNNER_REVISION_WAIT_RETRY_REASON, STORAGE_UNAVAILABLE_WAIT_RETRY_REASON]); + AI_CONNECTION_POOL_WAIT_RETRY_REASON, RUNNER_REVISION_WAIT_RETRY_REASON, STORAGE_UNAVAILABLE_WAIT_RETRY_REASON, + AGENT_DIRECTORY_WAIT_RETRY_REASON]); // A due same-run resume dispatches within seconds; past this, offer Retry. const NATIVE_RESUME_DISPATCH_GRACE_MS = 2 * 60_000; diff --git a/ui/src/components/TaskChatThread.tsx b/ui/src/components/TaskChatThread.tsx index 1a73d555d15..03b9728a210 100644 --- a/ui/src/components/TaskChatThread.tsx +++ b/ui/src/components/TaskChatThread.tsx @@ -1689,7 +1689,7 @@ export function TaskChatThread(props: TaskChatThreadProps) { const historical = statusRelevance.isHistoricalRun(source.id); // A workspace admission attempt never started provider work. Its live // successor owns the waiting indicator; retain this attempt in the run log. - if (source.status === "cancelled" && meta?.errorCode === "workspace_busy") { + if (source.status === "cancelled" && (meta?.errorCode === "workspace_busy" || meta?.errorCode === "agent_directory_busy")) { settledRunIds.add(source.id); continue; } diff --git a/ui/src/context/LiveUpdatesProvider.tsx b/ui/src/context/LiveUpdatesProvider.tsx index 782f0d72c0c..5efd825d746 100644 --- a/ui/src/context/LiveUpdatesProvider.tsx +++ b/ui/src/context/LiveUpdatesProvider.tsx @@ -1326,7 +1326,7 @@ function buildRunStatusToast( // feedback; the terminal event must not announce a cancelled/failed run. if (errorCode === "operator_interrupted") return null; // Workspace contention is ordinary scheduling, not a failed user action. - if (errorCode === "workspace_busy") return null; + if (errorCode === "workspace_busy" || errorCode === "agent_directory_busy") return null; const contextSource = readString(payload.contextSource); const triggerDetail = readString(payload.triggerDetail); const name = nameOf(agentId) ?? "Agent";