mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(native): continue a task whose quarantined completed turn's holds settle (#20662)
<!-- Write all pull request text in Simplified Technical English (ASD-STE100): short sentences, one instruction per sentence, simple approved vocabulary, and the active voice. --> ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Before a remote run starts, workspace preparation attaches the verified saved work of earlier runs of the same task. > - Preparation failed with `remote_workspace_recovery_receipt_invalid` when any earlier run had a recovery receipt that it could not verify. > - A refusal cannot change the saved state, so every Retry failed in the same way, and the task stayed blocked. Recoveries must fall back automatically. > - This pull request skips an unverifiable source. It attaches nothing unverified and starts from the workspace's newest verified save. > - The benefit is that Retry and continuations start again, while strict receipt checks still decide what is attached, and no saved work is changed. ## Linked Issues or Issue Description Refs #20631 **What happened?** On staging, after the deploy that included #20631, a user pressed Retry on a task whose previous run failed with `native_controller_recovery_missing_evidence:runner_not_settled`. The new run failed in setup in under a minute with `setup_failed` / `remote_workspace_recovery_receipt_invalid` (`executionRecovery: bootstrap`, `providerWorkStarted: false`). The task stayed blocked. Another Retry would fail the same way. #20631 accepted one shape of this refusal (a source that saved only the snapshot its own restore loaded). This task had a different shape. **Expected behavior** A Retry starts. A source whose receipt cannot be verified is not attached, and its saved state stays intact for workspace recovery. **Steps to reproduce** 1. Let a remote native run fail before provider work and leave a recovery receipt or restore hold that does not verify (for example a receipt on an unsaved reference, a receipt from another epoch, or a saved reference without a receipt). 2. Press Retry on the task. 3. See the new run fail in setup with `remote_workspace_recovery_receipt_invalid`. **Paperclip version or commit** Staging deploy that includes `e3bc4ca798`. The gap remains on `master` at `a3b34a9d88`. ## What Changed - `attachPreservedRemoteWorkspaceSnapshots` no longer throws for an unverifiable source. Each of the five refusal shapes now skips the source and logs `remote_workspace_recovery_receipt_invalid` with a reason: `receipt_without_reference`, `reference_unreadable`, `receipt_on_unsaved_reference`, `receipt_unverified`, and `receipt_snapshot_missing` (was `remote_workspace_recovery_snapshot_missing`). - `retainRemoteRecoveryAttachments` no longer throws for a retained recovered copy that it cannot verify. It does not publish that copy again. The files stay in the workspace. - The writer check (`remote_workspace_recovery_writer_changed`), the workspace access check, and snapshot read and digest failures still refuse. - `doc/execution-semantics.md` describes the fallback rule. ## Verification - Red on unfixed code: `remote-workspace-recovery-receipt-fallback.test.ts` (moved out of `workspace-recovery-preparation.test.ts` to keep that file under 1,000 lines) "starts a Retry from the newest verified save when a predecessor has %s" (6 shapes: `receipt_without_reference`, `unreadable_reference_with_hold`, `receipt_on_prepared_reference`, `superseded_snapshot_with_hold`, `receipt_from_other_epoch`, `receipt_for_other_run`). All 6 failed with `remote_workspace_recovery_receipt_invalid`. The source is a native run that failed with `runner_not_settled`, and the writer is a Retry (`retryOfRunId`, `taskRetryRequestId`) that follows the real ancestry query. - Green: the same 6 tests pass. They also assert that nothing is attached, and that the source run row, its workspace operations, and its snapshot row are unchanged. - Updated safety tests: the canonical-attachment fault matrix now expects no attachment for receipt faults, and still expects a refusal for a missing or corrupt archive and for a wrong company or task. The retained-history test expects an unverifiable retained copy to be left out of the published list, while verified copies stay published. - `npx vitest run src/services/remote-workspace-recovery-receipt-fallback.test.ts src/services/workspace-recovery-preparation.test.ts`: 91 passed. - `npx tsc --noEmit -p server/tsconfig.json` passes. - `pnpm verify` returns exit 2, because a global doc input changed (`doc/execution-semantics.md`). Required CI supplies the broad coverage. - Not verified live on Daytona. I could not read the staging rows for this task: the browser had no free tab. The tests cover every refusal shape of this function, so the exact staging shape is one of them. ## Risks - Previous-release state: the rule reads the same rows as before (run reference, result hold, workspace operations, recovery actions, snapshots). Rows that earlier releases saved carry forward. No migration. A blocked task recovers on its next Retry or wake. - A Retry can now start without files from an unverifiable source. Those files are not lost: the source's snapshot, restore hold, receipts and recovery action stay unchanged, and workspace recovery still owns them. The skip is logged with a named reason. - A retained copy that fails verification is dropped from the published list. Its directory stays in the workspace and is no longer retired automatically. - A missing or corrupt snapshot archive still refuses. A permanently lost archive therefore still blocks attach. This PR does not change that. > 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, via Claude Code with tool use (shell, file edits). Extended reasoning enabled. ## 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
8f6ca25976
commit
48c6e996da
5 files changed
+322
-48
No files matched your search
@@ -1790,9 +1790,31 @@ reconciler lists them as `native_completed_turn_cleanup_hold`. Once it proves
|
||||
the controller that closed the session gone (its in-process cleanup gate went
|
||||
with it), it settles each such hold as `restored` and removes its replay block.
|
||||
A hold that still protects files that are not verified as saved stays for
|
||||
workspace recovery. Settling starts no turn: the task keeps its own state (for
|
||||
example, waiting on a child task) and continues through normal admission when
|
||||
that wait ends. Saved user input is resumed through the ordinary resume path.
|
||||
workspace recovery. Settling starts no turn by itself. Saved user input is
|
||||
resumed through the ordinary resume path.
|
||||
|
||||
The run's failure usually also set the task's Blocked status. In the same
|
||||
transaction, the reconciler decides that status once and records the decision on
|
||||
the settled holds (`completedTurnCleanupSettlement.blockedStatus`):
|
||||
|
||||
- `released`: a settled hold still owns the Blocked status
|
||||
(`holdOwnsBlockedStatus`: its `nativeFailureBlock` names the current status
|
||||
version), the run's agent is still the assignee, and the task has no
|
||||
unfinished blocker task. The task moves to `todo`, and that hold records one
|
||||
durable continuation (`continuationDelivery: "pending"`). The
|
||||
execution-continuation worker delivers it once, as a fresh session, through
|
||||
normal admission. When saved user input waits for the task, that input owns
|
||||
the continuation and no second one is recorded.
|
||||
- `dependency_wait`: the task also waits on an unfinished blocker task. It stays
|
||||
Blocked, and the blocker's completion wakes it.
|
||||
- `not_owned`: a newer decision owns the Blocked status, or the task changed
|
||||
hands. The status is kept and nothing starts.
|
||||
|
||||
Holds that an earlier release settled without this decision are listed again
|
||||
while their task is still Blocked, so such tasks continue on the first pass
|
||||
after deploy, without a migration. A task never rests Blocked with no blocker,
|
||||
no recovery action and no next action because its quarantined run's cleanup
|
||||
settled.
|
||||
|
||||
A continuation cancelled by the queued-run gate while waiting for review did not start a provider. Its matching dispatch-gate receipt and unclaimed execution fields keep this deliberate wait out of unknown-action reconciliation. Pending review and decision gates still apply. If an older server already created a recovery hold for this exact state, Retry or a new user message can request one fresh turn after the server verifies there is no provider, coordinator, or unfinished cleanup. Saved user input is reconsidered through the same admission path; the cancelled turn is not replayed.
|
||||
|
||||
|
||||
@@ -1,15 +1,18 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
agentWakeupRequests,
|
||||
agents,
|
||||
authUsers,
|
||||
companies,
|
||||
controllerInstances,
|
||||
createDb,
|
||||
heartbeatRuns,
|
||||
issueRecoveryActions,
|
||||
issueRelations,
|
||||
issues,
|
||||
nativeRunFinalizations,
|
||||
workspaceOperations,
|
||||
@@ -17,6 +20,8 @@ import {
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { currentControllerBootId, currentProcessSpaceHash } from "../services/controller-process-identity.js";
|
||||
import { getExecutionBlocker } from "../services/execution-blocker.js";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { deliverReconciledExecutions } from "../services/execution-recovery-resolution.js";
|
||||
import { registerControllerInstanceDb, resetControllerInstanceLeaseForTests } from "../services/controller-instance-lease.js";
|
||||
import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js";
|
||||
|
||||
@@ -50,8 +55,11 @@ suite("completed-turn cleanup holds on the shared ownership path", () => {
|
||||
temporary = await startEmbeddedPostgresTestDatabase("paperclip-completed-turn-cleanup-");
|
||||
db = createDb(temporary.connectionString);
|
||||
companyId = randomUUID(); agentId = randomUUID();
|
||||
const responsibleUserId = randomUUID();
|
||||
await db.insert(authUsers).values({ id: responsibleUserId, name: "Board", email: `${responsibleUserId}@example.test`,
|
||||
emailVerified: true, createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: "CTC",
|
||||
requireBoardApprovalForNewAgents: false });
|
||||
requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: responsibleUserId });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Coder", role: "engineer", status: "idle",
|
||||
adapterType: "paperclip_runner", adapterConfig: {}, runtimeConfig: {}, permissions: {} });
|
||||
}, 30_000);
|
||||
@@ -61,18 +69,29 @@ suite("completed-turn cleanup holds on the shared ownership path", () => {
|
||||
await db.delete(issueRecoveryActions);
|
||||
await db.delete(workspaceOperations);
|
||||
await db.delete(nativeRunFinalizations);
|
||||
await db.delete(heartbeatRuns);
|
||||
// A delivered continuation leaves a wake and its run referencing each other.
|
||||
await db.execute(sql`truncate table agent_wakeup_requests, heartbeat_runs cascade`);
|
||||
await db.delete(issueRelations);
|
||||
await db.delete(issues);
|
||||
});
|
||||
|
||||
afterAll(async () => { await temporary?.cleanup(); });
|
||||
|
||||
async function quarantinedRun(options: { controllerAlive?: boolean; turnCompleted?: boolean; filesSaved?: boolean;
|
||||
controllerBootId?: string; hostHashUnrecorded?: boolean } = {}) {
|
||||
controllerBootId?: string; hostHashUnrecorded?: boolean; waitsOnChild?: boolean; statusVersion?: number;
|
||||
settledByPreviousRelease?: boolean } = {}) {
|
||||
const issueId = randomUUID(), runId = randomUUID(), leaseId = randomUUID();
|
||||
const finishedAt = new Date(Date.now() - 60_000);
|
||||
// The run's failure set Blocked at status version 6. A later version means
|
||||
// a newer decision owns the Blocked status.
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Set up visual reviews", status: "blocked",
|
||||
assigneeAgentId: agentId, statusVersion: 6 });
|
||||
assigneeAgentId: agentId, statusVersion: options.statusVersion ?? 6, blockedTransitionAt: finishedAt });
|
||||
if (options.waitsOnChild) {
|
||||
const childId = randomUUID();
|
||||
await db.insert(issues).values({ id: childId, companyId, title: "Capture baselines", status: "in_progress",
|
||||
assigneeAgentId: agentId, parentId: issueId });
|
||||
await db.insert(issueRelations).values({ companyId, issueId: childId, relatedIssueId: issueId, type: "blocks" });
|
||||
}
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "automation", issueId,
|
||||
nativeIssueId: issueId,
|
||||
status: "failed", errorCode: "native_session_cleanup_quarantined", runtimeMode: "native",
|
||||
@@ -123,15 +142,40 @@ suite("completed-turn cleanup holds on the shared ownership path", () => {
|
||||
.where(eq(workspaceOperations.id, saved!.id));
|
||||
await db.insert(workspaceOperations).values({ companyId, issueId, heartbeatRunId: runId, phase: "workspace_finalize",
|
||||
status: "succeeded", metadata: { owningService: "workspace_recovery", preservationOperationIds } });
|
||||
if (options.settledByPreviousRelease) {
|
||||
// As the previous release (#20628) left them: both holds settled as
|
||||
// restored, their replay block removed, the task still Blocked.
|
||||
for (const id of [quarantineHold!.id, filesHold!.id]) {
|
||||
const [row] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id));
|
||||
const { automaticRecovery: _removed, ...evidence } = row!.evidence;
|
||||
await db.update(issueRecoveryActions).set({ outcome: "restored", ownerType: "system",
|
||||
resolutionNote: "The run finished its turn and saved its work; its session cleanup no longer blocks the task.",
|
||||
evidence: { ...evidence, completedTurnCleanupSettlement: { runId, settledAt: new Date().toISOString(),
|
||||
ownerProof: "process_exited" } } }).where(eq(issueRecoveryActions.id, id));
|
||||
}
|
||||
}
|
||||
return { issueId, runId, quarantineHoldId: quarantineHold!.id, filesHoldId: filesHold!.id };
|
||||
}
|
||||
|
||||
const issue = async (id: string) => (await db.select().from(issues).where(eq(issues.id, id)))[0]!;
|
||||
/** Runs the continuation worker with a recording wake. */
|
||||
async function deliverContinuations() {
|
||||
const wakes: Array<{ agentId: string; issueId: unknown; previousRunId: unknown }> = [];
|
||||
await deliverReconciledExecutions(db, (async (wakeAgentId: string, request: { payload?: Record<string, unknown>;
|
||||
contextSnapshot?: Record<string, unknown> }) => {
|
||||
wakes.push({ agentId: wakeAgentId, issueId: request.payload?.issueId, previousRunId: request.contextSnapshot?.previousRunId });
|
||||
return null;
|
||||
}) as unknown as Parameters<typeof deliverReconciledExecutions>[1]);
|
||||
return wakes;
|
||||
}
|
||||
|
||||
const hold = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!;
|
||||
const pass = async () => (await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: true }))
|
||||
.kinds.find(entry => entry.kind === KIND);
|
||||
|
||||
it("settles both no-replay holds once the controller that held the cleanup gate is gone", async () => {
|
||||
const f = await quarantinedRun();
|
||||
// As on staging: the task also waits on an unfinished child task.
|
||||
const f = await quarantinedRun({ waitsOnChild: true });
|
||||
// As on staging: the files hold blocks the task with no Retry and no continuation.
|
||||
expect(await getExecutionBlocker(db, companyId, f.issueId)).toMatchObject({ recoveryActionId: f.filesHoldId,
|
||||
canRetry: false, canContinue: false });
|
||||
@@ -146,12 +190,85 @@ suite("completed-turn cleanup holds on the shared ownership path", () => {
|
||||
expect(await getExecutionBlocker(db, companyId, f.issueId)).toBeNull();
|
||||
// Settling replays nothing: no new run, and the task keeps its own wait.
|
||||
expect((await db.select().from(heartbeatRuns)).map(run => run.id)).toEqual([f.runId]);
|
||||
expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "blocked" });
|
||||
// The child's dependency wait keeps the task Blocked; its completion continues it.
|
||||
expect(await issue(f.issueId)).toMatchObject({ status: "blocked" });
|
||||
expect(await deliverContinuations()).toEqual([]);
|
||||
expect(await db.select().from(activityLog).where(eq(activityLog.action, "issue.recovery_action_resolved"))).toHaveLength(1);
|
||||
// Bounded: nothing is left to list.
|
||||
expect(await pass()).toMatchObject({ inProgress: 0 });
|
||||
}, 60_000);
|
||||
|
||||
// The staging dead end: the same quarantined completed turn, but the task has
|
||||
// no blocker of its own. Its Blocked status came only from the run's failure.
|
||||
it("releases the Blocked status the quarantined run set and continues the task once", async () => {
|
||||
const f = await quarantinedRun();
|
||||
expect(await pass()).toMatchObject({ inProgress: 1, settled: 1 });
|
||||
expect(await getExecutionBlocker(db, companyId, f.issueId)).toBeNull();
|
||||
expect(await issue(f.issueId)).toMatchObject({ status: "todo", blockedTransitionAt: null });
|
||||
// One durable continuation, owned by the hold that owned the Blocked status.
|
||||
const owner = await hold(f.quarantineHoldId);
|
||||
expect(owner.evidence).toMatchObject({ continuationDelivery: "pending",
|
||||
executionReconciliation: { runId: f.runId, providerStopped: true, actionOutcome: "completed" } });
|
||||
expect(owner.returnOwnerAgentId).toBe(agentId);
|
||||
expect((await hold(f.filesHoldId)).evidence.continuationDelivery).toBe("invalidated");
|
||||
expect(await deliverContinuations()).toEqual([{ agentId, issueId: f.issueId, previousRunId: f.runId }]);
|
||||
// Settling itself replays nothing, and the next pass lists nothing.
|
||||
expect((await db.select().from(heartbeatRuns)).map(run => run.id)).toEqual([f.runId]);
|
||||
expect(await pass()).toMatchObject({ inProgress: 0 });
|
||||
// Normal admission accepts the continuation exactly once, as a fresh session.
|
||||
const heartbeat = heartbeatService(db);
|
||||
await deliverReconciledExecutions(db, heartbeat.wakeup);
|
||||
await deliverReconciledExecutions(db, heartbeat.wakeup);
|
||||
const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.runId));
|
||||
expect(successors).toHaveLength(1);
|
||||
expect(await db.select().from(agentWakeupRequests)).toHaveLength(1);
|
||||
expect(successors[0]!.contextSnapshot).toMatchObject({ issueId: f.issueId, forceFreshSession: true,
|
||||
wakeReason: "issue_recovery_action_restored" });
|
||||
expect((await hold(f.quarantineHoldId)).evidence).toMatchObject({ continuationDelivery: "delivered",
|
||||
continuationRunId: successors[0]!.id });
|
||||
}, 60_000);
|
||||
|
||||
it("lets saved user input own the continuation instead of recording a second one", async () => {
|
||||
const f = await quarantinedRun();
|
||||
await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "on_demand", status: "deferred_issue_execution",
|
||||
requestedByActorType: "user", requestedByActorId: "board-user", payload: { issueId: f.issueId } });
|
||||
expect(await pass()).toMatchObject({ inProgress: 1, settled: 1 });
|
||||
expect(await issue(f.issueId)).toMatchObject({ status: "todo" });
|
||||
for (const id of [f.quarantineHoldId, f.filesHoldId])
|
||||
expect((await hold(id)).evidence.continuationDelivery).not.toBe("pending");
|
||||
expect(await deliverContinuations()).toEqual([]);
|
||||
expect(await pass()).toMatchObject({ inProgress: 0 });
|
||||
}, 60_000);
|
||||
|
||||
// Upgrade: rows exactly as the previous release left that task. Both holds are
|
||||
// already restored, so there is no blocker and no recovery action, but the
|
||||
// task is still Blocked with no Retry. The next pass after deploy continues it.
|
||||
it("continues a task whose holds the previous release settled but left Blocked", async () => {
|
||||
const f = await quarantinedRun({ settledByPreviousRelease: true });
|
||||
expect(await getExecutionBlocker(db, companyId, f.issueId)).toBeNull();
|
||||
expect(await issue(f.issueId)).toMatchObject({ status: "blocked" });
|
||||
|
||||
expect(await pass()).toMatchObject({ inProgress: 1, settled: 1 });
|
||||
expect(await issue(f.issueId)).toMatchObject({ status: "todo" });
|
||||
expect(await deliverContinuations()).toEqual([{ agentId, issueId: f.issueId, previousRunId: f.runId }]);
|
||||
expect(await pass()).toMatchObject({ inProgress: 0 });
|
||||
}, 60_000);
|
||||
|
||||
it("keeps a Blocked status that a newer decision owns and starts nothing", async () => {
|
||||
const f = await quarantinedRun({ statusVersion: 7 });
|
||||
expect(await pass()).toMatchObject({ inProgress: 1, settled: 1 });
|
||||
expect(await hold(f.quarantineHoldId)).toMatchObject({ outcome: "restored" });
|
||||
expect(await issue(f.issueId)).toMatchObject({ status: "blocked" });
|
||||
expect(await deliverContinuations()).toEqual([]);
|
||||
expect(await pass()).toMatchObject({ inProgress: 0 });
|
||||
}, 60_000);
|
||||
|
||||
it("keeps the Blocked status of previously settled holds while the closing controller is alive", async () => {
|
||||
const f = await quarantinedRun({ settledByPreviousRelease: true, controllerAlive: true });
|
||||
expect(await pass()).toMatchObject({ inProgress: 1, settled: 0 });
|
||||
expect(await issue(f.issueId)).toMatchObject({ status: "blocked" });
|
||||
}, 60_000);
|
||||
|
||||
it("proves a controller saved before its PID space was recorded gone through its stopped server boot", async () => {
|
||||
const bootId = randomUUID();
|
||||
await db.insert(controllerInstances).values({ bootId, processSpace: "old-container", pid: 7,
|
||||
|
||||
@@ -1,10 +1,16 @@
|
||||
import { and, eq, isNull, sql, type SQL } from "drizzle-orm";
|
||||
import { heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { and, eq, isNull, or, sql, type SQL } from "drizzle-orm";
|
||||
import { agentWakeupRequests, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import type { ExecutionReconciliation } from "@paperclipai/shared";
|
||||
import { logActivity } from "../activity-log.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
|
||||
import { issueService } from "../issues.js";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
import type { InProgressRecord, SettleOutcome } from "../recovery/ownership-registry.js";
|
||||
import type { OwnerProof } from "../owner-proof.js";
|
||||
import { OWNER_GONE_ATTESTATION_KEY } from "../recovery/owner-gone-attestation.js";
|
||||
import { holdOwnsBlockedStatus } from "../recovery-hold-blocked-status.js";
|
||||
import { notifyRoutineExecutionStatus } from "../routine-execution-status-notifications.js";
|
||||
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
|
||||
import { completedWorkspaceRecoveryCondition } from "../workspace-recovery-receipt.js";
|
||||
import { NATIVE_SESSION_CLEANUP_QUARANTINED } from "./completed-turn-cleanup.js";
|
||||
|
||||
@@ -30,6 +36,20 @@ import { NATIVE_SESSION_CLEANUP_QUARANTINED } from "./completed-turn-cleanup.js"
|
||||
* (an assigned person, an owner-gone attestation, a reconciliation or explicit
|
||||
* continuation) are never reopened. Every rule reads fields that earlier
|
||||
* releases already saved, so their holds settle without a migration.
|
||||
*
|
||||
* Settling the holds is not enough when the run's own failure set the task's
|
||||
* Blocked status: nothing else ever releases it, so the task rested Blocked
|
||||
* with no blocker, no Retry and no next action. In the same
|
||||
* transaction, a Blocked status that a settled hold of the run still owns
|
||||
* (`holdOwnsBlockedStatus`) is released to `todo`, and one durable
|
||||
* continuation is recorded on that hold (`continuationDelivery: "pending"`).
|
||||
* The execution-continuation worker delivers it as a fresh session. Saved user
|
||||
* input owns the continuation instead, so the task never starts two turns. A task
|
||||
* that still waits on unfinished blocker tasks stays Blocked; their completion
|
||||
* wakes it. A Blocked status that a newer decision owns is kept. Holds that the
|
||||
* previous release settled without this decision are listed again until it is
|
||||
* recorded (`completedTurnCleanupSettlement.blockedStatus`), so those tasks
|
||||
* continue on the first pass after deploy, without a migration.
|
||||
*/
|
||||
|
||||
const COMPLETED_TURN_CHECKPOINT = sql`run.runner_profile_json->'sessionCheckpoint'->'identity'->>'runId' = run.id::text
|
||||
@@ -52,16 +72,36 @@ function settleableHoldCondition(runId: SQL | string): SQL {
|
||||
)!;
|
||||
}
|
||||
|
||||
/** A hold of `runId` that this kind settled, in this or an earlier release,
|
||||
* with no recorded decision about the task's Blocked status yet. */
|
||||
function settledHoldAwaitingBlockedStatusCondition(runId: SQL | string): SQL {
|
||||
return and(
|
||||
sql`${issueRecoveryActions.evidence}->>'runId' = ${runId}::text`,
|
||||
sql`${issueRecoveryActions.evidence}->'completedTurnCleanupSettlement'->>'runId' = ${runId}::text`,
|
||||
sql`${issueRecoveryActions.evidence}->'completedTurnCleanupSettlement'->'blockedStatus' is null`,
|
||||
eq(issueRecoveryActions.status, "resolved"),
|
||||
eq(issueRecoveryActions.outcome, "restored"),
|
||||
)!;
|
||||
}
|
||||
|
||||
/** Coordinators of completed-turn cleanup quarantines that still have a
|
||||
* settleable no-replay hold. */
|
||||
* settleable no-replay hold, or a settled hold whose task is still Blocked
|
||||
* with no recorded decision about that status. */
|
||||
export function completedTurnCleanupHoldWhere(): SQL {
|
||||
return and(
|
||||
isNull(nativeRunFinalizations.leaseOwner),
|
||||
sql`exists (select 1 from heartbeat_runs run where run.company_id = ${nativeRunFinalizations.companyId}
|
||||
and run.id = ${nativeRunFinalizations.runId} and run.runtime_mode = 'native' and run.status = 'failed'
|
||||
and run.error_code = ${NATIVE_SESSION_CLEANUP_QUARANTINED} and ${COMPLETED_TURN_CHECKPOINT})`,
|
||||
sql`exists (select 1 from ${issueRecoveryActions} where ${issueRecoveryActions.companyId} = ${nativeRunFinalizations.companyId}
|
||||
and ${settleableHoldCondition(sql`${nativeRunFinalizations.runId}`)})`,
|
||||
or(
|
||||
sql`exists (select 1 from ${issueRecoveryActions} where ${issueRecoveryActions.companyId} = ${nativeRunFinalizations.companyId}
|
||||
and ${settleableHoldCondition(sql`${nativeRunFinalizations.runId}`)})`,
|
||||
sql`exists (select 1 from ${issueRecoveryActions} join ${issues} on ${issues.id} = ${issueRecoveryActions.sourceIssueId}
|
||||
and ${issues.companyId} = ${issueRecoveryActions.companyId}
|
||||
where ${issueRecoveryActions.companyId} = ${nativeRunFinalizations.companyId}
|
||||
and ${issueRecoveryActions.sourceIssueId} = ${nativeRunFinalizations.issueId} and ${issues.status} = 'blocked'
|
||||
and ${settledHoldAwaitingBlockedStatusCondition(sql`${nativeRunFinalizations.runId}`)})`,
|
||||
),
|
||||
)!;
|
||||
}
|
||||
|
||||
@@ -89,8 +129,11 @@ export async function settleCompletedTurnCleanupHolds(db: Db, record: InProgress
|
||||
.from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, companyId),
|
||||
eq(nativeRunFinalizations.runId, record.recordId), completedTurnCleanupHoldWhere())).for("update");
|
||||
if (!coordinator) return null;
|
||||
const [run] = await tx.select({ agentId: heartbeatRuns.agentId }).from(heartbeatRuns)
|
||||
const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, companyId),
|
||||
eq(issues.id, coordinator.issueId))).for("update");
|
||||
const [run] = await tx.select().from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, coordinator.runId)));
|
||||
if (!issue || !run) return null;
|
||||
const holds = await tx.update(issueRecoveryActions).set({
|
||||
outcome: "restored", ownerType: "system", updatedAt: now,
|
||||
resolutionNote: "The run finished its turn and saved its work; its session cleanup no longer blocks the task.",
|
||||
@@ -99,20 +142,103 @@ export async function settleCompletedTurnCleanupHolds(db: Db, record: InProgress
|
||||
})}::jsonb`,
|
||||
}).where(and(eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, coordinator.issueId),
|
||||
settleableHoldCondition(coordinator.runId))).returning({ id: issueRecoveryActions.id });
|
||||
if (!holds.length) return null;
|
||||
await logActivity(tx as unknown as Db, { companyId, actorType: "system", actorId: "system",
|
||||
action: "issue.recovery_action_resolved", entityType: "issue", entityId: coordinator.issueId,
|
||||
agentId: run?.agentId ?? null, runId: coordinator.runId,
|
||||
details: { reason: "native_completed_turn_cleanup_settled", recoveryActionIds: holds.map(hold => hold.id),
|
||||
ownerProof: context.proof.reason },
|
||||
});
|
||||
return coordinator.runId;
|
||||
if (holds.length) {
|
||||
await logActivity(tx as unknown as Db, { companyId, actorType: "system", actorId: "system",
|
||||
action: "issue.recovery_action_resolved", entityType: "issue", entityId: coordinator.issueId,
|
||||
agentId: run.agentId, runId: coordinator.runId,
|
||||
details: { reason: "native_completed_turn_cleanup_settled", recoveryActionIds: holds.map(hold => hold.id),
|
||||
ownerProof: context.proof.reason },
|
||||
});
|
||||
}
|
||||
const decided = await decideBlockedStatus(tx as unknown as Db, { companyId, issue, run, now });
|
||||
return holds.length || decided ? coordinator.runId : null;
|
||||
});
|
||||
if (!settled) return { settled: false, reason: "hold_changed_since_proof" };
|
||||
// Settling starts no turn. Saved input and the task's own wait continue
|
||||
// through normal admission, which still checks every other gate.
|
||||
// Settling starts no turn itself. Saved input, the recorded continuation and
|
||||
// the task's own wait continue through normal admission, which still checks
|
||||
// every other gate.
|
||||
await settledHoldResumer?.({ companyId, runId: settled }).catch(error => {
|
||||
logger.warn({ err: error, runId: settled }, "settled completed-turn cleanup hold could not resume saved input");
|
||||
});
|
||||
return { settled: true, reason: context.proof.reason };
|
||||
}
|
||||
|
||||
type BlockedStatusDecision = "released" | "dependency_wait" | "not_owned";
|
||||
|
||||
/** Recorded with the continuation. Admission of a reconciled continuation
|
||||
* requires workspace repair evidence after a restore failure; the settled
|
||||
* holds already required the run's files to be verified as saved. */
|
||||
const CONTINUATION_EVIDENCE = "The run completed its provider turn and saved its work before its session cleanup was quarantined. "
|
||||
+ "The controller that held the cleanup gate is proven gone, so the task continues in a fresh session.";
|
||||
const WORKSPACE_EVIDENCE = "Workspace recovery verified this run's saved files, or the run had no unsaved files; "
|
||||
+ "the continuation starts from that saved workspace.";
|
||||
|
||||
/**
|
||||
* Decides, once, what happens to the task's Blocked status after the run's
|
||||
* holds are settled, and records that decision on every settled hold of the
|
||||
* run so the kind does not list them again. Runs in the settlement
|
||||
* transaction with the issue row locked. Returns whether a decision was
|
||||
* recorded.
|
||||
*/
|
||||
async function decideBlockedStatus(tx: Db, input: { companyId: string; now: Date;
|
||||
issue: typeof issues.$inferSelect; run: typeof heartbeatRuns.$inferSelect }): Promise<boolean> {
|
||||
const { companyId, issue, run, now } = input;
|
||||
if (issue.status !== "blocked") return false;
|
||||
const settledHolds = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, companyId),
|
||||
eq(issueRecoveryActions.sourceIssueId, issue.id), settledHoldAwaitingBlockedStatusCondition(run.id)))
|
||||
.orderBy(issueRecoveryActions.createdAt, issueRecoveryActions.id).for("update");
|
||||
if (!settledHolds.length) return false;
|
||||
// The hold that recorded the run's failure block is its receipt. Only holds
|
||||
// saved before receipts existed fall back to the block's timing.
|
||||
const receipts = settledHolds.filter(hold => hold.evidence.nativeFailureBlock !== undefined);
|
||||
const owner = (receipts.length ? receipts : settledHolds).find(hold => holdOwnsBlockedStatus(hold, issue, run)) ?? null;
|
||||
// Only the task's current assignee continues it.
|
||||
const ownsTask = Boolean(owner && issue.assigneeAgentId && issue.assigneeAgentId === run.agentId);
|
||||
// The shared dependency rule: an unfinished blocker task keeps the task
|
||||
// Blocked, and its completion wakes it (issue_blockers_resolved).
|
||||
const readiness = ownsTask ? await issueService(tx).listDependencyReadiness(companyId, [issue.id], tx) : null;
|
||||
const decision: BlockedStatusDecision = !ownsTask ? "not_owned"
|
||||
: readiness?.get(issue.id)?.isDependencyReady === true ? "released" : "dependency_wait";
|
||||
// Saved user input continues the task itself through normal admission (the
|
||||
// registered resumer). Recording a second continuation would start another
|
||||
// turn after it, so the input owns the continuation, as in
|
||||
// continueFromLastSavedWorkspace. A saved Retry that dispatch declined for
|
||||
// its requester is shown with a Retry that any user with access can press
|
||||
// (retry_authority_denied), so it is not a dead end either; a second,
|
||||
// automatic continuation could race that Retry once access returns.
|
||||
const [waitingInput] = decision === "released" ? await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests)
|
||||
.where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
eq(agentWakeupRequests.requestedByActorType, "user"), sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`))
|
||||
.limit(1) : [];
|
||||
const continuationHoldId = decision === "released" && !waitingInput ? owner!.id : null;
|
||||
// Register the work before the writes it describes.
|
||||
if (decision === "released") {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
|
||||
await notifyStatusCardIssueChange(tx, companyId);
|
||||
await notifyRoutineExecutionStatus(tx, issue.id, companyId);
|
||||
}
|
||||
const reconciliation: ExecutionReconciliation = { runId: run.id, providerStopped: true, actionOutcome: "completed",
|
||||
outcomeEvidence: CONTINUATION_EVIDENCE, workspaceRepairEvidence: WORKSPACE_EVIDENCE };
|
||||
for (const hold of settledHolds) {
|
||||
const continuation = hold.id === continuationHoldId;
|
||||
const marked = sql`jsonb_set(${issueRecoveryActions.evidence}, '{completedTurnCleanupSettlement,blockedStatus}', ${JSON.stringify(decision)}::jsonb)`;
|
||||
await tx.update(issueRecoveryActions).set({ updatedAt: now,
|
||||
...(continuation ? {
|
||||
returnOwnerAgentId: run.agentId, nextAction: "The task continues in a fresh session.",
|
||||
evidence: sql`${marked} || ${JSON.stringify({
|
||||
executionReconciliation: { ...reconciliation, actorId: "completed-turn-cleanup", recordedAt: now.toISOString() },
|
||||
continuationDelivery: "pending" })}::jsonb`,
|
||||
} : { evidence: marked }),
|
||||
}).where(and(eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.id, hold.id)));
|
||||
}
|
||||
if (decision !== "released") return true;
|
||||
await tx.update(issues).set({ status: "todo", unblockDescriptor: null, blockedTransitionAt: null,
|
||||
blockedOwnerNotifiedAt: null, updatedAt: now }).where(and(eq(issues.companyId, companyId), eq(issues.id, issue.id)));
|
||||
await logActivity(tx, { companyId, actorType: "system", actorId: "system", action: "issue.updated",
|
||||
entityType: "issue", entityId: issue.id, agentId: run.agentId, runId: run.id,
|
||||
details: { status: "todo", previousStatus: "blocked", reason: "native_completed_turn_cleanup_continued",
|
||||
recoveryActionId: owner!.id, continuation: continuationHoldId ? "recorded" : "saved_input", _previous: { status: "blocked" } } });
|
||||
return true;
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
import type { heartbeatRuns, issueRecoveryActions, issues } from "@paperclipai/db";
|
||||
import { nativeWorkspaceFinalizationBlockSchema } from "./native-runtime/preserved-native-result-finalization.js";
|
||||
|
||||
const object = (value: unknown): Record<string, unknown> =>
|
||||
value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
|
||||
/**
|
||||
* Whether the task's Blocked status still belongs to this recovery 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, and settling the
|
||||
* hold must keep it.
|
||||
*/
|
||||
export function holdOwnsBlockedStatus(
|
||||
action: Pick<typeof issueRecoveryActions.$inferSelect, "evidence" | "createdAt">,
|
||||
issue: Pick<typeof issues.$inferSelect, "id" | "status" | "statusVersion" | "lastStatusDecisionId" | "blockedTransitionAt">,
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "finishedAt">,
|
||||
): boolean {
|
||||
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);
|
||||
}
|
||||
@@ -24,7 +24,7 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
|
||||
import { hasRecoveredCleanupStopReceipt, hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
import { abandonUnverifiedRemoteWorkspaceEpoch } from "./remote-workspace-epoch-abandonment.js";
|
||||
import { hasStopOnlyCleanup, prepareHistoricalSandboxStopVerification, prepareSandboxStopAndRetain, restoreSandboxStopAfterFailedRestart, settleStopOnlyCleanup } from "./sandbox-stop-and-retain.js";
|
||||
import { nativeWorkspaceFinalizationBlockSchema } from "./native-runtime/preserved-native-result-finalization.js";
|
||||
import { holdOwnsBlockedStatus } from "./recovery-hold-blocked-status.js";
|
||||
import { withNativeWorkspaceFinalizationOwnership } from "./native-runtime/native-workspace-finalization-ownership.js";
|
||||
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";
|
||||
@@ -303,27 +303,6 @@ 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> {
|
||||
|
||||
Reference in new issue
Block a user