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";