diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index d2472c1d519..4d7260d3cdc 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -612,6 +612,17 @@ failing setup. The save's retry budget bounds that wait. A save that exhausts its retries is handed to the board as a named recovery action. After that hand-off, the cutoff no longer advances. +Automatic wakes do not run into a hold that preparation would refuse. A parent or +dependent woken because a child or blocker stalled on the board +(`issue_blocker_stalled`) is not woken again when its assignee's last run on the +task failed setup on its bound workspace's hold and that hold is still past the +cutoff, or held by a writer that cannot be named: another run could only fail +the same way. The wake is delivered once the hold settles or the task changes +assignee, and the sweep checks again at most hourly. Failed runs that carry a +stall, including resource-wait retries of its wake, are retried after ten +minutes, then twenty, and not after the third failed run. The stall stays +listed on the waiting task, and a new stall wakes it again. + An instance admission drain keeps accepted new turns queued and displays **Waiting for maintenance**. Ending the audited drain requests admission for these existing runs immediately. The shared delivery coordinator owns the drain's diff --git a/server/src/__tests__/issue-blocker-stall-wakeups.test.ts b/server/src/__tests__/issue-blocker-stall-wakeups.test.ts index d77227daabe..480cc52a3d6 100644 --- a/server/src/__tests__/issue-blocker-stall-wakeups.test.ts +++ b/server/src/__tests__/issue-blocker-stall-wakeups.test.ts @@ -1,8 +1,9 @@ import { randomUUID } from "node:crypto"; -import { eq, sql } from "drizzle-orm"; +import { eq, inArray, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { - activityLog, agents, agentWakeupRequests, companies, createDb, heartbeatRuns, issueRecoveryActions, issueRelations, issues, type Db, + activityLog, agents, agentWakeupRequests, companies, createDb, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issueRelations, + issues, type Db, } from "@paperclipai/db"; import { ISSUE_BLOCKER_STALLED_WAKE_REASON } from "@paperclipai/shared"; import { recoveryService } from "../services/recovery/service.js"; @@ -241,12 +242,32 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu contextSnapshot: { ...calls[0]!.input.contextSnapshot, wakeReason: "workspace_busy_retry", retryOfRunId: randomUUID() } }).returning(); await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); expect(calls).toHaveLength(1); - // A retry chain that was itself cancelled, or failed before the agent ran, no longer counts. - await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, retry.id)); + // A retry chain that failed no longer carries the stall. Its failure counts + // as a failed run of this stall, so the stall is retried after the wait. + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, retry.id)); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(1); + await db.update(heartbeatRuns).set({ finishedAt: new Date(Date.now() - 11 * 60_000) }).where(eq(heartbeatRuns.id, retry.id)); await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); expect(calls).toHaveLength(2); }); + it("counts failed resource-wait retries toward the stall's failed-run limit", async () => { + const f = await fixture(); + const { calls, enqueueWakeup } = wakeRecorder(); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + // Each keyed wake is cancelled by a workspace wait; its keyless retry then fails. + for (let attempt = 0; attempt < 3; attempt += 1) { + await db.update(agentWakeupRequests).set({ status: "cancelled", requestedAt: new Date(Date.now() - 24 * 60 * 60_000) }) + .where(eq(agentWakeupRequests.companyId, f.companyId)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.parentAgent, invocationSource: "automation", + status: "failed", finishedAt: new Date(Date.now() - 24 * 60 * 60_000), + contextSnapshot: { ...calls[0]!.input.contextSnapshot, wakeReason: "workspace_busy_retry", retryOfRunId: randomUUID() } }); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + } + expect(calls).toHaveLength(3); + }); + it("reaches a waiter through its own sweep when the stalled blocker is unassigned", async () => { const f = await fixture(); await db.update(issues).set({ parentId: null, assigneeAgentId: null }).where(eq(issues.id, f.child.id)); @@ -330,4 +351,128 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu for (const f of [agentOwned, doneParent, unassigned, hidden]) await service.reconcileActiveRecoveryActions({ companyId: f.companyId }); expect(calls.filter(call => call.input.reason === ISSUE_BLOCKER_STALLED_WAKE_REASON)).toEqual([]); }); + + // Staging, 2026-11-09: a legacy child (created before subtasks got + // their own workspace) shares its parent's remote workspace. The child's save + // failed and its automatic recovery was exhausted, so its ended run still + // holds the shared workspace, days past the holder cutoff. Each stall wake of + // the parent created a run that could only fail setup ("The workspace still + // has an unfinished writer or save operation."), twice every ten minutes. + async function legacySharedWorkspaceFixture() { + const f = await fixture(); + const endedAt = new Date(Date.now() - 4 * 24 * 60 * 60_000); + const [writer] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.childAgent, + runtimeMode: "native", nativeIssueId: f.child.id, status: "failed", startedAt: new Date(endedAt.getTime() - 15 * 60_000), + finishedAt: endedAt, lastOutputAt: endedAt, + resultJson: { failureCode: "native_workspace_sync_out_retry_exhausted", finalizationPhase: "terminal_failure" } }).returning(); + const authority = { schema: "paperclip.remote-workspace.v1" as const, epoch: 2, snapshotId: randomUUID(), + source: { kind: "empty" as const }, dirty: true, writerRunId: writer.id, + location: { leaseId: randomUUID(), providerLeaseId: randomUUID(), environmentId: randomUUID(), remoteCwd: "/home/daytona/ws" } }; + const [workspace] = await db.insert(executionWorkspaces).values({ companyId: f.companyId, sourceIssueId: f.parent.id, + name: "Task workspace", mode: "shared_workspace", strategyType: "project_primary", providerType: "cloud_sandbox", + status: "active", cwd: `/workspaces/${randomUUID()}`, metadata: { remoteWorkspace: authority } }).returning(); + await db.update(issues).set({ executionWorkspaceId: workspace.id, executionWorkspacePreference: "reuse_existing" }) + .where(inArray(issues.id, [f.parent.id, f.child.id])); + // The child's stall: its exhausted save recovery, handed to the board. + await db.update(issueRecoveryActions).set({ kind: "active_run_watchdog", cause: "native_workspace_sync_out_retry_exhausted", + ownerType: "board", status: "escalated", evidence: { runId: writer.id, systemRecovery: { state: "exhausted" } } }) + .where(eq(issueRecoveryActions.id, f.action.id)); + // The parent's own last run (an earlier wake) already failed setup on that hold. + const [refused] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.parentAgent, + invocationSource: "automation", status: "failed", errorCode: "setup_failed", + error: "The workspace still has an unfinished writer or save operation.", + createdAt: new Date(Date.now() - 30 * 60_000), finishedAt: new Date(Date.now() - 30 * 60_000), + contextSnapshot: { issueId: f.parent.id, wakeReason: "issue_commented" } }).returning(); + return { ...f, writer, workspace, authority, refused }; + } + + it("holds a stall wake while the waiter's workspace is held by a writer reservation would refuse", async () => { + const f = await legacySharedWorkspaceFixture(); + const { calls, enqueueWakeup } = wakeRecorder(); + const service = recoveryService(db, { enqueueWakeup: enqueueWakeup as never }); + for (let pass = 0; pass < 3; pass += 1) await service.reconcileActiveRecoveryActions({ companyId: f.companyId }); + const held = await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId, issueId: f.parent.id }); + // No run is created that could only fail setup; the sweep looks again later. + expect(calls.filter(call => call.input.reason === ISSUE_BLOCKER_STALLED_WAKE_REASON)).toEqual([]); + expect(held).toMatchObject({ woken: 0, deferred: 1 }); + expect(held.retryAt).toBeGreaterThan(Date.now()); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([]); + + // A writer still inside the holder cutoff parks the run in a workspace wait + // instead of failing it, so that wake is delivered as before. + await db.update(heartbeatRuns).set({ finishedAt: new Date(), lastOutputAt: new Date() }).where(eq(heartbeatRuns.id, f.writer.id)); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(1); + }); + + it("delivers the held stall wake once the workspace hold settles", async () => { + const f = await legacySharedWorkspaceFixture(); + const { calls, enqueueWakeup } = wakeRecorder(); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toEqual([]); + await db.update(executionWorkspaces).set({ metadata: { remoteWorkspace: { ...f.authority, dirty: false, writerRunId: null } } }) + .where(eq(executionWorkspaces.id, f.workspace.id)); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls.map(call => call.agentId)).toEqual([f.parentAgent]); + }); + + it("wakes a waiter with no evidence that its own run is refused by the hold", async () => { + // No run of this assignee has failed on the hold yet: reservation may still + // release the writer, or this agent may not use the workspace at all. + const fresh = await legacySharedWorkspaceFixture(); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, fresh.refused.id)); + // Reassigned to another agent: the old assignee's refused run is no evidence for it. + const reassigned = await legacySharedWorkspaceFixture(); + const otherAgent = randomUUID(); + await db.insert(agents).values({ id: otherAgent, companyId: reassigned.companyId, name: "New owner", role: "engineer", status: "idle" }); + companyIdOf.set(otherAgent, reassigned.companyId); + await db.update(issues).set({ assigneeAgentId: otherAgent }).where(eq(issues.id, reassigned.parent.id)); + const { calls, enqueueWakeup } = wakeRecorder(); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: fresh.companyId }); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: reassigned.companyId }); + expect(calls.map(call => call.agentId)).toEqual([fresh.parentAgent, otherAgent]); + }); + + it("backs off a stall wake whose run keeps failing, and stops after a bounded number of failed runs", async () => { + const f = await fixture(); + const { calls, enqueueWakeup } = wakeRecorder(); + const age = async (minutesAgo: number) => { + const at = new Date(Date.now() - minutesAgo * 60_000); + await db.update(agentWakeupRequests).set({ requestedAt: at }).where(eq(agentWakeupRequests.companyId, f.companyId)); + await db.update(heartbeatRuns).set({ finishedAt: at }).where(eq(heartbeatRuns.companyId, f.companyId)); + }; + // The latest wake's run fails, as the queue records it. + const fail = async (minutesAgo: number) => { + await db.update(agentWakeupRequests).set({ status: "failed" }).where(eq(agentWakeupRequests.companyId, f.companyId)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.parentAgent, invocationSource: "automation", + status: "failed", contextSnapshot: calls.at(-1)!.input.contextSnapshot }); + await age(minutesAgo); + }; + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(1); + // First failed run: retried after ten minutes. + await fail(11); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(2); + // Second failed run: the wait doubles. + await fail(11); + const backingOff = await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(2); + expect(backingOff.retryAt).toBeGreaterThan(Date.now() + 8 * 60_000); + await age(21); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(3); + // Third failed run: the stall is not retried again, and no timer is asked for. + await fail(24 * 60); + const spent = await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(3); + expect(spent.retryAt).toBeNull(); + // A new stall (a new recovery action) is delivered again. + await db.update(issueRecoveryActions).set({ status: "resolved", resolvedAt: new Date(), outcome: "restored" }) + .where(eq(issueRecoveryActions.id, f.action.id)); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.child.id, kind: "run_recovery", + status: "escalated", ownerType: "board", cause: "execution_recovery_budget_exhausted", fingerprint: "fixture-2", nextAction: "Retry" }); + await wakeWaitersOfStalledIssues(db, enqueueWakeup as never, { companyId: f.companyId }); + expect(calls).toHaveLength(4); + }); }); diff --git a/server/src/services/heartbeat/workspace-preparation.ts b/server/src/services/heartbeat/workspace-preparation.ts index f34e4cdd289..25ac839f4cb 100644 --- a/server/src/services/heartbeat/workspace-preparation.ts +++ b/server/src/services/heartbeat/workspace-preparation.ts @@ -2,7 +2,8 @@ import { NATIVE_REMOTE_RECOVERY_LEASE_MISMATCH, remoteRecoveryLeaseMismatch } fr import { readInheritedRunnerRetainedSandbox } from "../native-runtime/native-inherited-runner-retry.js"; import { resolveProviderApiPlacement } from "../native-runtime/provider-api-placement.js"; import { assertEphemeralWorkspaceExecution, readRemoteWorkspaceAuthority, RemoteWorkspaceWriterActiveError, reserveRemoteWorkspace, shouldUseRemoteWorkspaces } from "../remote-workspaces.js"; -import { WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS, WorkspaceBusyDeferral } from "./retries.js"; +import { WorkspaceBusyDeferral } from "./retries.js"; +import { remoteWorkspaceHoldRefusesStart } from "../remote-workspace-writer-hold.js"; import { WORKSPACE_BUSY_RETRY_REASON } from "../../modules/run-dispatch/index.js"; import { readFreshWorkspaceReadiness } from "../fresh-workspace-continuation.js"; import { assertNativeSessionPreparationAuthority } from "../native-runtime/native-execution-lease.js"; @@ -1056,8 +1057,7 @@ async function prepareHeartbeatWorkspaceLocked(db: Db, input: HeartbeatWorkspace } : {}), }).catch(error => { if (!(error instanceof RemoteWorkspaceWriterActiveError) - || (!error.writer.ownerFenceHeld - && error.writer.lastActivityAt.getTime() < Date.now() - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS)) throw error; + || remoteWorkspaceHoldRefusesStart({ kind: "held", writer: error.writer })) throw error; throw new WorkspaceBusyDeferral({ holder: error.writer, executionWorkspaceId: error.workspaceId, deferralAttempt: run.scheduledRetryReason === WORKSPACE_BUSY_RETRY_REASON ? run.scheduledRetryAttempt ?? 0 : 0, wasIssueAssignee: input.task.wasIssueAssignee, holderEnded: error.writer.ended === true }); diff --git a/server/src/services/recovery/blocker-stall-wakeups.ts b/server/src/services/recovery/blocker-stall-wakeups.ts index 9b178731e57..cb44233719d 100644 --- a/server/src/services/recovery/blocker-stall-wakeups.ts +++ b/server/src/services/recovery/blocker-stall-wakeups.ts @@ -19,13 +19,25 @@ * one `blocks`) is woken once per stall. The idempotency key names the * recovery action, so a repeated sweep never wakes twice, and a later stall * (a new action after a Retry or another failure) wakes again. + * + * A wake that did not reach the agent is retried, boundedly. A refusal + * (wake-on-demand off, agent paused) is retried after ten minutes. A stall whose + * runs failed (the keyed wake's run or a resource-wait retry carrying it) is + * retried with a doubling wait and given up after MAX_FAILED_STALL_DELIVERIES + * failed runs; the stall stays listed on the waiter's page. A waiter whose last + * run failed setup on its bound remote workspace's hold is not woken again while + * reservation would still refuse that hold (remoteWorkspaceHoldRefusesStart): + * its run could only fail the same way. It is woken once the hold settles. */ -import { and, eq, gt, inArray, isNull, ne, notInArray, or, sql } from "drizzle-orm"; -import { agents, agentWakeupRequests, heartbeatRuns, issueRecoveryActions, issueRelations, issues, type Db } from "@paperclipai/db"; -import { ISSUE_BLOCKER_STALLED_WAKE_REASON, type IssueBlockerStalledWakeContext, type IssueBlockerStallRelation } from "@paperclipai/shared"; +import { and, desc, eq, gt, inArray, isNull, ne, notInArray, or, sql } from "drizzle-orm"; +import { agents, agentWakeupRequests, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issueRelations, issues, type Db } from "@paperclipai/db"; +import { ISSUE_BLOCKER_STALLED_WAKE_REASON, readRemoteWorkspaceAuthority, type IssueBlockerStalledWakeContext, + type IssueBlockerStallRelation } from "@paperclipai/shared"; import { logActivity } from "../activity-log.js"; import { parseHeartbeatPolicy } from "../heartbeat/timer-policy.js"; import { authorizationService } from "../authorization.js"; +import { readRemoteWorkspaceWriterHold, REMOTE_WORKSPACE_WRITER_PENDING_MESSAGE, + remoteWorkspaceHoldRefusesStart } from "../remote-workspace-writer-hold.js"; const TERMINAL_ISSUE_STATUSES = ["done", "cancelled"]; /** Waiters that may be woken: not finished, and not parked in the backlog by the board. */ @@ -41,6 +53,10 @@ const waiterCursors = new Map(); const DELIVERING_RUN_STATUSES = ["queued", "scheduled_retry", "running", "succeeded"]; /** A refused wake (skipped, failed, cancelled) is retried, but not more often than this. */ const REFUSED_WAKE_RETRY_MS = 10 * 60_000; +/** Failed runs of one stall's wake before it is no longer retried; each retry waits twice as long. */ +const MAX_FAILED_STALL_DELIVERIES = 3; +/** Re-check for a waiter whose workspace hold refuses a run; settling the hold is what lets it run. */ +const HELD_WORKSPACE_RECHECK_MS = 60 * 60_000; const NON_INVOKABLE_AGENT_STATUSES = ["paused", "terminated", "pending_approval"]; /** Company-wide passes resume after the last stall a capped pass reached, so every stall is visited. */ const companyCursors = new Map(); @@ -115,6 +131,30 @@ async function readStalls(db: Db, filter: StallFilter, limit = MAX_STALLS_PER_PA type StallRow = Awaited>[number]; +/** + * Whether a run of the waiter's assignee could only fail setup on its bound + * remote workspace's hold. The evidence is the assignee's own latest run on the + * task: it failed setup on that hold, so this agent uses the workspace and + * reservation already tried, and refused, to release the writer. The hold must + * still be one reservation refuses outright (an ended writer past the holder + * cutoff with its save unsettled, or a writer that cannot be named). Another + * assignee, a later run, or a settled hold lets the wake through. + */ +async function waiterWorkspaceRefusesStart(db: Db, companyId: string, + waiter: { id: string; assigneeAgentId: string | null; executionWorkspaceId: string | null }) { + if (!waiter.executionWorkspaceId || !waiter.assigneeAgentId) return false; + const [latest] = await db.select({ status: heartbeatRuns.status, error: heartbeatRuns.error }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, waiter.assigneeAgentId), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${waiter.id}`)) + .orderBy(desc(heartbeatRuns.createdAt)).limit(1); + if (latest?.status !== "failed" || latest.error !== REMOTE_WORKSPACE_WRITER_PENDING_MESSAGE) return false; + const [workspace] = await db.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces) + .where(and(eq(executionWorkspaces.companyId, companyId), eq(executionWorkspaces.id, waiter.executionWorkspaceId))); + const authority = readRemoteWorkspaceAuthority(workspace?.metadata); + if (!authority?.writerRunId) return false; + return remoteWorkspaceHoldRefusesStart(await readRemoteWorkspaceWriterHold(db, { companyId, authority })); +} + /** What a waiter's agent may see of a stalled issue it cannot read: that it is stalled, nothing more. */ function redactedStallContext(stall: StallRow, relation: IssueBlockerStallRelation): IssueBlockerStalledWakeContext { return { @@ -200,7 +240,7 @@ export async function wakeWaitersOfStalledIssues(db: Db, enqueueWakeup: BlockerS let scanned = 0; for (;;) { const waiters = await db.select({ id: issues.id, identifier: issues.identifier, assigneeAgentId: issues.assigneeAgentId, - status: issues.status, responsibleUserId: issues.responsibleUserId, + status: issues.status, responsibleUserId: issues.responsibleUserId, executionWorkspaceId: issues.executionWorkspaceId, relation: sql<"child" | "blocker">`case when ${issues.id} = ${stall.parentId ?? null} then 'child' else 'blocker' end` }) .from(issues) .where(and( @@ -227,20 +267,36 @@ export async function wakeWaitersOfStalledIssues(db: Db, enqueueWakeup: BlockerS // (company_id, payload->>'issueId') is indexed; the key narrows to this stall. // A delivered or pending wake ends it. A refused one (wake-on-demand off, // agent paused) is retried once the refusal is old, so re-enabling wakes - // still delivers the stall. - const [existing] = await db.select({ id: agentWakeupRequests.id, status: agentWakeupRequests.status, - requestedAt: agentWakeupRequests.requestedAt }).from(agentWakeupRequests) + // still delivers the stall. Failed runs carrying this stall (the keyed + // wake's run, or a resource-wait retry that keeps its context) are + // retried after a doubling wait, and not after MAX_FAILED_STALL_DELIVERIES. + const [prior] = await db.select({ + delivered: sql`coalesce(bool_or(${inArray(agentWakeupRequests.status, DELIVERED_OR_PENDING_WAKE_STATUSES)}), false)`, + lastRequestedAt: sql`max(${agentWakeupRequests.requestedAt})`, + }).from(agentWakeupRequests) .where(and(eq(agentWakeupRequests.companyId, stall.companyId), sql`${agentWakeupRequests.payload}->>'issueId' = ${waiter.id}`, - eq(agentWakeupRequests.idempotencyKey, idempotencyKey), - or(inArray(agentWakeupRequests.status, DELIVERED_OR_PENDING_WAKE_STATUSES), - gt(agentWakeupRequests.requestedAt, new Date(Date.now() - REFUSED_WAKE_RETRY_MS))))).limit(1); - if (existing) { - result.alreadyWoken += 1; - // A recent refusal is retried once it is old; the worker comes back then. - if (!(DELIVERED_OR_PENDING_WAKE_STATUSES as readonly string[]).includes(existing.status)) - retryAt(existing.requestedAt.getTime() + REFUSED_WAKE_RETRY_MS); - continue; + eq(agentWakeupRequests.idempotencyKey, idempotencyKey))); + if (prior?.delivered) { result.alreadyWoken += 1; continue; } + const [failed] = await db.select({ + runs: sql`count(*)::int`, + lastAt: sql`max(coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.createdAt}))`, + }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, stall.companyId), eq(heartbeatRuns.status, "failed"), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${waiter.id}`, + sql`${heartbeatRuns.contextSnapshot}#>>'{blockerStall,stalledIssueId}' = ${stall.issueId}`, + sql`${heartbeatRuns.contextSnapshot}#>>'{blockerStall,recoveryAction,id}' = ${stall.actionId}`)); + const failedRuns = Number(failed?.runs ?? 0); + if (failedRuns >= MAX_FAILED_STALL_DELIVERIES) { result.alreadyWoken += 1; continue; } + const lastAttemptAt = Math.max(prior?.lastRequestedAt ? new Date(prior.lastRequestedAt).getTime() : 0, + failed?.lastAt ? new Date(failed.lastAt).getTime() : 0); + if (lastAttemptAt > 0) { + const dueAt = lastAttemptAt + REFUSED_WAKE_RETRY_MS * 2 ** Math.max(0, failedRuns - 1); + if (dueAt > Date.now()) { + result.alreadyWoken += 1; + retryAt(dueAt); + continue; + } } // A resource wait cancels the stall wake's run and schedules a retry // with no idempotency key; the retry keeps the stall in its context. @@ -251,6 +307,11 @@ export async function wakeWaitersOfStalledIssues(db: Db, enqueueWakeup: BlockerS sql`${heartbeatRuns.contextSnapshot}#>>'{blockerStall,stalledIssueId}' = ${stall.issueId}`, sql`${heartbeatRuns.contextSnapshot}#>>'{blockerStall,recoveryAction,id}' = ${stall.actionId}`)).limit(1); if (carried) { result.alreadyWoken += 1; continue; } + if (await waiterWorkspaceRefusesStart(db, stall.companyId, waiter)) { + result.deferred += 1; + retryAt(Date.now() + HELD_WORKSPACE_RECHECK_MS); + continue; + } if (!(await acceptsWakes(waiter.assigneeAgentId!))) { result.deferred += 1; retryAt(Date.now() + REFUSED_WAKE_RETRY_MS); diff --git a/server/src/services/remote-workspace-writer-hold.ts b/server/src/services/remote-workspace-writer-hold.ts index 721239f9777..cf9f9c946c9 100644 --- a/server/src/services/remote-workspace-writer-hold.ts +++ b/server/src/services/remote-workspace-writer-hold.ts @@ -4,6 +4,7 @@ import type { RemoteWorkspaceAuthority } from "@paperclipai/shared"; import { hasRecoveredCleanupStopReceipt, hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { canonicalExecutionUuid } from "./execution-identity.js"; import { REMOTE_WORKSPACE_PREPARATION_TIMEOUT_CODE } from "./remote-workspace-transport.js"; +import { WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS } from "../modules/run-dispatch/index.js"; const object = (value: unknown): Record => typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record : {}; @@ -21,6 +22,24 @@ export function runOwnerFenceHeldSql() { and ${nativeRunFinalizations.leaseOwner} is not null and ${nativeRunFinalizations.leaseExpiresAt} > now()))`; } +/** The setup error a run gets when reservation refuses a remote workspace hold. */ +export const REMOTE_WORKSPACE_WRITER_PENDING_MESSAGE = "The workspace still has an unfinished writer or save operation."; + +type RemoteWorkspaceWriterHold = Awaited>; + +/** + * Whether a hold refuses a new run outright instead of parking it in a + * workspace wait: a writer that cannot be named, or one silent past the holder + * cutoff with no live owner fence. Workspace reservation fails setup on such a + * hold, so senders of automatic wakes read this same rule to hold a wake until + * the hold settles rather than create a run that can only fail. + */ +export function remoteWorkspaceHoldRefusesStart(hold: RemoteWorkspaceWriterHold, now = Date.now()): boolean { + if (hold.kind === "unattributed") return true; + if (hold.kind !== "held") return false; + return !hold.writer.ownerFenceHeld && hold.writer.lastActivityAt.getTime() < now - WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS; +} + /** * Who still holds a remote workspace through its recorded writer run, other * than `runId`. A running writer holds it. An ended writer keeps holding it diff --git a/server/src/services/remote-workspaces.ts b/server/src/services/remote-workspaces.ts index f61475bce95..df596b35590 100644 --- a/server/src/services/remote-workspaces.ts +++ b/server/src/services/remote-workspaces.ts @@ -23,7 +23,7 @@ import { logger } from "../middleware/logger.js"; import { createGitRemoteAuthProvider } from "./git-credentials.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { assertExecutionWorkspaceRecordAccess, assertTaskWorkspaceExecutionAccess } from "./task-workspace-access.js"; -import { readRemoteWorkspaceWriterHold, type RemoteWorkspaceWriter } from "./remote-workspace-writer-hold.js"; +import { readRemoteWorkspaceWriterHold, REMOTE_WORKSPACE_WRITER_PENDING_MESSAGE, type RemoteWorkspaceWriter } from "./remote-workspace-writer-hold.js"; import { persistActivity } from "./activity-log.js"; import { hasRecoveredCleanupStopReceipt, hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import { @@ -171,7 +171,7 @@ export class RemoteWorkspaceWriterActiveError extends HttpError { readonly workspaceId: string, readonly writer: RemoteWorkspaceWriter, ) { - super(409, "The workspace still has an unfinished writer or save operation.", { code: "remote_workspace_writer_pending" }); + super(409, REMOTE_WORKSPACE_WRITER_PENDING_MESSAGE, { code: "remote_workspace_writer_pending" }); } } @@ -247,7 +247,7 @@ export async function reserveRemoteWorkspace(input: { } const hold = await readRemoteWorkspaceWriterHold(tx as unknown as Db, { companyId: input.companyId, authority, runId: input.runId }); if (hold.kind === "held") throw new RemoteWorkspaceWriterActiveError(existing.id, hold.writer); - if (hold.kind === "unattributed") throw conflict("The workspace still has an unfinished writer or save operation.", { code: "remote_workspace_writer_pending" }); + if (hold.kind === "unattributed") throw conflict(REMOTE_WORKSPACE_WRITER_PENDING_MESSAGE, { code: "remote_workspace_writer_pending" }); if (authority.writerRunId === input.runId) return existing; const next: RemoteWorkspaceAuthority = { ...authority, source, epoch: authority.epoch + 1, writerRunId: input.runId }; const [saved] = await tx.update(executionWorkspaces).set({ metadata: { ...existing.metadata, remoteWorkspace: next,