fix(server): defer a run that meets a busy agent directory instead of failing it (#20829)

## 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 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed 2026-11-11 13:56:45 -06:00
1 parent a56b648602
commit 34f71f7d57
22 files changed
+789 -23

No files matched your search

+9
View File
@@ -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
+22
View File
@@ -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
@@ -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<void>) }));
vi.mock("../services/agent-instruction-working-copies.js", async importOriginal => {
const actual = await importOriginal<typeof import("../services/agent-instruction-working-copies.js")>();
return {
...actual,
agentInstructionWorkingCopyService: (...args: Parameters<typeof actual.agentInstructionWorkingCopyService>) => {
const service = actual.agentInstructionWorkingCopyService(...args);
return {
...service,
prepare: async (input: Parameters<typeof service.prepare>[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<typeof createDb>;
let heartbeat!: ReturnType<typeof heartbeatService>;
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
const executedRunIds: string[] = [];
const executedContexts = new Map<string, Record<string, unknown>>();
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<string, unknown>);
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([]);
});
});
@@ -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<string, unknown>).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),
@@ -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<string> = 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 ||
+3
View File
@@ -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,
@@ -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<string, PreparedAdapterExecutionTargetRuntime>();
const key = (row: Pick<Copy, "companyId" | "runId">) => `${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);
}));
}
@@ -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" });
+6 -1
View File
@@ -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<typeof heartbeatRuns.$inferSelect, keyof typeof executionRunColumns>;
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");
@@ -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<string, unknown> } = {
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);
});
});
@@ -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<string> = 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;
}
@@ -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) ||
+21 -1
View File
@@ -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
@@ -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<T>(prepare: () => Promise<T>): Promise<T> {
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<void>;
random?: () => number;
}): Promise<AgentDirectoryDeferralOutcome> {
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";
}
+66 -7
View File
@@ -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<boolean> {
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,
@@ -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<string, unknown> } = {}) {
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 };
@@ -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
@@ -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 };
@@ -8,10 +8,29 @@ export function isPreDispatchResourceWait(run: Pick<typeof heartbeatRuns.$inferS
const evidence = run.resultJson?.executionRecovery as Record<string, unknown> | 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<typeof heartbeatRuns.$inferSelect,
"runtimeMode" | "status" | "errorCode" | "resultJson">): boolean {
const evidence = run.resultJson?.executionRecovery as Record<string, unknown> | undefined;
const cancellation = run.resultJson?.cancellation as Record<string, unknown> | 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",
@@ -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<string>([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;
+1 -1
View File
@@ -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;
}
+1 -1
View File
@@ -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";