mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(recovery): stop stall wakes from failing setup on a held workspace (#20273)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Recovery tells a waiting parent or dependent when a child or blocker stalls on the board (`issue_blocker_stalled`) > - A legacy child that an older release bound to its parent's remote workspace failed its save, and its save recovery was exhausted. Its ended run still holds the shared workspace, days past the one-hour holder cutoff > - Workspace reservation refuses a run on such a hold and fails setup with "The workspace still has an unfinished writer or save operation." Every stall wake of the parent created a run that could only fail this way > - The stall sweep treated each failed run as an undelivered wake and retried it every ten minutes, with no backoff and no limit. On staging this made 72 failed runs (58 of 60 failures since a deploy), in pairs every ten minutes > - This pull request makes the stall sweep hold a wake while the waiter's workspace hold refuses a run, and bounds retries of a wake whose run fails > - The benefit is that a held workspace no longer produces a failed run every ten minutes, and the wake still arrives once the hold settles ## Linked Issues or Issue Description **What happened?** A parent task was woken by `issue_blocker_stalled` twice every ten minutes. Each run failed in setup (`setup_failed`, `executionRecovery.kind=bootstrap`, `providerWorkStarted=false`) with "The workspace still has an unfinished writer or save operation." The parent's remote workspace was held by a child's ended writer run whose save recovery was exhausted (`native_workspace_sync_out_retry_exhausted`, last failure `agent_stop_unconfirmed`). The child shared the parent's workspace because an older release created it before subtasks got their own workspace (#19860). **Expected behavior** A wake that cannot start because its workspace is held must not create a failed run every time. It should wait until the hold settles. Automatic retries must be bounded. **Root cause** - `server/src/services/heartbeat/workspace-preparation.ts` rethrows `RemoteWorkspaceWriterActiveError` as a setup failure when the ended writer is silent past `WORKSPACE_BUSY_HOLDER_STALE_AFTER_MS` (documented, intended). - `server/src/services/recovery/blocker-stall-wakeups.ts` retried every non-delivered wake row (including `failed`) after `REFUSED_WAKE_RETRY_MS` (10 minutes) with no backoff and no limit, and it did not check whether the waiter could start at all. ## What Changed - `remote-workspace-writer-hold.ts`: add `remoteWorkspaceHoldRefusesStart`, the one rule for a hold that reservation refuses outright (unattributed writer, or a writer silent past the cutoff with no live owner fence). - `workspace-preparation.ts`: use that rule instead of an inline copy, so reservation and wake senders cannot drift. - `blocker-stall-wakeups.ts`: before waking a waiter, check for proof that its run would be refused. The proof is that the assignee's own latest run on the task failed setup with the unfinished-writer error, and the bound workspace hold still refuses a run. Then do not wake it; count it as deferred and re-check at most hourly. A different assignee (for example a controller-only agent that never uses the workspace), a later run, or a settled hold lets the wake through. Reservation already tried and failed the unchanged-writer release for that run, so the sweep does not repeat that proof. - `blocker-stall-wakeups.ts`: count failed runs that carry the stall (the keyed wake's run and keyless resource-wait retries that keep its context). Retry after 10, then 20 minutes, and stop after the third failed run. Refusals (wake-on-demand off, agent paused) keep the 10-minute retry. A new stall (new recovery action) wakes again. - `remote-workspace-writer-hold.ts` / `remote-workspaces.ts`: share the unfinished-writer error text as one constant. - `doc/execution-semantics.md`: document both rules. ## Verification - Red before the fix (`server/src/__tests__/issue-blocker-stall-wakeups.test.ts`): 3 new tests failed. The held-workspace test got a delivered wake; the backoff test got a third wake 11 minutes after the second failure. - Green after the fix: `npx vitest run src/__tests__/issue-blocker-stall-wakeups.test.ts` passes 22/22. Added after review: a waiter with no refused run, or a reassigned waiter, is still woken; failed resource-wait retries count toward the limit. - The new tests use the staging shape: a legacy `shared_workspace` remote workspace owned by the parent, a child bound to it, the child's failed native writer run ended four days ago with `native_workspace_sync_out_retry_exhausted`, a dirty authority, and the child's board-owned exhausted recovery action. The parent's own earlier run failed setup on the hold, as on staging. They also prove that a writer inside the cutoff still gets its wake (the run parks in a workspace wait), and that the wake is delivered once the hold settles. - Related suites pass: `heartbeat-remote-workspace-busy.test.ts`, `heartbeat-dependency-scheduling.test.ts`, `native-cancelled-workspace-preservation.test.ts`, `src/modules/run-dispatch`. - `npx tsc --noEmit -p server` passes. - `pnpm verify` exits 2 because `doc/execution-semantics.md` is a global CI input; required CI supplies the broad coverage. ## Risks - Low. No schema or persisted-shape change. Saved state from the previous release (existing wake rows, existing workspace authority) is read as-is: earlier `failed` stall wake rows count toward the new three-run limit, which stops the current loop at once after deploy. - A parent whose workspace stays held is not woken for the stall until the hold settles. The stall is still listed on its page and the child's recovery action stays with the board. - Not fixed here: the exhausted save hold itself is never settled automatically (`agent_stop_unconfirmed` exhaustion in workspace recovery), and legacy children that still share a parent's workspace are not moved. Both touch workspace recovery code that is under concurrent change and are left for a follow-up. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, browser read-only inspection of staging). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
2dd7a98753
commit
1004e55180
6 files changed
+262
-26
No files matched your search
@@ -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
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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 });
|
||||
|
||||
@@ -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<string, string>();
|
||||
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<string, string>();
|
||||
@@ -115,6 +131,30 @@ async function readStalls(db: Db, filter: StallFilter, limit = MAX_STALLS_PER_PA
|
||||
|
||||
type StallRow = Awaited<ReturnType<typeof readStalls>>[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<boolean>`coalesce(bool_or(${inArray(agentWakeupRequests.status, DELIVERED_OR_PENDING_WAKE_STATUSES)}), false)`,
|
||||
lastRequestedAt: sql<string | null>`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<number>`count(*)::int`,
|
||||
lastAt: sql<string | null>`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);
|
||||
|
||||
@@ -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<string, unknown> =>
|
||||
typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
@@ -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<ReturnType<typeof readRemoteWorkspaceWriterHold>>;
|
||||
|
||||
/**
|
||||
* 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
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in new issue
Block a user