fix(workspaces): release the Blocked status of a sibling-settled no-replay hold (#20648)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - When a remote workspace save fails, recovery actions hold the task
until the run's files are saved
> - #20641 lets a resolved no-replay hold settle when a sibling action
of the same run saved the files
> - Review found that the settlement did not release the task's Blocked
status, and that a settled hold with a historical source stayed
selectable
> - A new test with the waiting user comment also showed that the worker
skipped the second hold of the same run in every pass, so it never
settled
> - #20641 was merged before these fixes were pushed
> - This pull request completes #20641 with those fixes
> - The benefit is that the task leaves Blocked and continues without a
person

## Linked Issues or Issue Description

Refs #20641, Refs #20569

**What happened?**
#20641 merged at its first head. The review follow-ups for it were
pushed later and are not on master:
- A settled no-replay hold that owned the task's Blocked status
(`nativeFailureBlock`) left the task `blocked`.
- A hold settled by a sibling stayed selectable through
`workspaceRecoverySourceEligible` when it named a
`historicalWorkspaceSource`.
- With saved user input waiting, the export hold stayed selectable and
came first in each worker pass. `reconcileRetainedWorkspaces` skipped
the other hold of the same run, so the no-replay hold never settled.

**Expected behavior**
After the save, the no-replay hold settles in the same or next pass. The
task returns to `todo` when that hold owns its Blocked status. The
settled hold is never selected again.

**Steps to reproduce**
1. Use the staging-shape test in
`workspace-recovery-capture-policy.test.ts`, with a waiting "try again"
comment.
2. Run the worker until nothing is due.
3. On master, the worker never goes idle and the no-replay hold stays
`resolved`/`blocked`.

## What Changed

- `workspace-recovery-fallback.ts`: new shared `holdOwnsBlockedStatus`,
used by `continueFromLastSavedWorkspace` and
`settleSiblingRecoveredHold`. A settled no-replay hold that owns the
Blocked status returns the task to `todo`. A newer decision's Blocked
status is kept. An export hold's status stays with its own result
finalization.
- `workspace-recovery-source.ts`: `workspaceRecoverySourceEligible`
excludes a resolved hold that carries `siblingWorkspaceRecovery`.
- `workspace-recovery.ts`: `reconcileRetainedWorkspaces` runs several
holds of one run one after another in that run's job, instead of
skipping the later ones.
- `workspace-recovery-capture-policy.test.ts`: the staging test now has
the waiting comment, and a variant with `nativeFailureBlock` and
`historicalWorkspaceSource`. It checks `todo`, no re-selection, and no
repeat handoff in a later pass.
- `doc/execution-semantics.md`: documents these rules.

## Verification

- Red, with each rule disabled: the status stayed `blocked`; the settled
hold was selected again; the worker never went idle (starvation).
- Green: `npx vitest run --root server` on 14 workspace recovery suites
(including `workspace-recovery-capture-policy`, `-fallback`,
`-admission`, `-source`, `execution-blocker-read`,
`preserved-native-response-wait.integration`): all passed.
- `npx tsc --noEmit -p server/tsconfig.json`: clean.

## Risks

- Several holds of one run now run in sequence in one job. Concurrency
per run is unchanged (one job per run).
- A no-replay hold releases the Blocked status only when it still owns
it. Otherwise the status is kept.
- Previous-release story: no new persisted state. Holds saved by earlier
releases are handled on the first worker pass after deploy.

> 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 (Anthropic), model ID `claude-opus-5-5`, used through
Claude Code with tool use and code execution.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed 2026-11-10 15:24:04 -06:00
1 parent 805531cc71
commit 4f33bb3578
6 files changed
+113 -32

No files matched your search

+6 -1
View File
@@ -2801,7 +2801,12 @@ take the same path after deploy. An exhausted hold whose run's files a sibling
recovery action already saved settles as recovered. This includes a no-replay
hold recorded resolved-but-blocked: settling it removes its replay block, so it
stops blocking the task, and starts no turn itself; the run's own result and
any waiting input decide what continues.
any waiting input decide what continues. When that no-replay hold still owns
the task's Blocked status (its recorded failure block names the current status
version), the task returns to `todo`; a newer decision's Blocked status is kept.
A settled hold is never selected for recovery again. The worker handles several
holds of one run one after another in the same pass, so a hold that stays
selectable (for example while saved input waits) cannot starve a later one.
An exhausted save whose last failure was `agent_stop_unconfirmed` is not a
refusal of the saved files: its recovery window ran out before the agent's stop
@@ -3,7 +3,7 @@ import { mkdir, readFile, symlink, writeFile } from "node:fs/promises";
import path from "node:path";
import { eq } from "drizzle-orm";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { activityLog, executionWorkspaces, heartbeatRuns, issueRecoveryActions } from "@paperclipai/db";
import { activityLog, agentWakeupRequests, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues } from "@paperclipai/db";
import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
import { readRemoteWorkspaceAuthority } from "@paperclipai/shared";
import { getExecutionBlocker } from "./execution-blocker.js";
@@ -110,7 +110,7 @@ describe("saves exhausted under an older workspace capture policy", () => {
* Its six attempts failed the same way and it was recorded
* resolved-but-blocked. That second hold is the task's execution blocker.
*/
async function stagingIncident() {
async function stagingIncident(failureBlock: boolean) {
const f = await exhaustedCaptureHold();
const exportHold = await action(f.actionId);
const recovery = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] };
@@ -129,6 +129,7 @@ describe("saves exhausted under an older workspace capture policy", () => {
historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: f.runId, leaseIds: [f.lease.id] } } })
.where(eq(issueRecoveryActions.id, f.actionId));
const exhausted = readSystemRecovery((await action(f.actionId)).evidence)!;
const [task] = await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, f.issueId)).returning();
const noReplayHoldId = randomUUID(), resolvedAt = new Date(Date.now() - 30 * 60_000);
await db.insert(issueRecoveryActions).values({ id: noReplayHoldId, companyId: f.companyId, sourceIssueId: f.issueId,
kind: "active_run_watchdog", status: "resolved", outcome: "blocked", ownerType: "board", resolvedAt,
@@ -140,12 +141,21 @@ describe("saves exhausted under an older workspace capture policy", () => {
workspaceRecovery: exportHold.evidence.workspaceRecovery,
originalFailureCode: "native_workspace_sync_out_retry_exhausted", continuationDelivery: "invalidated",
systemRecoveryActiveAt: Date.now() - 40 * 60_000, workspaceRestoreFailure: "restore_failed",
workspaceRestoreRecovery: recovery, previousWorkspaceRecovery: null } });
return { ...f, noReplayHoldId };
workspaceRestoreRecovery: recovery, previousWorkspaceRecovery: null,
// Rows of the same release that recorded the Blocked status their run's
// failure set, and the historical source an older adoption named.
...(failureBlock ? { nativeFailureBlock: { runId: f.runId, statusVersion: Number(task.statusVersion) },
historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: f.runId, leaseIds: [f.lease.id] } } : {}) } });
// The user's "try again" comment waits for the task to be free.
const [waiting] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: source.agentId, source: "on_demand",
reason: "issue_commented", status: "deferred_issue_execution", requestedByActorType: "user", requestedByActorId: "operator",
payload: { issueId: f.issueId } }).returning();
return { ...f, noReplayHoldId, waitingInputId: waiting.id };
}
it("settles the staging incident's export hold and its resolved no-replay sibling without a person", async () => {
const f = await stagingIncident();
it.each([["as recorded on staging", false], ["that owns the task's Blocked status", true]] as const)(
"settles the export hold and a resolved no-replay sibling %s without a person", async (_shape, failureBlock) => {
const f = await stagingIncident(failureBlock);
// As on staging: the no-replay sibling is the execution blocker, and it
// offers no task Retry.
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ recoveryActionId: f.noReplayHoldId,
@@ -154,13 +164,17 @@ describe("saves exhausted under an older workspace capture policy", () => {
// The recovery worker's own discovery, as after a deploy: it runs while
// work is due and goes idle once nothing is.
const blockersSeenByContinuation: unknown[] = [];
const onRecovered = vi.fn(async () => { blockersSeenByContinuation.push(await getExecutionBlocker(db, f.companyId, f.issueId)); });
// Normal dispatch consumes the saved input once nothing blocks the task.
const onRecovered = vi.fn(async () => {
const blocker = await getExecutionBlocker(db, f.companyId, f.issueId);
blockersSeenByContinuation.push(blocker);
if (!blocker) await db.update(agentWakeupRequests).set({ status: "queued" }).where(eq(agentWakeupRequests.id, f.waitingInputId));
});
let passes = 0;
for (; passes < 6; passes += 1) {
const due = await nextWorkspaceRecoveryAt(db);
if (due === null || due > Date.now()) break;
await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true) ?? []);
}
await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true) ?? []); }
expect(passes).toBeGreaterThan(0);
expect(passes).toBeLessThan(6);
@@ -182,11 +196,23 @@ describe("saves exhausted under an older workspace capture policy", () => {
// The continuation hook ran once the task was free, so saved input is dispatched.
expect(blockersSeenByContinuation.at(-1)).toBeNull();
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).map(row => row.id)).toEqual([f.runId]);
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.waitingInputId)))[0]!.status).toBe("queued");
// A hold that owned the Blocked status releases it; otherwise the status is
// left to the run's own result finalization.
const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId));
if (failureBlock) expect(task!.status).toBe("todo");
// Bounded: one re-arm, and the worker has nothing left to do.
const next = await nextWorkspaceRecoveryAt(db);
expect(next === null || next > Date.now()).toBe(true);
expect((await db.select().from(activityLog).where(eq(activityLog.action, "issue.workspace_recovery_rearmed")))
.filter(row => row.entityId === f.issueId)).toHaveLength(1);
// A later pass, even with new saved input, never selects or hands off the settled hold again.
await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" }).where(eq(agentWakeupRequests.id, f.waitingInputId));
expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).not.toContain(f.noReplayHoldId);
expect((await retainedWorkspaceCandidates(db)).map(row => row.action.id)).not.toContain(f.noReplayHoldId);
await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true) ?? []);
expect(await action(f.noReplayHoldId)).toEqual(sibling);
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
}, 90_000);
it("does not re-arm a save that the current policy already exhausted", async () => {
@@ -465,10 +465,13 @@ describe("retained source fallback", () => {
expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true);
await db.update(issueRecoveryActions).set({ status: "resolved", resolvedAt: new Date() }).where(eq(issueRecoveryActions.id, f.actionId));
const staleId = randomUUID(), now = Date.now();
// An export hold that names the task's Blocked status leaves the status to
// its saved result's finalization; the sibling settlement must not change it.
const [task] = await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, f.issueId)).returning();
await db.insert(issueRecoveryActions).values({ id: staleId, companyId: f.companyId, sourceIssueId: f.issueId,
kind: "active_run_watchdog", status: "active", ownerType: "board", cause: "native_workspace_sync_out_retry_exhausted",
fingerprint: randomUUID(), nextAction: "Automatic recovery of saved work stopped after 6 attempts.",
evidence: { runId: f.runId,
evidence: { runId: f.runId, nativeFailureBlock: { runId: f.runId, statusVersion: Number(task.statusVersion) },
systemRecovery: { schema: "paperclip.system-recovery.v1", policyVersion: 1, sourceRunId: f.runId, revision: 54,
generationId: randomUUID(), intentId: null, phase: "preserve_files", state: "exhausted", attemptId: randomUUID(),
attempts: 6, retriesUsed: 5, firstFailureAt: now - 60_000, nextAttemptAt: null, deadlineAt: now - 1, providerCooldownAt: null,
@@ -488,6 +491,7 @@ describe("retained source fallback", () => {
const live = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId),
inArray(issueRecoveryActions.status, ["active", "escalated"])));
expect(live).toHaveLength(0);
expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]!.status).toBe("blocked");
}, 60_000);
it("stops discovering a stale hold whose sibling receipt no longer verifies", async () => {
@@ -26,7 +26,7 @@ import { abandonUnverifiedRemoteWorkspaceEpoch } from "./remote-workspace-epoch-
import { hasStopOnlyCleanup, prepareHistoricalSandboxStopVerification, prepareSandboxStopAndRetain, restoreSandboxStopAfterFailedRestart, settleStopOnlyCleanup } from "./sandbox-stop-and-retain.js";
import { nativeWorkspaceFinalizationBlockSchema } from "./native-runtime/preserved-native-result-finalization.js";
import { withNativeWorkspaceFinalizationOwnership } from "./native-runtime/native-workspace-finalization-ownership.js";
import { PERMANENT_SOURCE_RETRY_CLASSES, RETAINED_SOURCE_FALLBACK_CHECK_KEY, UNVERIFIABLE_SOURCE_CODES, isRetainedFileHold,
import { NATIVE_NO_REPLAY_HOLD_CAUSE, PERMANENT_SOURCE_RETRY_CLASSES, RETAINED_SOURCE_FALLBACK_CHECK_KEY, UNVERIFIABLE_SOURCE_CODES, isRetainedFileHold,
retainedHoldFallbackEligible } from "./workspace-recovery-fallback-discovery.js";
import { captureReadOnlyWorkspaceArchive, sandboxProviderFailureLabel } from "./remote-workspace-transport.js";
import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js";
@@ -303,6 +303,27 @@ async function confirmSourceAllocationStopped(db: Db, input: Source & { environm
* the recovery delivery starts one continuation. Bounded and idempotent: a
* settled hold is never revisited, and a lost race leaves the hold unchanged.
*/
/**
* Whether the task's Blocked status still belongs to this hold: its own
* receipt names the current status version or, for holds saved before
* receipts existed, this run's terminal failure set the block and nothing
* changed it since. Otherwise a newer decision owns the block.
*/
function holdOwnsBlockedStatus(action: typeof issueRecoveryActions.$inferSelect, issue: typeof issues.$inferSelect,
run: typeof heartbeatRuns.$inferSelect) {
const block = nativeWorkspaceFinalizationBlockSchema.safeParse(action.evidence.workspaceFinalizationBlock);
// A native no-replay hold records the Blocked status version its run's failure set.
const failureBlock = object(action.evidence.nativeFailureBlock);
const blockedAt = issue.blockedTransitionAt?.getTime();
return issue.status === "blocked" && (block.success
? block.data.issueId === issue.id && block.data.runId === run.id
&& block.data.statusVersion === Number(issue.statusVersion) && block.data.priorDecisionId === issue.lastStatusDecisionId
: action.evidence.nativeFailureBlock !== undefined
? failureBlock.runId === run.id && failureBlock.statusVersion === Number(issue.statusVersion)
: action.evidence.workspaceFinalizationBlock === undefined && issue.lastStatusDecisionId === null && blockedAt !== undefined
&& run.finishedAt !== null && blockedAt >= run.finishedAt.getTime() - 1_000 && blockedAt <= action.createdAt.getTime() + 60_000);
}
export async function continueFromLastSavedWorkspace(input: Source & {
db: Db; environmentRuntime: EnvironmentRuntimeService; ownership: Ownership;
}): Promise<RetainedSourceFallbackResult> {
@@ -346,17 +367,7 @@ export async function continueFromLastSavedWorkspace(input: Source & {
// holds saved before receipts existed, a block set by this run's terminal
// failure that nothing changed since) still names it. Otherwise a newer
// decision owns the block: it is kept and no continuation is started.
const block = nativeWorkspaceFinalizationBlockSchema.safeParse(action.evidence.workspaceFinalizationBlock);
// A native no-replay hold records the Blocked status version its run's failure set.
const failureBlock = object(action.evidence.nativeFailureBlock);
const blockedAt = issue.blockedTransitionAt?.getTime();
const ownsBlock = issue.status === "blocked" && (block.success
? block.data.issueId === issue.id && block.data.runId === run.id
&& block.data.statusVersion === Number(issue.statusVersion) && block.data.priorDecisionId === issue.lastStatusDecisionId
: action.evidence.nativeFailureBlock !== undefined
? failureBlock.runId === run.id && failureBlock.statusVersion === Number(issue.statusVersion)
: action.evidence.workspaceFinalizationBlock === undefined && issue.lastStatusDecisionId === null && blockedAt !== undefined
&& blockedAt >= run.finishedAt!.getTime() - 1_000 && blockedAt <= action.createdAt.getTime() + 60_000);
const ownsBlock = holdOwnsBlockedStatus(action, issue, run);
const dispatchable = issue.status !== "blocked" || ownsBlock;
const userText = quarantine.state === "captured"
? "Paperclip could not verify the previous run's unsaved files, so the task continues from the last saved version. Those files are attached to the task, unapplied."
@@ -460,7 +471,7 @@ export async function settleExhaustedRetainedHold(input: Source & { db: Db; envi
*/
export async function settleSiblingRecoveredHold(db: Db, input: Source) {
const settled = await db.transaction(async tx => {
const [issue] = await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, input.companyId),
const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, input.companyId),
eq(issues.id, input.issueId))).for("update");
const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId),
eq(heartbeatRuns.id, input.runId))).for("update");
@@ -468,10 +479,13 @@ export async function settleSiblingRecoveredHold(db: Db, input: Source) {
eq(issueRecoveryActions.sourceIssueId, input.issueId), eq(issueRecoveryActions.id, input.actionId))).for("update");
if (!issue || !run || !action || action.evidence.runId !== run.id) return null;
if (object(action.evidence.siblingWorkspaceRecovery).schema === SIBLING_WORKSPACE_RECOVERY_SCHEMA) return "already" as const;
// isRetainedFileHold also accepts export holds; only the no-replay cause
// may release the status here. An export hold leaves it to its result.
const noReplayHold = action.cause === NATIVE_NO_REPLAY_HOLD_CAUSE && isRetainedFileHold(action);
const system = readSystemRecovery(action.evidence);
// A no-replay hold is recorded resolved-but-blocked. It blocks the task
// until its run's files are saved, so a sibling's save settles it too.
if ((!["active", "escalated"].includes(action.status) && !isRetainedFileHold(action))
if ((!["active", "escalated"].includes(action.status) && !noReplayHold)
|| system?.state !== "exhausted" || system.sourceRunId !== run.id) return null;
const receiptId = await verifiedWorkspaceRepairReceipt(tx as unknown as Db, run);
const [receipt] = receiptId ? await tx.select({ metadata: workspaceOperations.metadata }).from(workspaceOperations).where(and(
@@ -498,6 +512,18 @@ export async function settleSiblingRecoveredHold(db: Db, input: Source) {
siblingWorkspaceRecovery: { schema: SIBLING_WORKSPACE_RECOVERY_SCHEMA, receiptId, recoveryActionId: siblingActionId,
recordedAt: now.toISOString() } },
}).where(and(eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.id, action.id)));
// A no-replay hold that still owns the task's Blocked status releases it,
// so saved input and normal wakes continue the task. An export hold's
// status stays with its own result finalization. A newer decision's
// Blocked status is kept.
if (noReplayHold && holdOwnsBlockedStatus(action, issue, run)) {
await tx.update(issues).set({ status: "todo", unblockDescriptor: null, blockedTransitionAt: null,
blockedOwnerNotifiedAt: null, updatedAt: now }).where(eq(issues.id, issue.id));
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
await notifyStatusCardIssueChange(tx, input.companyId);
await notifyRoutineExecutionStatus(tx, issue.id, input.companyId);
}
await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation);
return persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system", actorId: "workspace-recovery",
action: "issue.workspace_recovered", entityType: "issue", entityId: issue.id, runId: run.id,
@@ -26,13 +26,15 @@ export function workspaceSourceBindingFailure(reason: WorkspaceSourceVerificatio
export class WorkspaceSourceInspectionLimit extends Error {}
// A resolved hold that a sibling's save settled has nothing left to recover,
// so it is never selected again, even when it names a historical source.
export function workspaceRecoverySourceEligible(includeCompleted = false) {
const actionEligible = sql`(${issueRecoveryActions.status} in ('active', 'escalated') or
(${issueRecoveryActions.status} = 'cancelled'
and ${inArray(issueRecoveryActions.cause, workspaceExportCauses)}
and jsonb_typeof(${issueRecoveryActions.evidence}#>'{workspaceRecovery,requestIds}') = 'array'
and ${issueRecoveryActions.evidence}#>'{workspaceRecovery,requestIds}' <> '[]'::jsonb) or
(${issueRecoveryActions.status} = 'resolved' and (
(${issueRecoveryActions.status} = 'resolved' and ${issueRecoveryActions.evidence}->'siblingWorkspaceRecovery' is null and (
(${issueRecoveryActions.evidence}#>>'{automaticRecovery,policy}' = 'preserve_without_replay_v1'
and ${issueRecoveryActions.evidence}#>>'{automaticRecovery,replay}' = 'blocked'
and ${issueRecoveryActions.evidence}#>>'{automaticRecovery,runId}' = ${heartbeatRuns.id}::text) or
+24 -6
View File
@@ -663,16 +663,36 @@ export async function reconcileRetainedWorkspaces(db: Db, runtime: EnvironmentRu
if (!rows.length && !passiveSweep) rows = await retainedWorkspaceCandidates(db, undefined, false, obligationsOnly, inFlightRunIds);
const work: Promise<void>[] = [];
let hasUnvisited = false;
const advance = (actionId: string) => {
if (unfinished) unfinishedCursors.set(db, actionId);
else if (passiveSweep) passiveSweep.cursor = actionId;
else cursors.set(db, actionId);
};
// Several actions of one run (an export hold and a later no-replay hold)
// run one after another in that run's job. Skipping the later one would
// starve it while the earlier one stays selectable (for example while saved
// input waits for the later one to settle).
const groups = new Map<string, { run: typeof heartbeatRuns.$inferSelect; actions: (typeof issueRecoveryActions.$inferSelect)[] }>();
for (const { action, run } of rows) {
const group = groups.get(run.id);
if (group) { group.actions.push(action); advance(action.id); continue; }
if (pending.size >= 2) { hasUnvisited = true; break; }
if (unfinished) unfinishedCursors.set(db, action.id);
else if (passiveSweep) passiveSweep.cursor = action.id;
else cursors.set(db, action.id);
advance(action.id);
if (pending.has(run.id)) continue;
pending.add(run.id);
groups.set(run.id, { run, actions: [action] });
}
for (const { run, actions } of groups.values()) {
const finish = beginIdleTrackedWork();
const physical = ["scheduled", "running", "awaiting_evidence"].includes(readSystemRecovery(action.evidence)?.state ?? "");
const physical = actions.some(action => ["scheduled", "running", "awaiting_evidence"].includes(readSystemRecovery(action.evidence)?.state ?? ""));
work.push((async () => { try {
for (const action of actions) await recoverRunAction(action, run);
} finally { pending.delete(run.id); finish(); onFinished?.(physical || hasUnvisited); } })());
}
return work;
async function recoverRunAction(action: typeof issueRecoveryActions.$inferSelect, run: typeof heartbeatRuns.$inferSelect) {
try {
await recoverRetainedWorkspace({ db, environmentRuntime: runtime, companyId: action.companyId, issueId: action.sourceIssueId, runId: run.id, actionId: action.id });
const [updated] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id));
const recovered = readWorkspaceRecovery(updated?.evidence ?? null)?.status === "recovered";
@@ -691,7 +711,5 @@ export async function reconcileRetainedWorkspaces(db: Db, runtime: EnvironmentRu
}
}
} catch { logger.warn({ runId: run.id }, "Retained workspace remains pending"); }
finally { pending.delete(run.id); finish(); onFinished?.(physical || hasUnvisited); } })());
}
return work;
}