diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index defd7e6b3e2..3ab9a5f8091 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -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. diff --git a/server/src/__tests__/legacy-workspace-rehome.test.ts b/server/src/__tests__/legacy-workspace-rehome.test.ts new file mode 100644 index 00000000000..d8025a9dbdc --- /dev/null +++ b/server/src/__tests__/legacy-workspace-rehome.test.ts @@ -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>; + let db: ReturnType; + let heartbeat: ReturnType; + const adapterType = "legacy_rehome_test"; + const dispatched: Array<{ issueId: string; context: Record }> = []; + 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 }); + 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 }) => { + 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; + 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 }) => { + 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); +}); diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 62c85a4381d..019f9973e61 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -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 = {}) { diff --git a/server/src/services/recovery/blocker-stall-wakeups.ts b/server/src/services/recovery/blocker-stall-wakeups.ts index cb44233719d..f87d1e16aa8 100644 --- a/server/src/services/recovery/blocker-stall-wakeups.ts +++ b/server/src/services/recovery/blocker-stall-wakeups.ts @@ -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'`), )!; } diff --git a/server/src/services/recovery/legacy-workspace-rehome.ts b/server/src/services/recovery/legacy-workspace-rehome.ts new file mode 100644 index 00000000000..1e20517e7bb --- /dev/null +++ b/server/src/services/recovery/legacy-workspace-rehome.ts @@ -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; + requestedByActorType: "system"; requestedByActorId: string; contextSnapshot: Record; + issueStateGuard: { statuses: string[]; assigneeAgentId: string }; +}) => Promise; + +/** + * 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>; + 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`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`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 }; +} diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index f085bbb0d3c..0255f72bc0d 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -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: {