mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(recovery): move stranded legacy subtasks off a parent's held workspace (#20314)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Agents split work into subtasks. Since #19860, every subtask gets its own execution workspace from the project base, and a subtask that an older release bound to its parent's workspace moves to its own workspace on its next admission > - Some legacy subtasks never get a next admission. Another task held the shared workspace with unsaved files, so their runs failed setup before any provider work. Stranded-work recovery then blocked them and handed them to the board > - On staging, a parent and two of its children stayed blocked for four days this way. The children could have run in their own workspace at any time > - This pull request makes the scheduled recovery-action sweep move such a child to its own workspace and continue it, with no person > - The benefit is that a legacy subtask never stays blocked on a workspace that it should not share, and the parent keeps its own workspace ## Linked Issues or Issue Description Refs #19860 (subtasks get their own workspace; legacy children move at admission) Refs #20295 (the held workspace's exhausted save now settles automatically) Refs #20273 (stall wakes wait while a workspace hold refuses a run) **What happened?** A parent task and three children shared the parent's remote workspace, because an older release created the children before #19860. One child's ended run held the workspace with unsaved files. The other two children were woken, but each run failed setup with "The workspace still has an unfinished writer or save operation." (`setup_failed`, `executionRecovery.kind=bootstrap`, `providerWorkStarted=false`). Stranded-work recovery then set each child to Blocked with a board-owned escalation (`board_escalation_no_takeover_v1`) that asked a board operator to retry. The #19860 move happens only at the next admission, and a blocked child gets no admission. **Expected behavior** A subtask that failed only because of a legacy shared-workspace binding continues in its own workspace from the project base automatically. The parent and the shared workspace stay unchanged. **Steps to reproduce** 1. With a release before #19860, create a parent task and children that share its remote workspace. 2. Let one child's run end with unsaved files whose save is not settled. 3. Wake another child. Its run fails setup on the workspace hold, and stranded-work recovery blocks it with a board escalation. 4. See that the blocked child is never moved to its own workspace and never runs again. **Root cause** `server/src/services/execution-workspaces.ts` (`releaseAncestorWorkspaceBindingAtAdmission`) moves a legacy child only when a run of the child is admitted. `server/src/services/recovery/service.ts` (`reconcileStrandedAssignedIssues`) ignores Blocked tasks, so nothing admits a child that a setup failure on the shared hold blocked. ## What Changed - `server/src/services/execution-workspaces.ts`: extract the binding release into `releaseAncestorWorkspaceBinding(tx, …)`, which runs inside the caller's transaction. Admission uses it unchanged (same guards: no other running run of the task, no unsettled workspace recovery, only a relative's workspace). - New `server/src/services/recovery/legacy-workspace-rehome.ts` (`rehomeStrandedLegacyWorkspaceTasks`). It selects a board-owned stranded escalation whose latest run failed setup on the workspace writer hold before any provider work, for a Blocked task bound to a relative's workspace. In one transaction it releases the binding, resolves the escalation with the outcome `legacy_shared_workspace_rehomed`, sets the task to `todo`, records a pending wake (`legacyWorkspaceRehome.wake: "pending"`), and records `issue.legacy_workspace_rehomed` system activity. After commit it wakes the assignee once (`issue_recovery_action_restored`, with a fresh session) and marks the wake `delivered` only when a run is created. A wake lost to a crash, refused (`null`), or failed stays pending and backs off (5 minutes, doubling, at most an hour); the pass returns the earliest owed wake as its retry deadline, so the recovery worker schedules it. Candidates are paged by action id. - Guards: the Blocked status must come from that escalation (set after the failed run, and no status decision at or after the block), the shared dependency-readiness rule must allow work (cancelled blockers and done blockers that still save files keep it blocked), the failed run must belong to the current assignee, and the binding release must be allowed. Otherwise nothing changes. - `reconcileActiveRecoveryActions` (the company-scoped sweep that the scheduled issue-recovery worker runs for each company with an active recovery action) runs this pass first. `recoveryActionSweepCondition()` also selects a company whose moved task still owes its wake. - `doc/execution-semantics.md`: documents the move and what a moved child sees. ### What a moved child sees A fresh workspace from the project base, exactly like a new top-level task in the project. It does not get the parent's changes. It does not get its own earlier files from the shared workspace either, because nothing of its work ran there (its runs failed setup before any provider work). The child that still holds the shared workspace with unsaved files is not moved by this change: its save recovery owns the hold (#20295 settles it automatically), its files stay in the shared workspace and in that recovery's saved-work receipts, and it moves on its first admission after that recovery settles (the #19860 rule). ### Previous-release story (`rules/upgrade-safe-run-state.mdc`) - **Migrated, no schema change.** The change reads escalations and task bindings that older releases saved, and it changes only the task's binding row, the task status, and the escalation. It does not change saved run identity (native execution input, leases, checkpoints, sessions). The regression test seeds the exact staging shape (Nov 4–5 evidence fields, `retryReason: workspace_busy`, `latestRunErrorCode: setup_failed`). - **Mid-run deploy.** A child with a running run or an unsettled workspace recovery is not moved (the same guards as the admission release). ## Verification - Red (the move disabled), `npx vitest run src/__tests__/legacy-workspace-rehome.test.ts`: 1 failed, 2 passed. The staging-shape case failed because the stranded child stayed Blocked and bound to the parent's workspace. - Green: 8 of 8 pass. All cases run the scheduled path: the company is selected by `recoveryActionSweepCondition()` and the pass is `recoveryService(...).reconcileActiveRecoveryActions({ companyId })` with the real heartbeat wake. - Staging shape: a parent and 3 children share the parent's workspace. One child holds it with a dirty writer run. Two children are blocked by board escalations after setup failed on that hold. After the stranded-work sweep, both children are unbound from the shared workspace, their escalations are resolved by the system with text that asks no person to act, and each child is woken exactly once with a fresh session. Each child runs once through the real heartbeat path outside the shared workspace and completes. The parent and the writer child keep their bindings, the shared workspace row is unchanged, and nothing runs for them. A later pass changes nothing. - A move whose wake was lost to a crash: the scheduled worker still selects the company, delivers the wake once, and the child runs and completes. - A refused wake (`null`) or a thrown enqueue stays pending, backs off, does not hold up other wakes, is reported to the worker as the next deadline, and keeps the company selected after a restart; once accepted, it settles. - Thirty rejected escalations do not hide a later eligible one (paging). - A child with a cancelled blocker stays blocked with its board escalation. - A child whose last status decision came before the block moves; one decided after the block stays blocked. - A stranded child whose setup failed for another reason stays blocked with its board escalation. - A task bound to the workspace it created (the parent) is never moved. - Related suites pass: `subtask-workspace-independence`, `heartbeat-workspace-busy`, `heartbeat-remote-workspace-busy`, `heartbeat-held-comment-wake`, `recovery-deadline-acceptance`, `heartbeat-retry-scheduling`, `task-workspace-binding` (278 tests), and after the review changes `issue-blocker-stall-wakeups`, `issue-recovery-deadlines`, `issue-recovery-work-lifecycle`, `recovery-durable-wait-projection`, `owner-gone-attestation` (124 tests with the two workspace suites). - `pnpm verify`: exit 2 ("broader verification required", because `doc/execution-semantics.md` is a global CI input). Required CI supplies the broad coverage. - `npx tsc --noEmit -p server/tsconfig.json`: passes. ## Risks - The pass changes a Blocked task to `todo` only under narrow guards, and only for a task bound to a relative's workspace whose last run failed setup on that workspace's writer hold. A Blocked status from another decision, or an unresolved blocker, keeps the task blocked. - The moved child starts again from the project base. This is the product rule for subtasks. Nothing of its own work is dropped, because its runs failed before any provider work. - The pass runs inside the existing active recovery action sweep and selects at most 25 escalations and 25 pending wakes per pass. > 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 API checks) and extended reasoning. ## 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
1 parent
76254ead22
commit
f56b097fbd
6 files changed
+569
-31
No files matched your search
@@ -716,6 +716,28 @@ its task tree created: the task that created the workspace keeps it, and the
|
||||
relative bound to it (for example a parent that selected its child's workspace)
|
||||
moves to its own on its next admission.
|
||||
|
||||
A legacy child can have no next admission: when another task's writer held the
|
||||
shared workspace, the child's run failed setup before any provider work, and
|
||||
stranded-work recovery blocked it with a board-owned escalation. The
|
||||
active recovery action sweep moves such a child without a person, in one
|
||||
transaction, when the escalation's latest run failed setup on that workspace
|
||||
hold, the child's Blocked status came from that escalation (no status decision
|
||||
at or after the block), the shared dependency-readiness rule allows work, and
|
||||
the binding release above is allowed. The child's binding is released, the
|
||||
escalation resolves with the outcome `legacy_shared_workspace_rehomed`
|
||||
(`issue.legacy_workspace_rehomed` system activity), the child returns to
|
||||
`todo`, and a pending wake is recorded. The assignee is then woken once with a
|
||||
fresh session. Only a created run settles the wake: a wake lost to a crash or
|
||||
refused by the heartbeat stays pending, backs off (5 minutes, doubling, at most
|
||||
an hour), and is retried at that deadline while the child is still an assigned
|
||||
`todo`. What the moved child
|
||||
sees is a fresh workspace from the project base: no parent changes, and none of
|
||||
its own earlier files from the shared workspace. The child that still holds the
|
||||
shared workspace with unsaved files is not moved; its save recovery owns the
|
||||
hold, its files stay in the shared workspace and its saved-work receipts, and it
|
||||
moves on its first admission after that recovery settles. The parent keeps its
|
||||
workspace throughout.
|
||||
|
||||
### Blockers (`blockedByIssueIds`)
|
||||
|
||||
This is dependency semantics.
|
||||
|
||||
@@ -0,0 +1,305 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { activityLog, agents, agentWakeupRequests, authUsers, companies, companyMemberships, createDb, executionWorkspaces,
|
||||
heartbeatRuns, issueRecoveryActions, issueRelations, issues, statusDecisions } from "@paperclipai/db";
|
||||
import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.js";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { REMOTE_WORKSPACE_WRITER_PENDING_MESSAGE } from "../services/remote-workspace-writer-hold.js";
|
||||
import { LEGACY_WORKSPACE_REHOME_OUTCOME, rehomeStrandedLegacyWorkspaceTasks } from "../services/recovery/legacy-workspace-rehome.js";
|
||||
import { recoveryActionSweepCondition } from "../services/recovery/blocker-stall-wakeups.js";
|
||||
import { recoveryService } from "../services/recovery/service.js";
|
||||
import { recoveryPostgresSupport, startRecoveryPostgres } from "./helpers/recovery-postgres.js";
|
||||
|
||||
const describePostgres = recoveryPostgresSupport.supported ? describe : describe.skip;
|
||||
if (!recoveryPostgresSupport.supported) console.warn(`Skipping legacy workspace rehome tests: ${recoveryPostgresSupport.reason}`);
|
||||
|
||||
// Recovery text never asks a person to inspect or fix a sandbox.
|
||||
const adminText = /sandbox|environment|administrator|board member|board operator|owner must|inspect the/i;
|
||||
|
||||
/**
|
||||
* The staging shape (Nov 4-5): a parent task and three children that an older
|
||||
* release bound to the parent's remote workspace (before #19860). One child's
|
||||
* ended run still holds that workspace with unsaved files. The other two
|
||||
* children's runs failed setup on that hold before any provider work, and
|
||||
* stranded-work recovery blocked them with a board-owned escalation.
|
||||
*/
|
||||
describePostgres("stranded subtasks bound to a parent's held workspace", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startRecoveryPostgres>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let heartbeat: ReturnType<typeof heartbeatService>;
|
||||
const adapterType = "legacy_rehome_test";
|
||||
const dispatched: Array<{ issueId: string; context: Record<string, unknown> }> = [];
|
||||
const oldHome = process.env.PAPERCLIP_HOME;
|
||||
let home: string;
|
||||
|
||||
beforeAll(async () => {
|
||||
temporary = await startRecoveryPostgres("legacy-workspace-rehome-");
|
||||
db = createDb(temporary.connectionString);
|
||||
home = await mkdtemp(path.join(tmpdir(), "paperclip-legacy-rehome-"));
|
||||
process.env.PAPERCLIP_HOME = home;
|
||||
heartbeat = heartbeatService(db);
|
||||
registerServerAdapter({
|
||||
type: adapterType, supportsLocalAgentJwt: false,
|
||||
execute: async input => {
|
||||
const issueId = String(input.context.issueId);
|
||||
dispatched.push({ issueId, context: input.context as Record<string, unknown> });
|
||||
await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId));
|
||||
return { exitCode: 0, signal: null, timedOut: false, resultJson: {} };
|
||||
},
|
||||
testEnvironment: async () => ({ adapterType, status: "pass", checks: [], testedAt: new Date().toISOString() }),
|
||||
});
|
||||
}, 240_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await heartbeat?.drainActiveRunExecutions();
|
||||
unregisterServerAdapter(adapterType);
|
||||
await db?.$client.end({ timeout: 0 });
|
||||
await temporary?.cleanup();
|
||||
if (oldHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = oldHome;
|
||||
if (home) await rm(home, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
async function seedSharedFamily() {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), userId = `owner-${companyId}`;
|
||||
await db.insert(authUsers).values({ id: userId, name: "Owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companies).values({ id: companyId, name: "Legacy family", defaultResponsibleUserId: userId, issuePrefix: `L${companyId.slice(0, 6)}` });
|
||||
await db.insert(companyMemberships).values({ companyId, principalId: userId, principalType: "user", status: "active", membershipRole: "owner" });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType, status: "idle", adapterConfig: {} });
|
||||
const [parent] = await db.insert(issues).values({ companyId, title: "Landing page", status: "blocked", assigneeAgentId: agentId,
|
||||
responsibleUserId: userId, executionWorkspacePreference: "reuse_existing", issueNumber: 1 }).returning();
|
||||
const writerRunId = randomUUID();
|
||||
const [workspace] = await db.insert(executionWorkspaces).values({ companyId, sourceIssueId: parent.id, name: "Parent workspace",
|
||||
mode: "shared_workspace", strategyType: "project_primary", status: "active", providerType: "cloud_sandbox",
|
||||
metadata: { remoteWorkspace: { schema: "paperclip.remote-workspace.v1", epoch: 2, snapshotId: randomUUID(),
|
||||
source: { kind: "empty" }, writerRunId, dirty: true, location: null } } }).returning();
|
||||
await db.update(issues).set({ executionWorkspaceId: workspace.id }).where(eq(issues.id, parent.id));
|
||||
const child = async (title: string, issueNumber: number) => (await db.insert(issues).values({ companyId, title, parentId: parent.id,
|
||||
status: "blocked", assigneeAgentId: agentId, responsibleUserId: userId, executionWorkspaceId: workspace.id, executionWorkspacePreference: "reuse_existing",
|
||||
issueNumber, createdAt: new Date("2026-11-04T23:56:00Z") }).returning())[0];
|
||||
const [hero, pricing, writer] = [await child("Hero copy", 2), await child("Pricing table", 3), await child("FAQ", 4)];
|
||||
for (const blocker of [hero, pricing, writer])
|
||||
await db.insert(issueRelations).values({ companyId, issueId: blocker.id, relatedIssueId: parent.id, type: "blocks" });
|
||||
// The writer child's ended run holds the workspace (its save is still pending).
|
||||
await db.insert(heartbeatRuns).values({ id: writerRunId, companyId, agentId, issueId: writer.id, status: "failed",
|
||||
finishedAt: new Date(Date.now() - 4 * 86_400_000), contextSnapshot: { issueId: writer.id } });
|
||||
const strand = async (issueId: string, error = REMOTE_WORKSPACE_WRITER_PENDING_MESSAGE) => {
|
||||
const finishedAt = new Date(Date.now() - 4 * 86_400_000);
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, issueId, status: "failed", errorCode: "setup_failed",
|
||||
error, startedAt: new Date(finishedAt.getTime() - 90_000), finishedAt,
|
||||
resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } },
|
||||
contextSnapshot: { issueId, wakeReason: "workspace_busy_retry" } }).returning();
|
||||
const blockedAt = new Date(finishedAt.getTime() + 900);
|
||||
await db.update(issues).set({ blockedTransitionAt: blockedAt, statusVersion: 2 }).where(eq(issues.id, issueId));
|
||||
const [action] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "stranded_assigned_issue",
|
||||
status: "active", ownerType: "board", previousOwnerAgentId: agentId, returnOwnerAgentId: agentId, cause: "stranded_assigned_issue",
|
||||
fingerprint: `source_scoped_recovery:${companyId}:${issueId}:stranded_assigned_issue`,
|
||||
nextAction: "Board operator: inspect the evidence, repair the runtime if appropriate, then explicitly retry the original owner, reassign, or intentionally resolve the task.",
|
||||
evidence: { latestRunId: run.id, retryReason: "workspace_busy", recoveryCause: "stranded_assigned_issue",
|
||||
routingPolicy: "board_escalation_no_takeover_v1", sourceIssueId: issueId, previousStatus: "in_progress",
|
||||
latestRunStatus: "failed", latestIssueStatus: "in_progress", latestRunErrorCode: "setup_failed" },
|
||||
createdAt: blockedAt, updatedAt: blockedAt }).returning();
|
||||
return { runId: run.id, actionId: action.id };
|
||||
};
|
||||
return { companyId, agentId, workspaceId: workspace.id, parent, hero, pricing, writer, strand };
|
||||
}
|
||||
|
||||
/** The pass the scheduled issue-recovery worker runs for a company with active recovery actions. */
|
||||
async function scheduledRecoveryPass(companyId: string) {
|
||||
const [selected] = await db.selectDistinct({ companyId: issueRecoveryActions.companyId }).from(issueRecoveryActions)
|
||||
.where(and(recoveryActionSweepCondition(), eq(issueRecoveryActions.companyId, companyId)));
|
||||
if (!selected) return false;
|
||||
await recoveryService(db, { enqueueWakeup: (agentId, options) => heartbeat.wakeup(agentId, options as never) })
|
||||
.reconcileActiveRecoveryActions({ companyId });
|
||||
return true;
|
||||
}
|
||||
|
||||
const issue = async (id: string) => (await db.select().from(issues).where(eq(issues.id, id)))[0]!;
|
||||
const action = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!;
|
||||
|
||||
it("moves each stranded child to its own workspace and continues it, leaving the parent's workspace alone", async () => {
|
||||
const f = await seedSharedFamily();
|
||||
const hero = await f.strand(f.hero.id), pricing = await f.strand(f.pricing.id);
|
||||
const workspaceBefore = (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)))[0];
|
||||
|
||||
expect(await scheduledRecoveryPass(f.companyId)).toBe(true);
|
||||
|
||||
for (const [child, stranded] of [[f.hero, hero], [f.pricing, pricing]] as const) {
|
||||
const moved = await issue(child.id);
|
||||
// Its next run realizes its own workspace from the project base; nothing of the parent's is carried over.
|
||||
expect(moved).toMatchObject({ executionWorkspacePreference: null, workspaceBindingRevision: child.workspaceBindingRevision + 1 });
|
||||
expect(moved.executionWorkspaceId).not.toBe(f.workspaceId);
|
||||
const resolved = await action(stranded.actionId);
|
||||
expect(resolved).toMatchObject({ status: "resolved", outcome: "restored", ownerType: "system",
|
||||
resolutionNote: LEGACY_WORKSPACE_REHOME_OUTCOME, returnOwnerAgentId: f.agentId });
|
||||
expect(resolved.evidence).toMatchObject({
|
||||
legacyWorkspaceRehome: { previousExecutionWorkspaceId: f.workspaceId, runId: stranded.runId } });
|
||||
expect(String(resolved.nextAction)).not.toMatch(adminText);
|
||||
}
|
||||
// The parent keeps its own workspace, and the writer child keeps its saved-work binding.
|
||||
expect(await issue(f.parent.id)).toMatchObject({ executionWorkspaceId: f.workspaceId, status: "blocked" });
|
||||
expect(await issue(f.writer.id)).toMatchObject({ executionWorkspaceId: f.workspaceId, status: "blocked" });
|
||||
expect((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)))[0]).toEqual(workspaceBefore);
|
||||
const rehomed = await db.select().from(activityLog).where(eq(activityLog.action, "issue.legacy_workspace_rehomed"));
|
||||
expect(rehomed.filter(row => row.companyId === f.companyId).map(row => row.entityId).sort())
|
||||
.toEqual([f.hero.id, f.pricing.id].sort());
|
||||
|
||||
// Each moved child is woken exactly once, with a fresh session, and runs outside the shared workspace.
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
for (const child of [f.hero, f.pricing]) {
|
||||
const wakes = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId),
|
||||
eq(agentWakeupRequests.reason, "issue_recovery_action_restored")));
|
||||
expect(wakes.filter(wake => wake.payload?.issueId === child.id)).toHaveLength(1);
|
||||
const turns = dispatched.filter(turn => turn.issueId === child.id);
|
||||
expect(turns).toHaveLength(1);
|
||||
expect(turns[0].context).toMatchObject({ forceFreshSession: true, wakeReason: "issue_recovery_action_restored" });
|
||||
expect(turns[0].context.executionWorkspaceId ?? null).not.toBe(f.workspaceId);
|
||||
expect((await issue(child.id)).status).toBe("done");
|
||||
}
|
||||
// Nothing ran for the parent or the writer child in the shared workspace.
|
||||
expect(dispatched.filter(turn => [f.parent.id, f.writer.id].includes(turn.issueId))).toEqual([]);
|
||||
// A later pass changes nothing.
|
||||
expect((await rehomeStrandedLegacyWorkspaceTasks(db, null, { companyId: f.companyId })).rehomed).toEqual([]);
|
||||
expect((await issue(f.hero.id)).executionWorkspaceId).not.toBe(f.workspaceId);
|
||||
}, 60_000);
|
||||
|
||||
it("delivers a move's wake that a crash left pending", async () => {
|
||||
const f = await seedSharedFamily();
|
||||
const hero = await f.strand(f.hero.id);
|
||||
// The move commits, then the process stops before it wakes the assignee.
|
||||
expect((await rehomeStrandedLegacyWorkspaceTasks(db, null, { companyId: f.companyId })).rehomed).toEqual([f.hero.id]);
|
||||
expect((await action(hero.actionId)).evidence).toMatchObject({ legacyWorkspaceRehome: { wake: "pending" } });
|
||||
expect(await issue(f.hero.id)).toMatchObject({ status: "todo", executionWorkspaceId: null });
|
||||
|
||||
// The scheduled worker still selects the company and delivers the owed wake once.
|
||||
expect(await scheduledRecoveryPass(f.companyId)).toBe(true);
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
expect((await action(hero.actionId)).evidence).toMatchObject({ legacyWorkspaceRehome: { wake: "delivered" } });
|
||||
expect(dispatched.filter(turn => turn.issueId === f.hero.id)).toHaveLength(1);
|
||||
expect((await issue(f.hero.id)).status).toBe("done");
|
||||
}, 60_000);
|
||||
|
||||
it("keeps a refused or failing wake pending and backs it off without holding up the others", async () => {
|
||||
const f = await seedSharedFamily();
|
||||
const hero = await f.strand(f.hero.id), pricing = await f.strand(f.pricing.id);
|
||||
expect((await rehomeStrandedLegacyWorkspaceTasks(db, null, { companyId: f.companyId })).rehomed.sort()).toEqual([f.hero.id, f.pricing.id].sort());
|
||||
const woken: string[] = [];
|
||||
// The heartbeat refuses the hero's wake (it returns null, as for disabled on-demand wakes).
|
||||
const refusing = async (_agentId: string, options: { payload: Record<string, unknown> }) => {
|
||||
if (options.payload.issueId === f.hero.id) return null;
|
||||
woken.push(String(options.payload.issueId));
|
||||
return { id: randomUUID() };
|
||||
};
|
||||
|
||||
const first = await rehomeStrandedLegacyWorkspaceTasks(db, refusing as never, { companyId: f.companyId });
|
||||
expect(woken).toEqual([f.pricing.id]);
|
||||
const backedOff = (await action(hero.actionId)).evidence.legacyWorkspaceRehome as Record<string, unknown>;
|
||||
expect(backedOff).toMatchObject({ wake: "pending", wakeFailures: 1 });
|
||||
expect(Number(backedOff.wakeRetryAt)).toBeGreaterThan(Date.now());
|
||||
// The pass reports that deadline to the worker, so the retry is scheduled.
|
||||
expect(first.nextWakeRetryAt).toBe(Number(backedOff.wakeRetryAt));
|
||||
expect((await action(pricing.actionId)).evidence).toMatchObject({ legacyWorkspaceRehome: { wake: "delivered" } });
|
||||
// The company stays selected by the scheduled worker while a wake is owed, also after a restart.
|
||||
const [selected] = await db.selectDistinct({ companyId: issueRecoveryActions.companyId }).from(issueRecoveryActions)
|
||||
.where(and(recoveryActionSweepCondition(), eq(issueRecoveryActions.companyId, f.companyId)));
|
||||
expect(selected).toBeDefined();
|
||||
|
||||
// Not retried before its time, and a thrown wake backs off too.
|
||||
await rehomeStrandedLegacyWorkspaceTasks(db, refusing as never, { companyId: f.companyId });
|
||||
expect(woken).toEqual([f.pricing.id]);
|
||||
await db.update(issueRecoveryActions).set({ evidence: { ...(await action(hero.actionId)).evidence,
|
||||
legacyWorkspaceRehome: { ...backedOff, wakeRetryAt: Date.now() - 1 } } }).where(eq(issueRecoveryActions.id, hero.actionId));
|
||||
await rehomeStrandedLegacyWorkspaceTasks(db, (async () => { throw new Error("enqueue failed"); }) as never, { companyId: f.companyId });
|
||||
expect((await action(hero.actionId)).evidence).toMatchObject({ legacyWorkspaceRehome: { wake: "pending", wakeFailures: 2 } });
|
||||
|
||||
// Once due and accepted, the wake settles.
|
||||
await db.update(issueRecoveryActions).set({ evidence: { ...(await action(hero.actionId)).evidence,
|
||||
legacyWorkspaceRehome: { ...backedOff, wakeFailures: 2, wakeRetryAt: Date.now() - 1 } } }).where(eq(issueRecoveryActions.id, hero.actionId));
|
||||
const last = await rehomeStrandedLegacyWorkspaceTasks(db, (async (_agentId: string, options: { payload: Record<string, unknown> }) => {
|
||||
woken.push(String(options.payload.issueId)); return { id: randomUUID() }; }) as never, { companyId: f.companyId });
|
||||
expect(woken).toEqual([f.pricing.id, f.hero.id]);
|
||||
expect((await action(hero.actionId)).evidence).toMatchObject({ legacyWorkspaceRehome: { wake: "delivered" } });
|
||||
expect(last.nextWakeRetryAt).toBeNull();
|
||||
}, 60_000);
|
||||
|
||||
it("pages past escalations the guards reject", async () => {
|
||||
const f = await seedSharedFamily();
|
||||
// Thirty escalations that match the query but fail a guard (another setup error).
|
||||
for (let index = 0; index < 30; index++) {
|
||||
const [other] = await db.insert(issues).values({ companyId: f.companyId, title: `Other ${index}`, parentId: f.parent.id,
|
||||
status: "blocked", assigneeAgentId: f.agentId, executionWorkspaceId: f.workspaceId, issueNumber: 100 + index }).returning();
|
||||
await f.strand(other.id, "Project access denied");
|
||||
}
|
||||
const hero = await f.strand(f.hero.id);
|
||||
// The eligible escalation sorts after every rejected one.
|
||||
await db.update(issueRecoveryActions).set({ id: "ffffffff-ffff-4fff-bfff-ffffffffffff" }).where(eq(issueRecoveryActions.id, hero.actionId));
|
||||
expect((await rehomeStrandedLegacyWorkspaceTasks(db, null, { companyId: f.companyId })).rehomed).toEqual([f.hero.id]);
|
||||
}, 60_000);
|
||||
|
||||
it("keeps a stranded child whose dependency still forbids work", async () => {
|
||||
const f = await seedSharedFamily();
|
||||
const stranded = await f.strand(f.hero.id);
|
||||
const [cancelled] = await db.insert(issues).values({ companyId: f.companyId, title: "Cancelled prerequisite", status: "cancelled",
|
||||
issueNumber: 9 }).returning();
|
||||
await db.insert(issueRelations).values({ companyId: f.companyId, issueId: cancelled.id, relatedIssueId: f.hero.id, type: "blocks" });
|
||||
|
||||
await scheduledRecoveryPass(f.companyId);
|
||||
|
||||
expect(await issue(f.hero.id)).toMatchObject({ status: "blocked", executionWorkspaceId: f.workspaceId });
|
||||
expect(await action(stranded.actionId)).toMatchObject({ status: "active", ownerType: "board" });
|
||||
}, 60_000);
|
||||
|
||||
it("moves a child whose last status decision came before the block, and keeps one decided after it", async () => {
|
||||
const f = await seedSharedFamily();
|
||||
const earlier = await f.strand(f.hero.id), later = await f.strand(f.pricing.id);
|
||||
const decide = async (issueId: string, at: Date) => {
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, issueId, status: "succeeded",
|
||||
finishedAt: at, contextSnapshot: { issueId } }).returning();
|
||||
// Only the decision's time matters here; its assessment chain is out of scope.
|
||||
const decisionId = randomUUID();
|
||||
await db.transaction(async tx => {
|
||||
await tx.execute(sql`set local session_replication_role = replica`);
|
||||
await tx.insert(statusDecisions).values({ id: decisionId, companyId: f.companyId, issueId, runId: run.id,
|
||||
assessmentId: randomUUID(), decisionVersion: 1, policyVersion: "test", fromStatus: "todo", toStatus: "in_progress",
|
||||
reasonCode: "test", decisionJson: {}, decisionDigest: "test", createdAt: at });
|
||||
});
|
||||
await db.update(issues).set({ lastStatusDecisionId: decisionId }).where(eq(issues.id, issueId));
|
||||
};
|
||||
const blockedAt = (await issue(f.hero.id)).blockedTransitionAt!.getTime();
|
||||
await decide(f.hero.id, new Date(blockedAt - 86_400_000));
|
||||
await decide(f.pricing.id, new Date(blockedAt + 5_000));
|
||||
|
||||
await scheduledRecoveryPass(f.companyId);
|
||||
|
||||
expect((await issue(f.hero.id)).executionWorkspaceId).not.toBe(f.workspaceId);
|
||||
expect(await action(earlier.actionId)).toMatchObject({ status: "resolved" });
|
||||
expect(await issue(f.pricing.id)).toMatchObject({ status: "blocked", executionWorkspaceId: f.workspaceId });
|
||||
expect(await action(later.actionId)).toMatchObject({ status: "active" });
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
}, 60_000);
|
||||
|
||||
it("keeps a stranded child whose setup failed for another reason", async () => {
|
||||
const f = await seedSharedFamily();
|
||||
const other = await f.strand(f.hero.id, "Project access denied");
|
||||
|
||||
expect(await scheduledRecoveryPass(f.companyId)).toBe(true);
|
||||
|
||||
expect(await issue(f.hero.id)).toMatchObject({ status: "blocked", executionWorkspaceId: f.workspaceId });
|
||||
expect(await action(other.actionId)).toMatchObject({ status: "active", ownerType: "board" });
|
||||
}, 60_000);
|
||||
|
||||
it("keeps a task bound to its own workspace", async () => {
|
||||
const f = await seedSharedFamily();
|
||||
// The parent itself created the workspace; it never moves.
|
||||
const own = await f.strand(f.parent.id);
|
||||
await db.delete(issueRelations).where(eq(issueRelations.relatedIssueId, f.parent.id));
|
||||
|
||||
expect(await scheduledRecoveryPass(f.companyId)).toBe(true);
|
||||
|
||||
expect(await issue(f.parent.id)).toMatchObject({ status: "blocked", executionWorkspaceId: f.workspaceId });
|
||||
expect(await action(own.actionId)).toMatchObject({ status: "active" });
|
||||
}, 60_000);
|
||||
});
|
||||
@@ -1501,6 +1501,45 @@ async function hasPendingTaskWorkspaceRecovery(tx: DbTransaction, input: {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Clears a task's binding to a relative's execution workspace that an older
|
||||
* release created, so its next run realizes its own workspace from the project
|
||||
* base. The relative and the shared workspace are never changed. Does nothing
|
||||
* while another run of the task (other than `runId`) is running or its earlier
|
||||
* run still owns workspace finalization or recovery. Runs inside the caller's
|
||||
* transaction; returns whether the binding was released.
|
||||
*/
|
||||
export async function releaseAncestorWorkspaceBinding(tx: DbTransaction, input: { companyId: string; issueId: string; runId?: string }) {
|
||||
const [task] = await tx.select().from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId))).for("update");
|
||||
if (!task?.executionWorkspaceId || task.workspacePendingSelection) return false;
|
||||
if (!await bindsRelativesExecutionWorkspace(tx, { companyId: input.companyId, executionWorkspaceId: task.executionWorkspaceId, task })) {
|
||||
return false;
|
||||
}
|
||||
const [holder] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, input.companyId), input.runId ? ne(heartbeatRuns.id, input.runId) : undefined,
|
||||
or(eq(heartbeatRuns.nativeIssueId, input.issueId),
|
||||
sql`coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId') = ${input.issueId}`),
|
||||
eq(heartbeatRuns.status, "running"),
|
||||
)).limit(1);
|
||||
if (holder) return false;
|
||||
if (await hasPendingTaskWorkspaceRecovery(tx, { companyId: input.companyId, issueId: input.issueId, excludeRunId: input.runId })) return false;
|
||||
const selection = task.workspaceSelection;
|
||||
await notifyStatusCardIssueChange(tx, task.companyId);
|
||||
await tx.update(issues).set({
|
||||
executionWorkspaceId: null,
|
||||
executionWorkspacePreference: null,
|
||||
// Without a project workspace the old settings described the relative's
|
||||
// task files; the project's defaults apply instead, as for a top-level task.
|
||||
executionWorkspaceSettings: task.projectWorkspaceId
|
||||
? buildIndependentTaskWorkspaceSettings(task.executionWorkspaceSettings)
|
||||
: null,
|
||||
...(selection?.selection.kind === "existing" ? { workspaceSelection: null } : {}),
|
||||
workspaceBindingRevision: task.workspaceBindingRevision + 1,
|
||||
updatedAt: new Date(),
|
||||
}).where(eq(issues.id, task.id));
|
||||
return true;
|
||||
}
|
||||
|
||||
// Compaction trash older than this belongs to no compaction in progress: the
|
||||
// move and the guard finish in one short transaction.
|
||||
const WORKSPACE_COMPACTION_TRASH_STALE_MS = 60 * 60 * 1000;
|
||||
@@ -1746,36 +1785,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
||||
.from(issues).innerJoin(executionWorkspaces, eq(executionWorkspaces.id, issues.executionWorkspaceId))
|
||||
.where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId))).limit(1);
|
||||
if (!candidate || candidate.sourceIssueId === input.issueId) return false;
|
||||
return db.transaction(async tx => {
|
||||
const [task] = await tx.select().from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId))).for("update");
|
||||
if (!task?.executionWorkspaceId || task.workspacePendingSelection) return false;
|
||||
if (!await bindsRelativesExecutionWorkspace(tx, { companyId: input.companyId, executionWorkspaceId: task.executionWorkspaceId, task })) {
|
||||
return false;
|
||||
}
|
||||
const [holder] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, input.companyId), ne(heartbeatRuns.id, input.runId),
|
||||
or(eq(heartbeatRuns.nativeIssueId, input.issueId),
|
||||
sql`coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId') = ${input.issueId}`),
|
||||
eq(heartbeatRuns.status, "running"),
|
||||
)).limit(1);
|
||||
if (holder) return false;
|
||||
if (await hasPendingTaskWorkspaceRecovery(tx, { ...input, excludeRunId: input.runId })) return false;
|
||||
const selection = task.workspaceSelection;
|
||||
await notifyStatusCardIssueChange(tx, task.companyId);
|
||||
await tx.update(issues).set({
|
||||
executionWorkspaceId: null,
|
||||
executionWorkspacePreference: null,
|
||||
// Without a project workspace the old settings described the relative's
|
||||
// task files; the project's defaults apply instead, as for a top-level task.
|
||||
executionWorkspaceSettings: task.projectWorkspaceId
|
||||
? buildIndependentTaskWorkspaceSettings(task.executionWorkspaceSettings)
|
||||
: null,
|
||||
...(selection?.selection.kind === "existing" ? { workspaceSelection: null } : {}),
|
||||
workspaceBindingRevision: task.workspaceBindingRevision + 1,
|
||||
updatedAt: new Date(),
|
||||
}).where(eq(issues.id, task.id));
|
||||
return true;
|
||||
});
|
||||
return db.transaction(tx => releaseAncestorWorkspaceBinding(tx, input));
|
||||
}
|
||||
|
||||
async function bindTaskWorkspace(companyId: string, issueId: string, workspaceId: string, patch: TaskWorkspaceBindingPatch = {}) {
|
||||
|
||||
@@ -91,6 +91,10 @@ export function recoveryActionSweepCondition() {
|
||||
eq(issueRecoveryActions.ownerType, "board"),
|
||||
sql`exists (select 1 from ${issues} stalled where stalled.id = ${issueRecoveryActions.sourceIssueId}
|
||||
and stalled.company_id = ${issueRecoveryActions.companyId} and stalled.status = 'blocked')`),
|
||||
// A legacy subtask moved to its own workspace whose wake is still owed.
|
||||
// The pass returns its retry time, which schedules the company again.
|
||||
and(eq(issueRecoveryActions.status, "resolved"),
|
||||
sql`${issueRecoveryActions.evidence}#>>'{legacyWorkspaceRehome,wake}' = 'pending'`),
|
||||
)!;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
import { and, asc, eq, gt, inArray, isNull, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, issueRecoveryActions, issues, statusDecisions, type Db } from "@paperclipai/db";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
import { persistActivity, publishActivity } from "../activity-log.js";
|
||||
import { releaseAncestorWorkspaceBinding } from "../execution-workspaces.js";
|
||||
import { issueService } from "../issues.js";
|
||||
import { REMOTE_WORKSPACE_WRITER_PENDING_MESSAGE } from "../remote-workspace-writer-hold.js";
|
||||
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
|
||||
|
||||
export const LEGACY_WORKSPACE_REHOME_SCHEMA = "paperclip.legacy-workspace-rehome.v1";
|
||||
export const LEGACY_WORKSPACE_REHOME_OUTCOME = "legacy_shared_workspace_rehomed";
|
||||
const STRANDED_KIND = "stranded_assigned_issue";
|
||||
const BOARD_ESCALATION_POLICY = "board_escalation_no_takeover_v1";
|
||||
const PAGE_SIZE = 25;
|
||||
const REHOME_WAKE_RETRY_BASE_MS = 5 * 60_000;
|
||||
const REHOME_WAKE_RETRY_MAX_MS = 60 * 60_000;
|
||||
// A block set by the stranded escalation lands right after its failed run.
|
||||
const BLOCK_OWNERSHIP_SLACK_MS = 60_000;
|
||||
|
||||
export const LEGACY_WORKSPACE_REHOME_NOTE = "This task was set up in its parent's workspace by an earlier release, and that "
|
||||
+ "workspace was held by another task's unsaved files, so the task could not start. Paperclip moved the task to its own "
|
||||
+ "workspace from the project base and continues it there. No work was lost: nothing ran in the shared workspace.";
|
||||
|
||||
type RehomeWakeup = (agentId: string, options: {
|
||||
source: "automation"; triggerDetail: "system"; reason: string; payload: Record<string, unknown>;
|
||||
requestedByActorType: "system"; requestedByActorId: string; contextSnapshot: Record<string, unknown>;
|
||||
issueStateGuard: { statuses: string[]; assigneeAgentId: string };
|
||||
}) => Promise<unknown>;
|
||||
|
||||
/**
|
||||
* A subtask that an older release bound to its parent's (or another
|
||||
* relative's) execution workspace could not start while another task held
|
||||
* that workspace. Its run failed setup before any provider work, and
|
||||
* stranded-work recovery handed the blocked task to the board. That failure
|
||||
* belongs to the legacy binding, which the next admission would release anyway
|
||||
* (#19860). This pass does the release now, in one transaction: the task gets
|
||||
* its own workspace from the project base, the board escalation resolves with
|
||||
* a named outcome, the task returns to `todo`, and a pending wake is recorded.
|
||||
* After commit the assignee is woken once, like a board restore; a wake lost
|
||||
* to a crash stays pending and the next pass delivers it. The parent and the shared
|
||||
* workspace are never changed. Bounded and idempotent: a resolved escalation
|
||||
* is never revisited, and a failed guard leaves the task as it is. Returns the
|
||||
* IDs of the tasks that were moved and when the next owed wake is due.
|
||||
*/
|
||||
export async function rehomeStrandedLegacyWorkspaceTasks(db: Db, wake: RehomeWakeup | null,
|
||||
scope: { companyId?: string; issueId?: string } = {}) {
|
||||
const rehomed: string[] = [];
|
||||
// Page by action id, so escalations the guards reject never hide later eligible ones.
|
||||
let cursor: string | null = null;
|
||||
for (;;) {
|
||||
const candidates: Array<{ actionId: string; companyId: string; issueId: string }> = await db.select({
|
||||
actionId: issueRecoveryActions.id, companyId: issueRecoveryActions.companyId, issueId: issueRecoveryActions.sourceIssueId,
|
||||
}).from(issueRecoveryActions)
|
||||
.innerJoin(issues, and(eq(issues.companyId, issueRecoveryActions.companyId), eq(issues.id, issueRecoveryActions.sourceIssueId)))
|
||||
.where(and(
|
||||
scope.companyId ? eq(issueRecoveryActions.companyId, scope.companyId) : undefined,
|
||||
scope.issueId ? eq(issueRecoveryActions.sourceIssueId, scope.issueId) : undefined,
|
||||
cursor ? gt(issueRecoveryActions.id, cursor) : undefined,
|
||||
eq(issueRecoveryActions.kind, STRANDED_KIND), inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
||||
sql`${issueRecoveryActions.evidence}->>'routingPolicy' = ${BOARD_ESCALATION_POLICY}`,
|
||||
sql`${issueRecoveryActions.evidence}->>'latestRunErrorCode' = 'setup_failed'`,
|
||||
eq(issues.status, "blocked"), isNull(issues.hiddenAt), isNull(issues.assigneeUserId),
|
||||
sql`${issues.assigneeAgentId} is not null`, sql`${issues.executionWorkspaceId} is not null`,
|
||||
)).orderBy(asc(issueRecoveryActions.id)).limit(PAGE_SIZE);
|
||||
for (const candidate of candidates) {
|
||||
let moved: Awaited<ReturnType<typeof rehomeOne>>;
|
||||
try {
|
||||
moved = await db.transaction(tx => rehomeOne(tx as unknown as Db, candidate));
|
||||
} catch (error) {
|
||||
logger.warn({ err: error, issueId: candidate.issueId }, "Legacy shared-workspace task stays blocked until the next sweep");
|
||||
continue;
|
||||
}
|
||||
if (!moved) continue;
|
||||
rehomed.push(candidate.issueId);
|
||||
try { publishActivity(moved.activity.publication); } catch { logger.warn({ issueId: candidate.issueId }, "Recovery activity delivery deferred"); }
|
||||
}
|
||||
if (candidates.length < PAGE_SIZE) break;
|
||||
cursor = candidates.at(-1)!.actionId;
|
||||
}
|
||||
// The move records its wake as pending in the same transaction. Deliver every
|
||||
// due pending wake, including one that a crash after an earlier commit left behind.
|
||||
const nextWakeRetryAt = wake ? await deliverPendingRehomeWakes(db, wake, scope) : null;
|
||||
return { rehomed, nextWakeRetryAt };
|
||||
}
|
||||
|
||||
const wakeRetryAt = sql`coalesce((${issueRecoveryActions.evidence}#>>'{legacyWorkspaceRehome,wakeRetryAt}')::bigint, 0)`;
|
||||
|
||||
/** Delivers due pending wakes; returns when the earliest one still pending is due (epoch ms), or null. */
|
||||
async function deliverPendingRehomeWakes(db: Db, wake: RehomeWakeup, scope: { companyId?: string; issueId?: string }) {
|
||||
const pendingIn = and(
|
||||
scope.companyId ? eq(issueRecoveryActions.companyId, scope.companyId) : undefined,
|
||||
scope.issueId ? eq(issueRecoveryActions.sourceIssueId, scope.issueId) : undefined,
|
||||
eq(issueRecoveryActions.kind, STRANDED_KIND), eq(issueRecoveryActions.status, "resolved"),
|
||||
sql`${issueRecoveryActions.evidence}#>>'{legacyWorkspaceRehome,wake}' = 'pending'`);
|
||||
const pending = await db.select({ actionId: issueRecoveryActions.id, companyId: issueRecoveryActions.companyId,
|
||||
issueId: issueRecoveryActions.sourceIssueId, agentId: issues.assigneeAgentId,
|
||||
wakeFailures: sql<number>`coalesce((${issueRecoveryActions.evidence}#>>'{legacyWorkspaceRehome,wakeFailures}')::int, 0)`,
|
||||
}).from(issueRecoveryActions)
|
||||
.innerJoin(issues, and(eq(issues.companyId, issueRecoveryActions.companyId), eq(issues.id, issueRecoveryActions.sourceIssueId)))
|
||||
// A wake that was refused waits for its retry time, so it cannot hold up the others.
|
||||
.where(and(pendingIn, sql`${wakeRetryAt} <= ${Date.now()}::bigint`))
|
||||
.orderBy(wakeRetryAt, issueRecoveryActions.id).limit(PAGE_SIZE);
|
||||
for (const row of pending) {
|
||||
// A task that changed since the move no longer owes this wake.
|
||||
const [task] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId }).from(issues)
|
||||
.where(and(eq(issues.companyId, row.companyId), eq(issues.id, row.issueId)));
|
||||
const owed = task?.status === "todo" && Boolean(task.assigneeAgentId) && task.assigneeAgentId === row.agentId;
|
||||
let accepted = !owed;
|
||||
try {
|
||||
// Only a created run settles the wake; a refusal (for example a paused
|
||||
// company or disabled on-demand wakes) returns null and is retried.
|
||||
if (owed) accepted = Boolean(await wake(row.agentId!, {
|
||||
source: "automation", triggerDetail: "system", reason: "issue_recovery_action_restored",
|
||||
payload: { issueId: row.issueId, recoveryActionId: row.actionId, mutation: "recovery_action_resolution" },
|
||||
requestedByActorType: "system", requestedByActorId: "legacy-workspace-rehome",
|
||||
// A session saved in the shared workspace never resumes in the new one.
|
||||
contextSnapshot: { issueId: row.issueId, taskId: row.issueId, wakeReason: "issue_recovery_action_restored",
|
||||
source: "issue.recovery_action_resolution", recoveryActionId: row.actionId, forceFreshSession: true },
|
||||
issueStateGuard: { statuses: ["todo"], assigneeAgentId: row.agentId! },
|
||||
}));
|
||||
} catch (error) {
|
||||
logger.warn({ err: error, issueId: row.issueId }, "Moved task wake failed; it stays pending");
|
||||
}
|
||||
const owner = and(eq(issueRecoveryActions.companyId, row.companyId), eq(issueRecoveryActions.id, row.actionId),
|
||||
sql`${issueRecoveryActions.evidence}#>>'{legacyWorkspaceRehome,wake}' = 'pending'`);
|
||||
if (accepted) {
|
||||
await db.update(issueRecoveryActions).set({ evidence: sql`jsonb_set(${issueRecoveryActions.evidence},
|
||||
'{legacyWorkspaceRehome,wake}', ${JSON.stringify(owed ? "delivered" : "not_owed")}::jsonb)` }).where(owner);
|
||||
continue;
|
||||
}
|
||||
// Back off, up to an hour, so a wake that keeps being refused yields to the others.
|
||||
const failures = row.wakeFailures + 1;
|
||||
const retryAt = Date.now() + Math.min(REHOME_WAKE_RETRY_MAX_MS, REHOME_WAKE_RETRY_BASE_MS * 2 ** (failures - 1));
|
||||
await db.update(issueRecoveryActions).set({ evidence: sql`jsonb_set(jsonb_set(${issueRecoveryActions.evidence},
|
||||
'{legacyWorkspaceRehome,wakeFailures}', ${JSON.stringify(failures)}::jsonb),
|
||||
'{legacyWorkspaceRehome,wakeRetryAt}', ${JSON.stringify(retryAt)}::jsonb)` }).where(owner);
|
||||
}
|
||||
const [next] = await db.select({ at: sql<string>`min(${wakeRetryAt})` }).from(issueRecoveryActions).where(pendingIn);
|
||||
return next?.at === null || next?.at === undefined ? null : Math.max(Number(next.at), Date.now());
|
||||
}
|
||||
|
||||
async function rehomeOne(tx: Db, input: { actionId: string; companyId: string; issueId: string }) {
|
||||
// Lock order: task, then recovery action (the binding release locks the task again).
|
||||
const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update");
|
||||
const [action] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, input.companyId),
|
||||
eq(issueRecoveryActions.id, input.actionId))).for("update");
|
||||
if (!issue || !action || issue.status !== "blocked" || issue.hiddenAt || issue.assigneeUserId || !issue.assigneeAgentId
|
||||
|| !issue.executionWorkspaceId || action.sourceIssueId !== issue.id || action.kind !== STRANDED_KIND
|
||||
|| !["active", "escalated"].includes(action.status) || action.evidence.routingPolicy !== BOARD_ESCALATION_POLICY) return null;
|
||||
const latestRunId = action.evidence.latestRunId;
|
||||
if (typeof latestRunId !== "string") return null;
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId),
|
||||
eq(heartbeatRuns.id, latestRunId)));
|
||||
// Only a setup failure on a held workspace, before any provider work, by the current assignee.
|
||||
const recovery = (run?.resultJson?.executionRecovery ?? null) as { kind?: unknown; providerWorkStarted?: unknown } | null;
|
||||
if (!run || run.agentId !== issue.assigneeAgentId || run.status !== "failed" || run.errorCode !== "setup_failed"
|
||||
|| run.error !== REMOTE_WORKSPACE_WRITER_PENDING_MESSAGE || recovery?.kind !== "bootstrap" || recovery.providerWorkStarted !== false
|
||||
|| (run.nativeIssueId ?? run.issueId ?? run.contextSnapshot?.issueId) !== issue.id) return null;
|
||||
// The Blocked status belongs to this escalation: set after that run, and no blocker keeps it.
|
||||
const blockedAt = issue.blockedTransitionAt?.getTime();
|
||||
if (blockedAt === undefined || !run.finishedAt || blockedAt < run.finishedAt.getTime() - 1_000
|
||||
|| blockedAt > action.updatedAt.getTime() + BLOCK_OWNERSHIP_SLACK_MS) return null;
|
||||
// A status decision after the block owns the status now; an earlier one does not matter.
|
||||
if (issue.lastStatusDecisionId) {
|
||||
const [decision] = await tx.select({ createdAt: statusDecisions.createdAt }).from(statusDecisions).where(and(
|
||||
eq(statusDecisions.companyId, input.companyId), eq(statusDecisions.id, issue.lastStatusDecisionId)));
|
||||
if (!decision || decision.createdAt.getTime() >= blockedAt) return null;
|
||||
}
|
||||
// The shared dependency rule: any blocker that still forbids work keeps the task blocked.
|
||||
const readiness = await issueService(tx).listDependencyReadiness(input.companyId, [issue.id], tx);
|
||||
if (readiness.get(issue.id)?.isDependencyReady !== true) return null;
|
||||
const previousExecutionWorkspaceId = issue.executionWorkspaceId;
|
||||
if (!await releaseAncestorWorkspaceBinding(tx as never, { companyId: input.companyId, issueId: issue.id })) return null;
|
||||
const now = new Date();
|
||||
await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "restored", resolvedAt: now, updatedAt: now,
|
||||
ownerType: "system", ownerAgentId: null, ownerUserId: null, returnOwnerAgentId: issue.assigneeAgentId,
|
||||
resolutionNote: LEGACY_WORKSPACE_REHOME_OUTCOME, nextAction: LEGACY_WORKSPACE_REHOME_NOTE, wakePolicy: null, monitorPolicy: null,
|
||||
evidence: { ...action.evidence, legacyWorkspaceRehome: { schema: LEGACY_WORKSPACE_REHOME_SCHEMA, previousExecutionWorkspaceId,
|
||||
runId: run.id, recordedAt: now.toISOString(), wake: "pending" } },
|
||||
}).where(and(eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.id, action.id)));
|
||||
await tx.update(issues).set({ status: "todo", unblockDescriptor: null, blockedTransitionAt: null,
|
||||
blockedOwnerNotifiedAt: null, updatedAt: now }).where(and(eq(issues.companyId, input.companyId), eq(issues.id, issue.id)));
|
||||
await notifyStatusCardIssueChange(tx, input.companyId);
|
||||
const activity = await persistActivity(tx, { companyId: input.companyId, actorType: "system", actorId: "legacy-workspace-rehome",
|
||||
action: "issue.legacy_workspace_rehomed", entityType: "issue", entityId: issue.id, runId: run.id,
|
||||
details: { recoveryActionId: action.id, previousExecutionWorkspaceId, status: "todo", note: LEGACY_WORKSPACE_REHOME_NOTE,
|
||||
_previous: { status: "blocked" } } });
|
||||
return { activity, agentId: issue.assigneeAgentId };
|
||||
}
|
||||
@@ -180,6 +180,7 @@ import {
|
||||
import { withRecoveryContext } from "./status-only-context.js";
|
||||
import { currentRunFailureText } from "../native-runtime/native-exact-resume-failure.js";
|
||||
import { isAutomaticRecoverySuppressedByPauseHold } from "./pause-hold-guard.js";
|
||||
import { rehomeStrandedLegacyWorkspaceTasks } from "./legacy-workspace-rehome.js";
|
||||
import {
|
||||
collectDispositionRepairSourceState,
|
||||
readDispositionRepairDurableWaitingState,
|
||||
@@ -3470,6 +3471,11 @@ export function recoveryService(
|
||||
}
|
||||
|
||||
async function reconcileActiveRecoveryActions(opts?: { companyId?: string }) {
|
||||
// A blocked subtask that an older release bound to a relative's held
|
||||
// workspace moves to its own workspace and continues (its board escalation
|
||||
// is an active recovery action, so this company-scoped sweep reaches it).
|
||||
const rehome = await rehomeStrandedLegacyWorkspaceTasks(db, deps.enqueueWakeup, { companyId: opts?.companyId })
|
||||
.catch(error => { logger.warn({ err: error }, "Legacy shared-workspace rehome deferred to the next sweep"); return null; });
|
||||
const rows = await db
|
||||
.select({
|
||||
action: issueRecoveryActions,
|
||||
@@ -3641,8 +3647,10 @@ export function recoveryService(
|
||||
// A child or blocker stalled on a board-owned action: tell whoever waits on it, once per stall.
|
||||
const stalls = await wakeWaitersOfStalledIssues(db, deps.enqueueWakeup as BlockerStallWakeup, { companyId: opts?.companyId });
|
||||
// A failed delivery is reported so the recovery worker retries this pass.
|
||||
// The earliest owed wake of a moved legacy subtask is a deadline of this pass too.
|
||||
const retryAts = [stalls.retryAt, rehome?.nextWakeRetryAt].filter((at): at is number => typeof at === "number");
|
||||
return { ...result, blockerStallWakes: stalls.woken, blockerStallWakeFailures: stalls.failed,
|
||||
blockerStallRetryAt: stalls.retryAt };
|
||||
blockerStallRetryAt: retryAts.length ? Math.min(...retryAts) : null };
|
||||
}
|
||||
|
||||
async function escalateDispositionRepair(input: {
|
||||
|
||||
Reference in new issue
Block a user