From 7c7d2d8d2cfca96aa956b85d086f067bc9fd8880 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 17 Oct 2026 22:57:10 -0500 Subject: [PATCH] fix(runtime): continue from saved sources after workspace loss (#16690) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The runtime must preserve saved work and prevent duplicate execution. > - Older cleanup could destroy a sandbox before its workspace export finished. > - The existing recovery controls require either the original sandbox or recovered original files. > - Surviving attachments can support a rebuild, but they do not prove that missing edits were recovered. > - This pull request adds an explicit reconstruction decision and one fresh continuation through the normal execution gates. > - Operators can continue useful work while the original failed run remains truthful. ## Linked Issues or Issue Description **What happened?** A historical failed export can have an accepted result, a prepared workspace descriptor, and a confirmed destroyed sandbox. Its notice can also say `new_source_execution_path` and disappear. The task then has neither a truthful original-file repair path nor a supported reconstruction path. **Expected behavior** Keep the missing-file obligation visible. Let an authorized operator choose surviving task attachments and record lost work and external action outcomes. Start one fresh continuation after current execution checks. Keep the old result and failed export intact. **Steps to reproduce** 1. Seed a failed native export with a prepared descriptor and an exact destroyed-lease receipt. 2. Resolve its notice with the historical `new_source_execution_path` bookkeeping value. 3. Read the task. The baseline hides the hold even though no source reconciliation exists. 4. With this change, choose **Continue from saved sources** and submit a reconstruction decision. Related public work: #16451 preserves healthy warm sessions. #16583 prevents destructive cleanup of unexported work. This change handles historical source loss that those fixes cannot undo. Searches of open and closed work found no equivalent reconstruction endpoint. ## What Changed - Add a board-only reconstruction endpoint with `runtime:manage` and task visibility checks. - Validate the exact run, result, descriptor and destroyed lease. Reject active copyback, competing owners, newer work, stale task status, and foreign attachments. - Record attachment hashes, missing work, and external action outcomes. Preserve the original failed run, accepted result, prepared descriptor and terminal failure. - Hold other continuation paths until the durable reconstruction intent admits its successor. Record that successor in the same transaction and preserve the retry count. - Add an operator control that distinguishes reconstruction from original-file recovery. Update the recovery runbook and capability maps. ## Verification - Exact baseline source returns no blocker for the destroyed-source fixture. The same fixture returns a required recovery hold with this change. - New real PostgreSQL and assembled API/heartbeat suite: 25/25 pass. It covers duplicate submissions, lost acknowledgement, context reset, source identity, task ownership, concurrent execution attachment scope, later human blocks, deleted sources and agent pause. - Existing export retry, explicit continuation and recovery-action suites: 386 cases pass. One new-fixture constraint error in the combined run was corrected; the full new suite then passed. - UI recovery suites: 46/46 pass across three files. They cover disabled actions, loss acknowledgement, failure retention and runtime authority. - `pnpm -r typecheck` and `pnpm build` pass. The first sandboxed typecheck attempt could not open the TypeScript IPC socket; the authorized rerun passed. - Module boundaries, map checks, token gates and diff whitespace checks pass. - OpenAPI route coverage: 21/21 pass. The first CI head exposed a missing endpoint registration; this head adds the registration and the focused suite passes. - Review identified a missing dispatch-time check and an isolated attachment cache. Both are fixed. Claim and final handoff recheck the exact successor, sources and current execution gates. A controlled post-claim operator block prevents the assembled heartbeat from entering the provider. The shared attachment cache refreshes uploads and removes deleted selections. - Final focused run: 180/180 pass across six suites, including reconstruction, existing dispatch/stale-queue contracts and UI recovery. - A second review reproduced cancellation by an already satisfied historical connection card while the task was `todo`. Read-only connection availability now works before checkout; active-task mutation checks remain strict. The baseline failed two cases; after the fix all 74 connection and reconstruction tests pass, including revoked access, pending questions and pending approvals. - All current-head CI gates pass on `9d197236da11fa24f1ceadf5589cb2b81179d824`, including general/serialized tests, Runner, build and browser gates. Greptile is 5/5 for that exact head with no unresolved threads. The full local `pnpm test:run` is still running its serial server group; no full local pass is claimed. Live staging recovery is pending deployment. ## Risks - Reconstruction cannot restore missing bytes. The operator must identify surviving sources and actions that must not be repeated. The agent must verify those attachments before rebuilding. - This path is limited to exact, confirmed destroyed remote sources with an accepted result. Retained or uncertain sources keep their existing repair path. - No schema migration, credential change, budget increase or warm-session policy change is included. Normal pause, ownership, budget, approval and dispatch gates still apply. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, tool use and code execution. The session does not expose an exact backend model ID or context-window size. ## 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 (focused suites; full local run limitation recorded above) - [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 - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/native-workspace-finalization-recovery.md | 24 ++ maps/agent-access-map/recovery.md | 5 + maps/agent-access-map/workspaces.md | 8 + maps/feature-map/workspaces.md | 8 + packages/shared/src/index.ts | 1 + .../shared/src/types/execution-projection.ts | 11 + packages/shared/src/validators/index.ts | 1 + packages/shared/src/validators/issue.ts | 11 + .../satisfied-connection-intents.test.ts | 24 +- .../modules/run-dispatch/adapters/postgres.ts | 27 ++- server/src/routes/issues.ts | 19 +- server/src/routes/openapi.ts | 10 + server/src/services/execution-blocker.ts | 29 ++- .../services/explicit-native-continuation.ts | 3 + server/src/services/heartbeat.ts | 3 +- server/src/services/heartbeat/queue.ts | 24 +- .../native-workspace-reconstruction.test.ts | 222 ++++++++++++++++++ .../native-workspace-reconstruction.ts | 152 ++++++++++++ .../services/satisfied-connection-intents.ts | 11 +- ui/src/api/issues.ts | 3 + ui/src/components/ExecutionBlockerNotice.tsx | 10 +- .../WorkspaceReconstructionRecovery.test.tsx | 74 ++++++ .../WorkspaceReconstructionRecovery.tsx | 67 ++++++ 23 files changed, 724 insertions(+), 23 deletions(-) create mode 100644 server/src/services/native-runtime/native-workspace-reconstruction.test.ts create mode 100644 server/src/services/native-runtime/native-workspace-reconstruction.ts create mode 100644 ui/src/components/WorkspaceReconstructionRecovery.test.tsx create mode 100644 ui/src/components/WorkspaceReconstructionRecovery.tsx diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index 3d861bb82a9..570c221d0ee 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -359,3 +359,27 @@ A retained source receipt records control-plane evidence; it does not prove the provider still holds the original bytes or that copy-back succeeded. Operators must verify and recover the exact source separately. Later successful runs and completed tasks do not certify recovery of earlier workspace files. + +### Continue after confirmed source destruction + +When an old export failed and its exact sandbox has a confirmed destruction +receipt, the task can continue from surviving attachments. This is reconstruction, +not recovery of the missing edits. The task's recovery notice exposes **Continue +from saved sources** to board operators with `runtime:manage`. Select the surviving +attachments, describe the missing work, and record the observed external action +outcomes. A retained or unverified sandbox must use the existing export or file +recovery path instead. + +`POST /api/issues/{id}/recovery-actions/reconstruct-workspace` records this choice. +It validates the source run, accepted result, prepared descriptor, exact destroyed +lease, current task owner and status version. It rejects active copyback owners, +competing recovery, newer executions, and attachments from another task or company. +The transaction records attachment hashes and a durable fresh-session delivery +intent. The old run, result, descriptor and failed finalization remain intact. +Repeated submissions adopt the original intent. Normal admission rechecks source +identity, attachments, authority, budgets, holds and task eligibility. It preserves +the retry count and does not grant a new work window or replay completed actions. +The agent must verify the selected attachments before rebuilding files. +Queued successors recheck their exact reconstruction receipt, sources and current +execution gates when claimed and immediately before provider handoff. A later +human block cannot be overwritten by automatic checkout. diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index 016f34aef44..5a764e09a71 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -176,6 +176,7 @@ Review: `baseline`. API: `present` — representative operation IDs: - `POST /api/issues/{id}/recovery-actions/retry-workspace-export`. +- `POST /api/issues/{id}/recovery-actions/reconstruct-workspace`. - `POST /api/issues/{id}/recovery-actions/resume-workspace-finalization`. - `POST /api/issues/{id}/recovery-actions/resolve` for retained-source file repair. - `POST /api/execution-workspaces/{id}/reconcile-branch`. @@ -184,6 +185,10 @@ Agent access: `restricted` — Board/runtime management APIs; agent reads diagno performs permitted workspace work. Policy: Export retry and branch reconciliation require board runtime management. +Reconstruction also requires board runtime management. It verifies exact source +destruction and same-task attachment hashes, records lost work and external action +outcomes, and admits one fresh continuation through normal execution gates. The +old failed export remains failed. A retained or uncertain source cannot use it. Retained legacy file-only repair requires the board, exact stopped-source receipts, and recovered-file evidence. It can record repair for a successful legacy turn without changing its result or authorizing replay. Queued and running sources diff --git a/maps/agent-access-map/workspaces.md b/maps/agent-access-map/workspaces.md index 6d050c32fd7..ffaef2f422c 100644 --- a/maps/agent-access-map/workspaces.md +++ b/maps/agent-access-map/workspaces.md @@ -2,6 +2,14 @@ Product map: [Execution workspaces, services, and files](../feature-map/workspaces.md). +Lost-source reconstruction adds the board-only +`POST /api/issues/{id}/recovery-actions/reconstruct-workspace` action. It requires +`runtime:manage` and ordinary task visibility. Agents cannot authorize it. The +result is a durable fresh-session admission request, subject to current execution +gates. It grants no additional tool access, budget, work window or permission to +repeat completed actions. Attachments must belong to the same task and company. +The original failed run and lost-export evidence remain unchanged. + Review date: 2026-10-12. Source commit: `58e16dbddfc9c8d0cf60867f9d9847f0feda055b`. Native REST transport reconciled in this change; see the [shared transport contract](README.md#native-rest-transport). diff --git a/maps/feature-map/workspaces.md b/maps/feature-map/workspaces.md index 4bda0d21c3c..b1508f67781 100644 --- a/maps/feature-map/workspaces.md +++ b/maps/feature-map/workspaces.md @@ -18,6 +18,14 @@ Agent access: [companion](../agent-access-map/workspaces.md). - `files-git`: inspect workspace files and repository state used for the deliverable. - `closure`: close/reopen or reconcile through permitted controls without losing work. +For a historical failed export whose exact sandbox was destroyed, the task recovery +notice offers **Continue from saved sources** to an authorized board operator. +Select surviving task attachments, record lost work and external action outcomes, +then rebuild through fresh-session admission. The old failed run remains failed. +Retained sandboxes keep their existing export recovery path. Automated coverage: +[reconstruction](../../server/src/services/native-runtime/native-workspace-reconstruction.test.ts). +Live staging acceptance remains separate from these regression tests. + ## How to get to it (user POV) ### `task-workspace` diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index af3387712b2..7d9a9799b52 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -2001,6 +2001,7 @@ export { resolveIssueRecoveryActionSchema, isValidExistingBranchName, retryWorkspaceExportSchema, + reconstructWorkspaceSchema, resumeWorkspaceFinalizationSchema, issueReviewRequestSchema, issueExecutionWorkspaceSettingsSchema, diff --git a/packages/shared/src/types/execution-projection.ts b/packages/shared/src/types/execution-projection.ts index c84678fec66..2a8fb29bbc7 100644 --- a/packages/shared/src/types/execution-projection.ts +++ b/packages/shared/src/types/execution-projection.ts @@ -14,6 +14,9 @@ export interface ExecutionBlocker { /** Required workspace files must be recovered before another provider turn. */ workspaceRepairRequired?: boolean; canRecordWorkspaceRepair?: boolean; + canReconstructWorkspace?: boolean; + workspaceReconstructionPending?: boolean; + workspaceRecoveryStatusVersion?: number; savedMessageCount?: number; } @@ -77,4 +80,12 @@ export interface ExecutionReconciliation { outcomeEvidence: string; /** Operator evidence bound to this failed run; required after workspace restore failure. */ workspaceRepairEvidence?: string; + workspaceReconstruction?: { + schema: "paperclip.workspace-reconstruction.v1"; + leaseId: string; + resultId: string; + descriptorSha256: string; + missingWork: string; + sources: Array<{ attachmentId: string; sha256: string; byteSize: number; filename: string | null }>; + }; } diff --git a/packages/shared/src/validators/index.ts b/packages/shared/src/validators/index.ts index 8e224de0924..1684fdeaa4b 100644 --- a/packages/shared/src/validators/index.ts +++ b/packages/shared/src/validators/index.ts @@ -441,6 +441,7 @@ export { resolveIssueRecoveryActionSchema, isValidExistingBranchName, retryWorkspaceExportSchema, + reconstructWorkspaceSchema, resumeWorkspaceFinalizationSchema, issueReviewRequestSchema, issueExecutionWorkspaceSettingsSchema, diff --git a/packages/shared/src/validators/issue.ts b/packages/shared/src/validators/issue.ts index 23c0fd7e103..cf398d1f8bb 100644 --- a/packages/shared/src/validators/issue.ts +++ b/packages/shared/src/validators/issue.ts @@ -573,6 +573,17 @@ export const retryWorkspaceExportSchema = z.object({ repairNote: z.string().trim().min(20).max(12000), }).strict(); +export const reconstructWorkspaceSchema = z.object({ + actionId: z.string().guid(), + runId: z.string().guid(), + expectedStatusVersion: z.number().int().nonnegative(), + attachmentIds: z.array(z.string().guid()).min(1).max(20).refine(ids => new Set(ids).size === ids.length), + missingWork: z.string().trim().min(20).max(12000), + actionOutcome: z.enum(["completed", "not_performed", "mixed"]), + outcomeEvidence: z.string().trim().min(20).max(12000), + acknowledgeLostWork: z.literal(true), +}).strict(); + export const resolveIssueRecoveryActionSchema = z .object({ workspaceBaseRef: z.object({ diff --git a/server/src/__tests__/satisfied-connection-intents.test.ts b/server/src/__tests__/satisfied-connection-intents.test.ts index 8c8f4a53309..6b5fd7da053 100644 --- a/server/src/__tests__/satisfied-connection-intents.test.ts +++ b/server/src/__tests__/satisfied-connection-intents.test.ts @@ -7,7 +7,7 @@ import { activityLog, agentWakeupRequests, agents, approvals, authUsers, compani } from "@paperclipai/db"; import { createPostgresRunDispatchAdapter } from "../modules/run-dispatch/adapters/postgres.js"; import { connectionIntentDeliveryService } from "../services/connection-intent-delivery.js"; -import { findSatisfiedToolConnection, satisfiedConnectionIntentService } from "../services/satisfied-connection-intents.js"; +import { connectionContinuationPendingResponse, findSatisfiedToolConnection, satisfiedConnectionIntentService } from "../services/satisfied-connection-intents.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { materializeNativeInteractionResponses } from "../services/native-runtime/native-interaction-bridge.js"; import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; @@ -64,6 +64,28 @@ describePostgres("already available connection requests", () => { expect((await db.select().from(issueThreadInteractions))[0].status).toBe("pending"); }); + it.each(["ready", "revoked", "question", "approval"] as const)("checks retired connection cards while a continuation is todo: %s", async state => { + const f = await seed(); + const run = await retry(f); + await satisfiedConnectionIntentService(db).sweepPending(); + expect((await db.select().from(issueThreadInteractions))[0].status).toBe("expired"); + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, f.issueId)); + if (state === "revoked") await db.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.id, f.grant.id)); + if (state === "question") await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, + kind: "ask_user_questions", status: "pending", addresseeUserId: "connection-owner", payload: { version: 1, questions: [] } }); + if (state === "approval") { + const [approval] = await db.insert(approvals).values({ companyId: f.companyId, type: "tool_action", status: "pending", payload: {} }).returning(); + await db.insert(issueApprovals).values({ companyId: f.companyId, issueId: f.issueId, approvalId: approval.id }); + } + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(await findSatisfiedToolConnection(db, task, f.intent, run.id)).toBeNull(); + expect(await connectionContinuationPendingResponse(db, task, "connection-owner", run.id)) + .toBe(state === "ready" ? null : state === "approval" ? "approval" : "interaction"); + expect(await createPostgresRunDispatchAdapter(db).evaluateScheduledRetryGate({ companyId: f.companyId, + runId: run.id, retryReasonOverride: "native_safe_replacement", now: new Date() })) + .toMatchObject({ allowed: state === "ready" }); + }); + it("keeps a retry under another user's identity waiting for its response", async () => { const f = await seed(); const run = await retry(f); await db.update(heartbeatRuns).set({ responsibleUserId: "someone-else" }).where(eq(heartbeatRuns.id, run.id)); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 947e7d33863..28070336533 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -1,5 +1,6 @@ import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js"; import { getExecutionBlocker } from "../../../services/execution-blocker.js"; +import { validateReconstructedWorkspaceDispatch } from "../../../services/native-runtime/native-workspace-reconstruction.js"; import { getNativeReviewAssignment } from "../../../services/native-runtime/native-review-participant.js"; import { and, asc, eq, gte, inArray, lte, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -981,10 +982,27 @@ export function createPostgresRunDispatchAdapter( }; } - async function decideCurrentRunStaleness(tx: Db, run: HeartbeatRun, now: Date) { + async function decideCurrentRunStaleness(tx: Db, run: HeartbeatRun, now: Date, publications?: ActivityPublication[]) { const contextSnapshot = parseObject(run.contextSnapshot); const issueId = readNonEmptyString(contextSnapshot.issueId); if (!issueId) return { issueId: null, facts: null, executionBlocker: null, decision: { stale: false as const } }; + if (typeof contextSnapshot.workspaceReconstructionActionId === "string") { + try { + await validateReconstructedWorkspaceDispatch(tx, run); + const current = await loadGateFacts({ runId: run.id, companyId: run.companyId, agentId: run.agentId, + responsibleUserId: run.responsibleUserId, conversationContinuation: true, contextSnapshot, + scheduledRetryReason: run.scheduledRetryReason, retryReasonOverride: "native_safe_replacement", wakeupRequestId: run.wakeupRequestId, + }, now, tx, publications); + if (!current.agentFound || !decideScheduledRetryGate(current.facts, now).allowed) throw new Error("reconstruction_execution_not_eligible"); + } + catch { + return { issueId, facts: null, executionBlocker: null, decision: { stale: true as const, + errorCode: "execution_reconciliation_required" as const, + reason: "Workspace reconstruction changed before execution. Inspect the saved sources and current task before continuing.", + details: { issueId, recoveryActionId: contextSnapshot.workspaceReconstructionActionId }, + } }; + } + } const recovery = await getExecutionBlocker(tx, run.companyId, issueId, { conversationResetCommentId: deriveCommentId(contextSnapshot) }); if (recovery) return { issueId, facts: null, executionBlocker: recovery, decision: { stale: true as const, errorCode: "execution_reconciliation_required" as const, reason: recovery.nextAction, @@ -1070,9 +1088,9 @@ export function createPostgresRunDispatchAdapter( async function cancelStaleQueuedRun( input: CancelStaleQueuedRunInput, ): Promise { - const cancelLockedRun = async (tx: Db, run: HeartbeatRun) => { + const cancelLockedRun = async (tx: Db, run: HeartbeatRun, publications: ActivityPublication[]) => { if (run.status !== input.expectedStatus) return { outcome: "lost_race" as const }; - const { issueId, facts, decision, executionBlocker } = await decideCurrentRunStaleness(tx, run, input.now); + const { issueId, facts, decision, executionBlocker } = await decideCurrentRunStaleness(tx, run, input.now, publications); if (!decision.stale || !issueId) { if (input.expectedStatus === "queued" && facts?.isInteractionWake) { // Preserve the authority accepted under the issue/run locks. Later @@ -1109,7 +1127,7 @@ export function createPostgresRunDispatchAdapter( async function dispatchResolvedInteractionIfCurrent( input: DispatchResolvedInteractionInput, ): Promise> { - const dispatchLockedRun = async (tx: Db, run: HeartbeatRun) => { + const dispatchLockedRun = async (tx: Db, run: HeartbeatRun, publications: ActivityPublication[]) => { if (run.status !== input.expectedStatus || run.resultJson?.cancellation || run.resultJson?.startupCancellation) { return { dispatched: false as const, cancellation: { outcome: "lost_race" as const } }; } @@ -1117,6 +1135,7 @@ export function createPostgresRunDispatchAdapter( tx, run, input.now, + publications, ); const decision = !initialDecision.stale && issueId diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index eacaea637c2..6852b422809 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -19,6 +19,8 @@ import { createIssueReadTiming } from "../services/issue-read-timing.js"; import { isNativeWorkspaceExportRepairCause } from "@paperclipai/shared"; import { resumeNativeWorkspaceAfterOwnerStop } from "../services/native-runtime/native-workspace-owner-recovery.js"; import { retryNativeWorkspaceExport } from "../services/native-runtime/native-workspace-export-retry.js"; +import { reconstructNativeWorkspace, unreconciledNativeExportPredicate } from "../services/native-runtime/native-workspace-reconstruction.js"; +import { reconstructWorkspaceSchema } from "@paperclipai/shared"; import { queuedInteractionId, readQueuedInteractionResponse, hasQueuedInteractionResponse } from "../services/queued-interaction-response.js"; import { deliverConversationComments, isConversation } from "../services/agent-conversations.js"; import { issueRecoveryActionReadModel } from "../services/issue-recovery-actions.js"; @@ -9501,11 +9503,11 @@ export function issueRoutes( }); const retainedHolds = await db.select().from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, issue.companyId), eq(issueRecoveryActions.sourceIssueId, issue.id), - sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`, - or( + or(unreconciledNativeExportPredicate(), and( + sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`, or( sql`${issueRecoveryActions.evidence}->'workspaceRestoreRecovery'->>'schema' = ${LEGACY_WORKSPACE_RECOVERY_SCHEMA}`, inArray(issueRecoveryActions.cause, [...EXECUTION_RECONCILIATION_CAUSES]), - ), + ))), )).orderBy(desc(issueRecoveryActions.createdAt)); res.json({ active, @@ -9537,6 +9539,17 @@ export function issueRoutes( res.status(202).json(receipt); }); + router.post("/issues/:id/recovery-actions/reconstruct-workspace", validate(reconstructWorkspaceSchema), async (req, res) => { + assertBoard(req); + const issue = await getAccessibleResource(req, res, svc.getById(req.params.id as string), "Issue not found"); + if (!issue || !(await assertIssueReadAllowed(req, res, issue))) return; + const decision = await access.decide({ actor: req.actor, action: "runtime:manage", resource: { type: "company", companyId: issue.companyId } }); + if (!decision.allowed) throw forbidden(decision.explanation, authorizationDeniedDetails(decision)); + const receipt = await reconstructNativeWorkspace({ db, companyId: issue.companyId, issueId: issue.id, + actorId: getActorInfo(req).actorId, request: req.body }); + res.status(202).json(receipt); + }); + router.post( "/issues/:id/recovery-actions/resolve", validate(resolveIssueRecoveryActionSchema), diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index 120312ba872..d51610850fd 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -243,6 +243,7 @@ import { createAcceptedPlanDecompositionSchema, resolveIssueRecoveryActionSchema, retryWorkspaceExportSchema, + reconstructWorkspaceSchema, resumeWorkspaceFinalizationSchema, cancelIssueThreadInteractionSchema, // Secret provider configs and remote import @@ -10668,6 +10669,15 @@ registerCurrentRoute({ responses: { 202: r.ok(), 400: r.badRequest, 401: r.unauthorized, 403: r.forbidden, 404: r.notFound, 409: r.conflict }, }); +registerCurrentRoute({ + method: "post", + path: "/api/issues/{id}/recovery-actions/reconstruct-workspace", + tags: ["issues"], + summary: "Continue from saved task attachments after confirmed destruction of the original workspace", + body: reconstructWorkspaceSchema, + responses: { 202: r.ok(), 400: r.badRequest, 401: r.unauthorized, 403: r.forbidden, 404: r.notFound, 409: r.conflict }, +}); + registerCurrentRoute({ method: "post", path: "/api/issues/{id}/recovery-actions/resolve", diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index 38d615dfbd5..2d2ae87c852 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -8,6 +8,9 @@ import { canRetryStoppedRun } from "./cancelled-native-startup.js"; import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; import { EXECUTION_RECONCILIATION_CAUSES, type ExecutionBlocker } from "@paperclipai/shared"; import { hasRequiredWorkspaceRecovery, LEGACY_WORKSPACE_RECOVERY_SCHEMA, nativeRunHasNoWorkspaceAccess } from "./workspace-restore-recovery-state.js"; +import { unreconciledNativeExportPredicate } from "./native-runtime/native-workspace-reconstruction.js"; +import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; +import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; export async function hasPendingRunCleanup(db: Db, companyId: string, runId: string): Promise { const [state] = await db.execute<{ pending: boolean }>(sql`SELECT @@ -40,16 +43,18 @@ export async function hasPendingRunCleanup(db: Db, companyId: string, runId: str export function executionBlockerPredicate() { return and( not(conversationRecoveryActionPredicate()!), - inArray(issueRecoveryActions.cause, [...EXECUTION_RECONCILIATION_CAUSES]), - or(inArray(issueRecoveryActions.status, ["active", "escalated"]), - sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`), + or(unreconciledNativeExportPredicate(), and( + inArray(issueRecoveryActions.cause, [...EXECUTION_RECONCILIATION_CAUSES]), + or(inArray(issueRecoveryActions.status, ["active", "escalated"]), + sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`), + )), ); } export async function getExecutionBlocker(db: Db, companyId: string, issueId: string, options?: { conversationResetCommentId?: string | null }): Promise { const [conversation] = await db.select({ agentId: issues.conversationAgentId, boundaryId: issues.conversationBoundaryCommentId, status: issues.status, - assigneeAgentId: issues.assigneeAgentId }).from(issues).where(and( + assigneeAgentId: issues.assigneeAgentId, statusVersion: issues.statusVersion }).from(issues).where(and( eq(issues.companyId, companyId), eq(issues.id, issueId), )).limit(1); // Resetting model context cannot repair or recover a workspace. This hold @@ -59,6 +64,7 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st eq(issueRecoveryActions.sourceIssueId, issueId), executionBlockerPredicate(), or( + unreconciledNativeExportPredicate(), sql`${issueRecoveryActions.evidence}->>'workspaceRestoreFailure' = 'restore_unsafe_archive'`, sql`${issueRecoveryActions.evidence}->'workspaceRestoreRecovery'->>'schema' = ${LEGACY_WORKSPACE_RECOVERY_SCHEMA}`, ), @@ -94,6 +100,17 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st )).limit(1) : []; const withoutWorkspace = Boolean(run?.driverKind === "grok_bot_mcp" && nativeRunHasNoWorkspaceAccess(run)); const workspaceRepairRequired = Boolean(restoreHold && !withoutWorkspace); + const reconstructableExport = ["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"].includes(action.cause); + const exportSource = reconstructableExport ? readNativeWorkspaceSyncReference(run?.runnerProfileJson?.nativeWorkspaceSync) : null; + const [exportLease] = exportSource ? await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run!.id), eq(environmentLeases.id, exportSource.leaseId), + eq(environmentLeases.providerLeaseId, exportSource.providerLeaseId), + )).limit(1) : []; + const canReconstructWorkspace = Boolean(exportLease && hasRemoteTerminationReceipt(exportLease) + && (exportLease.metadata?.remoteExecutionTermination as { state?: string } | undefined)?.state === "destroyed" + && !action.evidence.executionReconciliation && conversation?.status === "blocked" && conversation.assigneeAgentId === run?.agentId); + const reconstructionPending = Boolean((action.evidence.executionReconciliation as { workspaceReconstruction?: unknown } | undefined)?.workspaceReconstruction + && action.evidence.continuationDelivery === "pending"); const queue = await db.select({ payload: agentWakeupRequests.payload }).from(agentWakeupRequests).where(and( eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.status, "deferred_issue_execution"), sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, @@ -163,12 +180,14 @@ export async function getExecutionBlocker(db: Db, companyId: string, issueId: st // A stopped reviewer can differ from the task owner who receives the work back. agentId: run?.agentId ?? null, cause: action.cause, - nextAction, + nextAction: reconstructionPending ? "Continuation from saved sources is queued and waiting for execution checks." : nextAction, runStatus: run?.status ?? null, runError: runError?.slice(0, 1024) ?? null, canContinue: eligibleContinuation && !chatBinding, canRetry: canRetry && !chatBinding, ...(restoreHold ? { workspaceRepairRequired, canRecordWorkspaceRepair: workspaceRepairRequired && hasRequiredWorkspaceRecovery(action.evidence) } : {}), savedMessageCount: saved.length, + ...(canReconstructWorkspace ? { canReconstructWorkspace, workspaceRecoveryStatusVersion: conversation!.statusVersion } : {}), + ...(reconstructionPending ? { workspaceReconstructionPending: true } : {}), }; } diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index aabb3e1b780..46686f7d7ba 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -144,6 +144,9 @@ export async function admitExplicitNativeContinuation(input: { if (!actions.length) return null; const blocker = await getExecutionBlocker(db, companyId, issueId); if (blocker && blocker.recoveryActionId === null) return null; + if (blocker?.workspaceRepairRequired && ["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"].includes(blocker.cause)) { + return blocked("workspace_reconstruction_required", "Recover the saved files or record reconstruction before continuing. Your message is saved."); + } const [pendingInteraction] = await db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and( eq(issueThreadInteractions.companyId, companyId), eq(issueThreadInteractions.issueId, issueId), eq(issueThreadInteractions.status, "pending"), diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 81632b1b83c..474559cd290 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -2173,7 +2173,7 @@ export function heartbeatService( await issuesSvc.checkout( issueId, agent.id, - ["todo", "backlog", "blocked"], + typeof context.workspaceReconstructionActionId === "string" ? ["todo"] : ["todo", "backlog", "blocked"], run.id, ); context[PAPERCLIP_HARNESS_CHECKOUT_KEY] = true; @@ -2844,6 +2844,7 @@ export function heartbeatService( if ( !issueId || (!isResolvedInteractionContinuationWakeContext(context) && + typeof context.workspaceReconstructionActionId !== "string" && !["native_safe_replacement", "native_provider_overloaded"].includes(run.scheduledRetryReason ?? "")) ) { return { dispatched: true, resultPromise: dispatch(() => {}) }; diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index be493ce9b40..758910670f2 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -33,6 +33,7 @@ import { } from "../explicit-native-continuation.js"; import { canRetryStoppedRun } from "../cancelled-native-startup.js"; import { getExecutionBlocker, hasPendingRunCleanup } from "../execution-blocker.js"; +import { validateNativeWorkspaceReconstruction } from "../native-runtime/native-workspace-reconstruction.js"; import { isConversationAdapter } from "../conversation-continuation.js"; import { recordExecutionWait } from "../execution-wait.js"; import { @@ -3102,6 +3103,13 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const [reconciledRun] = await tx.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, issue.companyId), eq(heartbeatRuns.id, sourceRunId), )); + if (decision.workspaceReconstruction) { + await validateNativeWorkspaceReconstruction(tx as unknown as Db, { + companyId: issue.companyId, issueId: issue.id, runId: sourceRunId, actionId, + }); + reconciledRestoreRetryCount = executionFailureRetryCount(reconciledRun!); + enrichedContextSnapshot.workspaceReconstructionActionId = action.id; + } if (hasWorkspaceRestoreFailure(reconciledRun?.resultJson)) { if ((readNonEmptyString(decision.workspaceRepairEvidence)?.length ?? 0) < 20) return { kind: "skipped" as const }; @@ -3170,7 +3178,9 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // A bound chat request has already rechecked its current principal // above. Treat its new user message like a board comment, but keep // failed-run retry actions on their separate exact-request path. - if (executionBlocker && !(await admitExplicitNativeContinuation({ + const ownsReconstructionContinuation = reconciledSourceRunId && executionBlocker?.runId === reconciledSourceRunId + && executionBlocker.recoveryActionId === enrichedContextSnapshot.recoveryActionId; + if (executionBlocker && !ownsReconstructionContinuation && !(await admitExplicitNativeContinuation({ db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id, agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId, reason: durableRequest && !failedChatRetry ? "issue_commented" : reason, @@ -3968,7 +3978,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende queuedCommentInterruptId: opts.queuedCommentInterruptId, queuedCommentRequestId: opts.queuedCommentRequestId, }); - if (!explicitContinuation && executionBlocker) return deferBlockedExecution(executionBlocker); + if (!explicitContinuation && executionBlocker && !ownsReconstructionContinuation) return deferBlockedExecution(executionBlocker); if (explicitContinuation) { enrichedContextSnapshot.forceFreshSession = true; enrichedContextSnapshot.previousRunId = explicitContinuation.previousRunId; @@ -4013,7 +4023,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // in separate turns, or finalization will deliver it a second time. const pendingComments = (executionWaitRequestId || canCoalesceComments) && - !(await getExecutionBlocker(tx as unknown as Db, issue.companyId, issue.id)) + (ownsReconstructionContinuation || !(await getExecutionBlocker(tx as unknown as Db, issue.companyId, issue.id))) ? await tx .select() .from(agentWakeupRequests) @@ -4095,6 +4105,14 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende .returning() .then((rows) => rows[0]); + if (ownsReconstructionContinuation) { + await tx.update(issueRecoveryActions).set({ evidence: sql`${issueRecoveryActions.evidence} || ${JSON.stringify({ + continuationDelivery: "delivered", continuationRunId: newRun.id, + })}::jsonb` }).where(and(eq(issueRecoveryActions.companyId, issue.companyId), + eq(issueRecoveryActions.id, String(enrichedContextSnapshot.recoveryActionId)), + sql`${issueRecoveryActions.evidence}->>'continuationDelivery' = 'pending'`)); + } + await tx .update(agentWakeupRequests) .set({ diff --git a/server/src/services/native-runtime/native-workspace-reconstruction.test.ts b/server/src/services/native-runtime/native-workspace-reconstruction.test.ts new file mode 100644 index 00000000000..52a65d4b785 --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-reconstruction.test.ts @@ -0,0 +1,222 @@ +import { randomUUID } from "node:crypto"; +import express from "express"; +import request from "supertest"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { activityLog, agents, agentWakeupRequests, assets, authUsers, companies, completionContracts, createDb, environmentLeases, heartbeatRuns, issueAttachments, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { errorHandler } from "../../middleware/index.js"; +import { issueRoutes } from "../../routes/issues.js"; +import { getExecutionBlocker } from "../execution-blocker.js"; +import { admitExplicitNativeContinuation } from "../explicit-native-continuation.js"; +import { deliverReconciledExecutions } from "../execution-recovery-resolution.js"; +import { heartbeatService } from "../heartbeat.js"; +import { remoteTerminationReceipt } from "../remote-execution-termination.js"; +import { reconstructNativeWorkspace } from "./native-workspace-reconstruction.js"; +import { createRunDispatch } from "../../modules/run-dispatch/index.js"; +import { drainHeartbeatRunsToQuiescence } from "../../__tests__/helpers/drain-heartbeat-runs.js"; +const providerExecute = vi.hoisted(() => vi.fn(async () => ({ exitCode: 0, signal: null, timedOut: false, + errorMessage: null, summary: "Bounded reconstruction completed.", provider: "test", model: "test" }))); +vi.mock("../../adapters/index.js", async () => ({ + ...await vi.importActual("../../adapters/index.js"), + getServerAdapter: () => ({ supportsLocalAgentJwt: false, execute: providerExecute }), +})); + +describe("reconstruction after confirmed native workspace loss", () => { + let temporary: Awaited>; + let db: ReturnType; + let runningHeartbeat: ReturnType | null = null; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("workspace-reconstruction-"); + db = createDb(temporary.connectionString); + }, 30_000); + afterAll(async () => { await temporary.cleanup(); }); + afterEach(async () => { + if (runningHeartbeat) await drainHeartbeatRunsToQuiescence(db, runningHeartbeat); + runningHeartbeat = null; + providerExecute.mockClear(); + await db.transaction(async tx => { + await tx.execute(sql`set local client_min_messages = warning`); + await tx.execute(sql`truncate companies cascade`); + }); }); + + async function seed() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(), resultId = randomUUID(), + contractId = randomUUID(), leaseId = randomUUID(), actionId = randomUUID(), providerLeaseId = randomUUID(), userId = randomUUID(), attachmentId = randomUUID(), assetId = randomUUID(); + await db.insert(authUsers).values({ id: userId, name: "Operator", email: `${userId}@example.test`, emailVerified: true, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companies).values({ id: companyId, name: "Reconstruction", issuePrefix: `REC${companyId.replaceAll("-", "").slice(0, 6)}`, defaultResponsibleUserId: userId, budgetMonthlyCents: 10000 }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", status: "idle", adapterType: "paperclip_runner", budgetMonthlyCents: 5000, runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }); + await db.insert(issues).values({ id: issueId, companyId, title: "Finish bounded work", status: "blocked", assigneeAgentId: agentId }); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: { objective: "Finish bounded work" }, canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "failed", runtimeMode: "native", nativeIssueId: issueId, nativePhase: "terminal_failure", completionContractId: contractId, + createdAt: new Date("2026-05-01T00:00:00Z"), finishedAt: new Date("2026-05-01T01:30:00Z"), responsibleUserId: userId, scheduledRetryAttempt: 2, + contextSnapshot: { issueId, originalWorkDeadline: "2026-05-01T01:30:00Z" }, errorCode: "native_session_interrupted", processPid: process.pid, + runnerProfileJson: { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v1", state: "prepared", descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId: randomUUID(), leaseId, providerLeaseId, remoteCwd: "/work", resourceDisposition: "destroy" } } }); + await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId, completionContractId: contractId, serverFingerprint: resultId, schemaStatus: "accepted", resultJson: { work: "Unfinished local edits; no published PR" }, canonicalSha256: resultId }); + const cause = "native_workspace_sync_out_unrecoverable"; + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "terminal_failure", resultId, failureCode: cause }); + const identity = { id: leaseId, companyId, heartbeatRunId: runId, provider: "daytona", providerLeaseId }; + await db.insert(environmentLeases).values({ ...identity, issueId, status: "expired", releasedAt: new Date(), cleanupStatus: "success", metadata: { remoteExecutionTermination: remoteTerminationReceipt(identity, { providerLeaseId, state: "destroyed" }) } }); + await db.insert(issueRecoveryActions).values({ id: actionId, companyId, sourceIssueId: issueId, kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: agentId, cause, fingerprint: runId, evidence: { runId }, nextAction: "Recover saved files", status: "resolved", outcome: "restored", resolutionNote: "new_source_execution_path" }); + await db.insert(assets).values({ id: assetId, companyId, provider: "local", objectKey: assetId, contentType: "text/plain", byteSize: 100, sha256: "c".repeat(64), originalFilename: "saved-baseline.ts" }); + await db.insert(issueAttachments).values({ id: attachmentId, companyId, issueId, assetId }); + const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); + const body = { actionId, runId, expectedStatusVersion: issue.statusVersion, attachmentIds: [attachmentId], missingWork: "The unexported patch is lost. Rebuild only from this earlier verified baseline.", actionOutcome: "not_performed" as const, outcomeEvidence: "The saved execution receipt and remote repository show no published PR or submitted external change.", acknowledgeLostWork: true as const }; + const input = { db, companyId, issueId, actorId: userId, request: body }; + return { ...input, body, agentId, leaseId, resultId, runId, actionId, providerLeaseId, attachmentId, assetId }; + } + function app(actor: unknown = { type: "board", source: "local_implicit" }) { + const app = express(); app.use(express.json()); + app.use((req, _res, next) => { (req as unknown as { actor: unknown }).actor = actor; next(); }); + app.use("/api", issueRoutes(db, {} as never)); app.use(errorHandler); return app; + } + it("keeps a falsely resolved export visible and prevents conversation reset from hiding it", async () => { + const f = await seed(); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ runId: f.runId, workspaceRepairRequired: true, canReconstructWorkspace: true }); + const commentId = randomUUID(); + await db.insert(issueComments).values({ id: commentId, companyId: f.companyId, issueId: f.issueId, authorType: "user", authorUserId: f.actorId, body: "/new" }); + await db.update(issues).set({ conversationAgentId: f.agentId, conversationUserId: f.actorId, conversationState: "active" }).where(eq(issues.id, f.issueId)); + expect(await getExecutionBlocker(db, f.companyId, f.issueId, { conversationResetCommentId: commentId })).not.toBeNull(); + expect(await admitExplicitNativeContinuation({ db, companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + actorType: "user", actorId: f.actorId, reason: "issue_commented", commentId, successorRunId: randomUUID() })).toBeNull(); + const response = await request(app()).get(`/api/issues/${f.issueId}/recovery-actions`).expect(200); + expect(response.body.actions).toEqual(expect.arrayContaining([expect.objectContaining({ id: f.actionId })])); + }); + it("records partial sources atomically, preserves failed history and budgets, and deduplicates concurrent submissions", async () => { + const f = await seed(); + const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await Promise.all([reconstructNativeWorkspace(f), reconstructNativeWorkspace(f)]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(before); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toMatchObject({ phase: "terminal_failure", resultId: f.resultId }); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0].schemaStatus).toBe("accepted"); + const [record] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + expect(record).toMatchObject({ outcome: "handed_back", evidence: { continuationDelivery: "pending", executionReconciliation: { workspaceReconstruction: { missingWork: f.body.missingWork, sources: [{ attachmentId: f.attachmentId, sha256: "c".repeat(64) }] } } } }); + expect(record.evidence.executionReconciliation).not.toHaveProperty("workspaceRepairEvidence"); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ runId: f.runId, workspaceRepairRequired: true }); + expect(await db.select().from(activityLog).where(eq(activityLog.entityId, f.issueId))).toHaveLength(1); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + expect((await db.select().from(companies).where(eq(companies.id, f.companyId)))[0].budgetMonthlyCents).toBe(10000); + expect((await db.select().from(agents).where(eq(agents.id, f.agentId)))[0].budgetMonthlyCents).toBe(5000); + }); + it.each(["retained", "receipt", "lease", "owner", "coordinator", "copyback", "newer", "status", "version", "attachment", "result"] as const)("rejects stale or unsafe reconstruction: %s", async kind => { + const f = await seed(); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)); + if (kind === "retained") await db.update(environmentLeases).set({ metadata: { remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: f.providerLeaseId, state: "stopped" }) } }).where(eq(environmentLeases.id, f.leaseId)); + if (kind === "receipt") await db.update(environmentLeases).set({ metadata: { remoteExecutionTermination: { ...(lease.metadata!.remoteExecutionTermination as object), runId: randomUUID() } } }).where(eq(environmentLeases.id, f.leaseId)); + if (kind === "lease") await db.update(environmentLeases).set({ status: "pending_cleanup" }).where(eq(environmentLeases.id, f.leaseId)); + if (kind === "owner") await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.issueId)); + if (kind === "coordinator") await db.update(nativeRunFinalizations).set({ leaseOwner: "other-worker" }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (kind === "copyback") await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: { token: "other-worker" } } }).where(eq(heartbeatRuns.id, f.runId)); + if (kind === "newer") await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, contextSnapshot: { issueId: f.issueId }, status: "succeeded" }); + if (kind === "status") await db.update(issues).set({ status: "done" }).where(eq(issues.id, f.issueId)); + if (kind === "version") f.request.expectedStatusVersion += 1; + if (kind === "attachment") f.request.attachmentIds = [(await seed()).attachmentId]; + if (kind === "result") await db.update(nativeRunResults).set({ schemaStatus: "rejected" }).where(eq(nativeRunResults.id, f.resultId)); + await expect(reconstructNativeWorkspace(f)).rejects.toThrow(); + expect(await db.select().from(activityLog).where(eq(activityLog.entityId, f.issueId))).toHaveLength(0); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].evidence).not.toHaveProperty("executionReconciliation"); + }); + it("exposes board recovery through the assembled API and rejects missing loss acknowledgement", async () => { + const f = await seed(); + await request(app()).post(`/api/issues/${f.issueId}/recovery-actions/reconstruct-workspace`).send({ ...f.body, acknowledgeLostWork: false }).expect(400); + await request(app({ type: "agent", companyId: f.companyId, agentId: f.agentId, runId: f.runId })).post(`/api/issues/${f.issueId}/recovery-actions/reconstruct-workspace`).send(f.body).expect(403); + await request(app()).post(`/api/issues/${f.issueId}/recovery-actions/reconstruct-workspace`).send(f.body).expect(202); + }); + it.each(["status", "attachment", "paused"] as const)("keeps delivery pending when %s changes after reconstruction is recorded", async kind => { + const f = await seed(); + await reconstructNativeWorkspace(f); + if (kind === "status") await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, f.issueId)); + if (kind === "attachment") await db.delete(issueAttachments).where(eq(issueAttachments.id, f.attachmentId)); + if (kind === "paused") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + const heartbeat = heartbeatService(db, { runtimeEnv: {} }); + await deliverReconciledExecutions(db, heartbeat.wakeup); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, f.agentId))).toHaveLength(1); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].evidence.continuationDelivery).toBe("pending"); + }); + it("admits one fresh continuation through public heartbeat, keeps retry budget, and adopts a lost acknowledgement", async () => { + const f = await seed(); + await reconstructNativeWorkspace(f); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const heartbeat = heartbeatService(db, { runtimeEnv: {} }); + const wake = vi.fn(heartbeat.wakeup); + let successorId: string | undefined; + let admissionError: unknown; + await deliverReconciledExecutions(db, async (...args) => { + let successor; + try { successor = await wake(...args); } catch (error) { admissionError = error; throw error; } + expect(successor).not.toBeNull(); successorId = successor!.id; + expect(successor).toMatchObject({ retryOfRunId: f.runId, scheduledRetryAttempt: 2, contextSnapshot: { forceFreshSession: true, previousRunId: f.runId } }); + throw new Error("lost acknowledgement"); + }); + if (admissionError) throw admissionError; + const queued = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); + expect(successorId, JSON.stringify(queued.map(row => ({ status: row.status, reason: row.reason, payload: row.payload })))).toBeDefined(); + await deliverReconciledExecutions(db, wake); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.idempotencyKey, `execution-reconciliation:${f.actionId}`))).toHaveLength(1); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].evidence).toMatchObject({ continuationDelivery: "delivered", continuationRunId: successorId }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + }); + it.each(["unchanged", "blocked", "attachment", "paused", "forged_successor"] as const)("rechecks a queued reconstruction at the final provider handoff: %s", async kind => { + const f = await seed(); + await reconstructNativeWorkspace(f); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const heartbeat = heartbeatService(db, { runtimeEnv: {} }); + await deliverReconciledExecutions(db, heartbeat.wakeup); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + const successorId = String(action.evidence.continuationRunId); + expect(successorId).not.toBe("undefined"); + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, successorId)); + await db.update(issues).set({ status: "in_progress", executionRunId: successorId }).where(eq(issues.id, f.issueId)); + if (kind === "blocked") await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, f.issueId)); + if (kind === "attachment") await db.delete(issueAttachments).where(eq(issueAttachments.id, f.attachmentId)); + if (kind === "paused") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agentId)); + if (kind === "forged_successor") await db.update(issueRecoveryActions).set({ evidence: { ...action.evidence, continuationRunId: randomUUID() } }).where(eq(issueRecoveryActions.id, f.actionId)); + const dispatch = vi.fn(async () => "provider entered"); + const outcome = await createRunDispatch(db).dispatchResolvedInteractionIfCurrent({ companyId: f.companyId, + runId: successorId, expectedStatus: "running", dispatch }); + expect(outcome.dispatched).toBe(kind === "unchanged"); + expect(dispatch).toHaveBeenCalledTimes(kind === "unchanged" ? 1 : 0); + if (!outcome.dispatched) expect(outcome.cancellation).toMatchObject({ outcome: "cancelled" }); + }); + it.each([false, true])("runs the assembled heartbeat without overwriting a block after claim: %s", async blockAfterClaim => { + const f = await seed(); + await db.update(agents).set({ adapterType: "codex_local" }).where(eq(agents.id, f.agentId)); + await reconstructNativeWorkspace(f); + const [busy] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }).returning(); + const heartbeat = runningHeartbeat = heartbeatService(db, { runtimeEnv: { PAPERCLIP_IN_WORKTREE: "false" } }); + await deliverReconciledExecutions(db, heartbeat.wakeup); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + const successorId = String(action.evidence.continuationRunId); + expect(successorId).not.toBe("undefined"); + if (blockAfterClaim) await db.execute(sql.raw(` + create or replace function block_reconstruction_after_claim() returns trigger as $$ + begin + if NEW.id = '${successorId}'::uuid and NEW.status = 'running' then + update issues set status = 'blocked' where id = '${f.issueId}'::uuid; + end if; + return NEW; + end; $$ language plpgsql; + create trigger block_reconstruction_after_claim after update of status on heartbeat_runs + for each row execute function block_reconstruction_after_claim(); + `)); + try { + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, busy.id)); + await heartbeat.resumeQueuedRuns(); + const deadline = Date.now() + 15000; + let current; + do { + [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, successorId)); + if (["succeeded", "failed", "cancelled"].includes(current.status)) break; + await new Promise(resolve => setTimeout(resolve, 25)); + } while (Date.now() < deadline); + expect(providerExecute).toHaveBeenCalledTimes(blockAfterClaim ? 0 : 1); + if (blockAfterClaim) { + expect(current).toMatchObject({ status: "cancelled", errorCode: "execution_reconciliation_required" }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("blocked"); + } else expect(current.status).toBe("succeeded"); + } finally { + if (blockAfterClaim) await db.execute(sql`drop trigger block_reconstruction_after_claim on heartbeat_runs; drop function block_reconstruction_after_claim()`); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, f.issueId)); + } + }, 30_000); +}); diff --git a/server/src/services/native-runtime/native-workspace-reconstruction.ts b/server/src/services/native-runtime/native-workspace-reconstruction.ts new file mode 100644 index 00000000000..a238dd95b26 --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-reconstruction.ts @@ -0,0 +1,152 @@ +import { and, eq, gt, inArray, ne, or, sql } from "drizzle-orm"; +import { assets, environmentLeases, heartbeatRuns, issueAttachments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { reconstructWorkspaceSchema, type ExecutionReconciliation } from "@paperclipai/shared"; +import type { z } from "zod"; +import { conflict } from "../../errors.js"; +import { persistActivity, publishActivity } from "../activity-log.js"; +import { markExecutionReconciliation } from "../execution-recovery-resolution.js"; +import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js"; +import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; +import { nativeSha256 } from "./canonical.js"; + +const causes = ["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"]; +const object = (value: unknown): Record => value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +const changed = () => conflict("The lost workspace recovery changed. Refresh the task before continuing.", { code: "workspace_reconstruction_stale" }); + +export function unreconciledNativeExportPredicate() { + return and( + inArray(issueRecoveryActions.cause, causes), + or(and( + or(inArray(issueRecoveryActions.status, ["active", "escalated"]), and( + eq(issueRecoveryActions.status, "resolved"), eq(issueRecoveryActions.resolutionNote, "new_source_execution_path"), + )), + sql`${issueRecoveryActions.evidence}->'executionReconciliation'->'workspaceReconstruction' is null`, + ), and(eq(issueRecoveryActions.status, "resolved"), + sql`${issueRecoveryActions.evidence}->'executionReconciliation'->'workspaceReconstruction'->>'schema' = 'paperclip.workspace-reconstruction.v1'`, + sql`${issueRecoveryActions.evidence}->>'continuationDelivery' = 'pending'`, + )), + sql`exists (select 1 from ${heartbeatRuns} where ${heartbeatRuns.companyId} = ${issueRecoveryActions.companyId} + and ${heartbeatRuns.nativeIssueId} = ${issueRecoveryActions.sourceIssueId} + and ${heartbeatRuns.id}::text = ${issueRecoveryActions.evidence}->>'runId' + and ${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} = 'failed' + and ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'state' = 'prepared')`, + ); +} + +export async function validateNativeWorkspaceReconstruction(db: Db, input: { + companyId: string; issueId: string; runId: string; actionId: string; + successorRunId?: string; +}) { + const [issue] = await db.select().from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update"); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId))).for("update"); + const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))).for("update"); + const [action] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.sourceIssueId, input.issueId), eq(issueRecoveryActions.id, input.actionId))).for("update"); + if (!issue || !run || !coordinator || !action || run.runtimeMode !== "native" || run.nativeIssueId !== issue.id + || run.status !== "failed" || run.nativePhase !== "terminal_failure" || coordinator.issueId !== issue.id + || coordinator.phase !== "terminal_failure" || coordinator.leaseOwner || coordinator.failureDetail?.successorRunId + || object(run.runnerProfileJson).nativeWorkspaceFinalizationOwner + || action.kind !== "active_run_watchdog" || action.ownerType !== "board" || !causes.includes(action.cause) + || action.evidence.runId !== run.id || coordinator.failureCode !== action.cause + || !["active", "escalated", "resolved"].includes(action.status) + || (action.status === "resolved" && action.resolutionNote !== "new_source_execution_path" && !object(action.evidence.executionReconciliation).workspaceReconstruction) + || issue.assigneeAgentId !== run.agentId || action.returnOwnerAgentId !== run.agentId + || !(input.successorRunId ? ["todo", "in_progress"] : ["blocked", "todo"]).includes(issue.status) + || (issue.executionRunId && issue.executionRunId !== input.successorRunId) + || (issue.checkoutRunId && issue.checkoutRunId !== input.successorRunId)) throw changed(); + const reference = readNativeWorkspaceSyncReference(object(run.runnerProfileJson).nativeWorkspaceSync); + if (!reference || reference.state !== "prepared") throw changed(); + const leases = await db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.heartbeatRunId, run.id))).for("update"); + const lease = leases.find(row => row.id === reference.leaseId); + if (!lease || lease.issueId !== issue.id || lease.providerLeaseId !== reference.providerLeaseId + || !leases.every(hasRemoteTerminationReceipt) || object(lease.metadata?.remoteExecutionTermination).state !== "destroyed" + || leases.some(row => row.metadata?.pendingCleanupInFlight === true)) { + throw conflict("Paperclip has not confirmed destruction of the original workspace. Recover its files or retry its export first.", { code: "workspace_reconstruction_source_not_destroyed" }); + } + const [otherLease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider!), eq(environmentLeases.providerLeaseId, lease.providerLeaseId!), + inArray(environmentLeases.status, ["active", "retained", "pending_cleanup"]), + )).limit(1); + const [newer] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), ne(heartbeatRuns.id, run.id), + input.successorRunId ? ne(heartbeatRuns.id, input.successorRunId) : undefined, + or(eq(heartbeatRuns.nativeIssueId, issue.id), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`), + or(gt(heartbeatRuns.createdAt, run.createdAt), inArray(heartbeatRuns.status, ["queued", "running"])), + )).limit(1); + const [otherAction] = await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.sourceIssueId, issue.id), + ne(issueRecoveryActions.id, action.id), inArray(issueRecoveryActions.status, ["active", "escalated"]), + )).limit(1); + const [result] = await db.select().from(nativeRunResults).where(and( + eq(nativeRunResults.companyId, input.companyId), eq(nativeRunResults.issueId, issue.id), eq(nativeRunResults.runId, run.id), + eq(nativeRunResults.id, coordinator.resultId ?? "00000000-0000-0000-0000-000000000000"), + eq(nativeRunResults.completionContractId, run.completionContractId ?? "00000000-0000-0000-0000-000000000000"), eq(nativeRunResults.schemaStatus, "accepted"), + )).limit(1); + if (otherLease || newer || otherAction || !result) throw changed(); + const decision = action.evidence.executionReconciliation as ExecutionReconciliation | undefined; + if (decision?.workspaceReconstruction) { + const reconstruction = decision.workspaceReconstruction; + if ((!input.successorRunId && issue.status !== "todo") || decision.runId !== run.id || reconstruction.schema !== "paperclip.workspace-reconstruction.v1" + || (input.successorRunId && (action.evidence.continuationRunId !== input.successorRunId || action.evidence.continuationDelivery !== "delivered")) + || reconstruction.leaseId !== lease.id || reconstruction.resultId !== result.id + || reconstruction.descriptorSha256 !== reference.descriptorSha256 || !reconstruction.sources?.length) throw changed(); + const files = await db.select({ attachmentId: issueAttachments.id, sha256: assets.sha256 }).from(issueAttachments) + .innerJoin(assets, and(eq(assets.id, issueAttachments.assetId), eq(assets.companyId, input.companyId))) + .where(and(eq(issueAttachments.companyId, input.companyId), eq(issueAttachments.issueId, input.issueId), + inArray(issueAttachments.id, reconstruction.sources.map(source => source.attachmentId)))).for("share"); + if (files.length !== reconstruction.sources.length || reconstruction.sources.some(source => !files.some(file => file.attachmentId === source.attachmentId && file.sha256 === source.sha256))) throw changed(); + } + return { issue, run, action, reference, lease, result }; +} + +export async function validateReconstructedWorkspaceDispatch(db: Db, run: typeof heartbeatRuns.$inferSelect) { + const context = object(run.contextSnapshot); + if (typeof context.workspaceReconstructionActionId !== "string") return; + const actionId = context.workspaceReconstructionActionId; + if (typeof context.issueId !== "string" || typeof context.previousRunId !== "string" + || context.source !== "execution.reconciled" || context.recoveryActionId !== actionId || context.forceFreshSession !== true + || run.retryOfRunId !== context.previousRunId) throw changed(); + const source = await validateNativeWorkspaceReconstruction(db, { companyId: run.companyId, issueId: context.issueId, + runId: context.previousRunId, actionId, successorRunId: run.id }); + if (source.run.agentId !== run.agentId || !object(source.action.evidence.executionReconciliation).workspaceReconstruction) throw changed(); +} + +export async function reconstructNativeWorkspace(input: { + db: Db; companyId: string; issueId: string; actorId: string; + request: z.infer; +}) { + const request = reconstructWorkspaceSchema.parse(input.request); + const requestSha256 = nativeSha256(request); + const outcome = await input.db.transaction(async tx => { + const db = tx as unknown as Db; + const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update"); + const [prior] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.sourceIssueId, input.issueId), eq(issueRecoveryActions.id, request.actionId))); + if (prior?.evidence.workspaceReconstructionRequest === requestSha256) return { replayed: true, publication: null }; + if (!issue || issue.status !== "blocked" || issue.statusVersion !== request.expectedStatusVersion || prior?.evidence.executionReconciliation) throw changed(); + const source = await validateNativeWorkspaceReconstruction(db, { ...input, ...request }); + const files = await tx.select({ attachmentId: issueAttachments.id, sha256: assets.sha256, byteSize: assets.byteSize, filename: assets.originalFilename }) + .from(issueAttachments).innerJoin(assets, and(eq(assets.id, issueAttachments.assetId), eq(assets.companyId, input.companyId))) + .where(and(eq(issueAttachments.companyId, input.companyId), eq(issueAttachments.issueId, input.issueId), inArray(issueAttachments.id, request.attachmentIds))); + if (files.length !== request.attachmentIds.length || files.some(file => !/^[a-f0-9]{64}$/.test(file.sha256) || file.byteSize <= 0)) { + throw conflict("Choose saved attachments from this task before reconstructing its workspace.", { code: "workspace_reconstruction_sources_invalid" }); + } + const decision: ExecutionReconciliation = { runId: request.runId, providerStopped: true, + actionOutcome: request.actionOutcome, outcomeEvidence: request.outcomeEvidence, + workspaceReconstruction: { schema: "paperclip.workspace-reconstruction.v1", leaseId: source.lease.id, + resultId: source.result.id, descriptorSha256: source.reference.descriptorSha256, missingWork: request.missingWork, sources: files } }; + await markExecutionReconciliation(db, source.action, decision, input.actorId); + const now = new Date(); + await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "handed_back", resolvedAt: now, updatedAt: now, + resolutionNote: "workspace_reconstruction_authorized", nextAction: "Continue from saved sources. The original unexported edits are lost; verify the selected attachments before rebuilding. Do not repeat completed external actions or reset the original work limits.", + evidence: sql`${issueRecoveryActions.evidence} || ${JSON.stringify({ workspaceReconstructionRequest: requestSha256 })}::jsonb`, + wakePolicy: null, monitorPolicy: null, + }).where(eq(issueRecoveryActions.id, source.action.id)); + await tx.update(issues).set({ status: "todo", unblockDescriptor: null, blockedTransitionAt: null, + blockedOwnerNotifiedAt: null, updatedAt: now }).where(eq(issues.id, issue.id)); + const publication = await persistActivity(db, { companyId: input.companyId, actorType: "user", actorId: input.actorId, + action: "issue.workspace_reconstruction_authorized", entityType: "issue", entityId: input.issueId, runId: request.runId, + details: { recoveryActionId: request.actionId, ...decision } }); + return { replayed: false, publication }; + }); + if (outcome.publication) publishActivity(outcome.publication.publication); + return { runId: request.runId, actionId: request.actionId, status: "queued" as const, replayed: outcome.replayed }; +} diff --git a/server/src/services/satisfied-connection-intents.ts b/server/src/services/satisfied-connection-intents.ts index 49d295ae97f..f6a3eb39ef4 100644 --- a/server/src/services/satisfied-connection-intents.ts +++ b/server/src/services/satisfied-connection-intents.ts @@ -11,10 +11,10 @@ type Task = Pick & Partial>; /** Verify existing access only. This never installs tools, grants access, or adopts an identity. */ -export async function findSatisfiedToolConnection(db: Db, task: Task, intent: Intent, policyRunId?: string | null) { +async function findAvailableToolConnection(db: Db, task: Task, intent: Intent, policyRunId?: string | null) { if (intent.kind !== "connection_intent" || intent.status !== "pending" || intent.companyId !== task.companyId || intent.issueId !== task.id - || task.assigneeUserId || !["in_progress", "in_review"].includes(task.status)) return null; + || task.assigneeUserId) return null; const parsed = connectionIntentPayloadSchema.safeParse(intent.payload); if (!parsed.success) return null; const payload = parsed.data; @@ -42,6 +42,11 @@ export async function findSatisfiedToolConnection(db: Db, task: Task, intent: In return access.allowedCount + access.askFirstCount > 0 ? connection : null; } +export async function findSatisfiedToolConnection(db: Db, task: Task, intent: Intent, policyRunId?: string | null) { + if (!["in_progress", "in_review"].includes(task.status)) return null; + return findAvailableToolConnection(db, task, intent, policyRunId); +} + /** Recheck both live requests and system-retired cards against the next run. */ export async function connectionContinuationPendingResponse(db: Db, task: Task, responsibleUserId: string | null, policyRunId?: string | null): Promise<"interaction" | "approval" | null> { @@ -55,7 +60,7 @@ export async function connectionContinuationPendingResponse(db: Db, task: Task, if (interaction.status !== "pending" && !retired) continue; if (retired && (interaction.payload as { requestingAgentId?: string }).requestingAgentId !== task.assigneeAgentId) continue; if (interaction.addresseeUserId !== responsibleUserId) return "interaction"; - const connection = await findSatisfiedToolConnection(db, task, { ...interaction, status: "pending" }, policyRunId); + const connection = await findAvailableToolConnection(db, task, { ...interaction, status: "pending" }, policyRunId); // Historical system retirements describe a service need, not a permanent // binding to an account the user may later replace. Delivery and native // response projection still validate their own exact recorded connection. diff --git a/ui/src/api/issues.ts b/ui/src/api/issues.ts index 40881632c32..af057efd112 100644 --- a/ui/src/api/issues.ts +++ b/ui/src/api/issues.ts @@ -287,6 +287,9 @@ export const issuesApi = { api.post<{ runId: string; resultId: string; status: "queued" }>(`/issues/${id}/recovery-actions/resume-workspace-finalization`, data), retryWorkspaceExport: (id: string, data: { actionId: string; runId: string; repairNote: string }) => api.post<{ runId: string; resultId: string; leaseId: string; status: "queued" }>(`/issues/${id}/recovery-actions/retry-workspace-export`, data), + reconstructWorkspace: (id: string, data: { actionId: string; runId: string; expectedStatusVersion: number; attachmentIds: string[]; + missingWork: string; actionOutcome: "completed" | "not_performed" | "mixed"; outcomeEvidence: string; acknowledgeLostWork: true }) => + api.post<{ runId: string; actionId: string; status: "queued"; replayed: boolean }>(`/issues/${id}/recovery-actions/reconstruct-workspace`, data), recoveryActions: (id: string) => api.get<{ active: IssueRecoveryAction | null; actions: IssueRecoveryAction[] }>(`/issues/${id}/recovery-actions`), resolveRecoveryAction: ( diff --git a/ui/src/components/ExecutionBlockerNotice.tsx b/ui/src/components/ExecutionBlockerNotice.tsx index f6f82de1774..676d3c347fd 100644 --- a/ui/src/components/ExecutionBlockerNotice.tsx +++ b/ui/src/components/ExecutionBlockerNotice.tsx @@ -1,6 +1,7 @@ import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import type { ExecutionBlocker, Issue } from "@paperclipai/shared"; import { WorkspaceFileRecovery } from "./WorkspaceFileRecovery"; +import { WorkspaceReconstructionRecovery } from "./WorkspaceReconstructionRecovery"; import { agentsApi } from "../api/agents"; import { activityApi } from "../api/activity"; import { issuesApi } from "../api/issues"; @@ -50,15 +51,16 @@ export function ExecutionBlockerNotice({ companyId, issueId, issueStatus, canMan return (
-

{modelRejected ? "Model unavailable." : "Recovery needed."}{blocker.runError ? ` ${blocker.runError}` : ""}

+

{blocker.workspaceReconstructionPending ? "Reconstruction queued." : modelRejected ? "Model unavailable." : "Recovery needed."}{blocker.runError && !blocker.workspaceReconstructionPending ? ` ${blocker.runError}` : ""}

{modelRejected &&

Choose a supported model or clear the task's model override, then retry.

} {blocker.workspaceRepairRequired ? ( <> {/* The stored next action is an operator runbook step; lead with what happened. This hold also covers unsafe archives and local runs, so do not claim retries or a retained sandbox here. */}

- Paperclip could not restore the agent's workspace files after its last run. A board operator must - recover the files before the agent can continue. + {blocker.workspaceReconstructionPending ? blocker.nextAction : blocker.canReconstructWorkspace + ? "The workspace was destroyed before its edits were saved. A board operator can continue from surviving source attachments." + : "Paperclip could not restore the agent's workspace files after its last run. A board operator must recover the files before the agent can continue."}

Technical details @@ -94,6 +96,8 @@ export function ExecutionBlockerNotice({ companyId, issueId, issueStatus, canMan {recoveryError &&

Unable to load recovery details: {recoveryError.message}

} {issueStatus && } + {blocker.canReconstructWorkspace && }
); } diff --git a/ui/src/components/WorkspaceReconstructionRecovery.test.tsx b/ui/src/components/WorkspaceReconstructionRecovery.test.tsx new file mode 100644 index 00000000000..eb025f2f8a4 --- /dev/null +++ b/ui/src/components/WorkspaceReconstructionRecovery.test.tsx @@ -0,0 +1,74 @@ +// @vitest-environment jsdom +import { act } from "react"; +import { createRoot, type Root } from "react-dom/client"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { WorkspaceReconstructionRecovery } from "./WorkspaceReconstructionRecovery"; +import { queryKeys } from "../lib/queryKeys"; +const api = vi.hoisted(() => ({ listAttachments: vi.fn(), reconstructWorkspace: vi.fn() })); +vi.mock("../api/issues", () => ({ issuesApi: api })); + +describe("rebuild a destroyed workspace from partial sources", () => { + let root: Root, container: HTMLDivElement, client: QueryClient; + const onRecorded = vi.fn(); + beforeEach(() => { + vi.clearAllMocks(); + api.listAttachments.mockResolvedValue([{ id: "source", originalFilename: "baseline.ts", byteSize: 100 }]); + api.reconstructWorkspace.mockResolvedValue({ status: "queued" }); + container = document.createElement("div"); document.body.append(container); root = createRoot(container); + client = new QueryClient({ defaultOptions: { queries: { retry: false }, mutations: { retry: false } } }); + }); + afterEach(async () => { await act(async () => root.unmount()); client.clear(); container.remove(); }); + async function mount(canManage = true) { + await act(async () => root.render( + )); + await act(async () => { await new Promise(resolve => setTimeout(resolve, 10)); }); + } + async function fill() { + await act(async () => { + for (const area of container.querySelectorAll("textarea")) { + Object.getOwnPropertyDescriptor(HTMLTextAreaElement.prototype, "value")!.set!.call(area, "Verified source and receipts show what must be rebuilt without repeating published work."); + area.dispatchEvent(new Event("input", { bubbles: true })); + } + const checkboxes = container.querySelectorAll('[role="checkbox"]'); + checkboxes[0].click(); checkboxes[1].click(); + }); + } + it("requires surviving sources, lost-work acknowledgement and outcome evidence without claiming original-file recovery", async () => { + await mount(); + const button = Array.from(container.querySelectorAll("button")).find(button => button.textContent === "Rebuild and continue")!; + expect(button.disabled).toBe(true); + expect(container.textContent).toContain("missing edits have not been recovered"); + expect(container.textContent).toContain("Existing work limits still apply"); + await fill(); expect(button.disabled).toBe(false); + await act(async () => button.click()); + expect(api.reconstructWorkspace).toHaveBeenCalledWith("task", expect.objectContaining({ runId: "lost-run", expectedStatusVersion: 3, attachmentIds: ["source"], acknowledgeLostWork: true, actionOutcome: "mixed" })); + expect(onRecorded).toHaveBeenCalledOnce(); + }); + it("preserves the reconstruction decision when current source checks reject it", async () => { + api.reconstructWorkspace.mockRejectedValue(new Error("The source owner changed.")); + await mount(); await fill(); + await act(async () => Array.from(container.querySelectorAll("button")).find(button => button.textContent === "Rebuild and continue")!.click()); + await act(async () => { await new Promise(resolve => setTimeout(resolve, 10)); }); + expect(container.querySelector('[role="alert"]')?.textContent).toContain("source owner changed"); + expect(container.querySelector("textarea")?.value).toContain("Verified source"); + expect(onRecorded).not.toHaveBeenCalled(); + }); + it("offers no reconstruction form to a reader without runtime management authority", async () => { + await mount(false); expect(container.querySelector("button")).toBeNull(); + expect(api.listAttachments).not.toHaveBeenCalled(); expect(api.reconstructWorkspace).not.toHaveBeenCalled(); + }); + it("refreshes uploaded sources with the task attachment cache and drops deleted selections", async () => { + await mount(); await fill(); + await act(async () => client.setQueryData(queryKeys.issues.attachments("task"), [ + { id: "new-source", originalFilename: "uploaded-source.ts", byteSize: 200 }, + ])); + await act(async () => { await new Promise(resolve => setTimeout(resolve, 10)); }); + expect(container.textContent).toContain("uploaded-source.ts"); + expect(container.textContent).not.toContain("baseline.ts"); + const button = Array.from(container.querySelectorAll("button")).find(button => button.textContent === "Rebuild and continue")!; + expect(button.disabled).toBe(true); + expect(api.reconstructWorkspace).not.toHaveBeenCalled(); + }); +}); diff --git a/ui/src/components/WorkspaceReconstructionRecovery.tsx b/ui/src/components/WorkspaceReconstructionRecovery.tsx new file mode 100644 index 00000000000..b3a1da8bcf0 --- /dev/null +++ b/ui/src/components/WorkspaceReconstructionRecovery.tsx @@ -0,0 +1,67 @@ +import { useId, useState } from "react"; +import { useMutation, useQuery } from "@tanstack/react-query"; +import type { ExecutionBlocker } from "@paperclipai/shared"; +import { issuesApi } from "../api/issues"; +import { queryKeys } from "../lib/queryKeys"; +import { Button } from "./ui/button"; +import { Checkbox } from "./ui/checkbox"; +import { Label } from "./ui/label"; +import { Textarea } from "./ui/textarea"; +import { NativeSelect } from "./ui/select"; + +export function WorkspaceReconstructionRecovery({ issueId, blocker, canManage, onRecorded }: { + issueId: string; blocker: ExecutionBlocker; canManage: boolean; onRecorded: () => void; +}) { + const id = useId(); + const [selected, setSelected] = useState([]); + const [missingWork, setMissingWork] = useState(""); + const [outcomeEvidence, setOutcomeEvidence] = useState(""); + const [actionOutcome, setActionOutcome] = useState<"completed" | "not_performed" | "mixed">("mixed"); + const [acknowledged, setAcknowledged] = useState(false); + const files = useQuery({ queryKey: queryKeys.issues.attachments(issueId), + queryFn: () => issuesApi.listAttachments(issueId), enabled: canManage }); + const currentSelection = selected.filter(id => files.data?.some(file => file.id === id)); + const submit = useMutation({ + mutationFn: () => issuesApi.reconstructWorkspace(issueId, { + actionId: blocker.recoveryActionId!, runId: blocker.runId!, expectedStatusVersion: blocker.workspaceRecoveryStatusVersion!, + attachmentIds: currentSelection, missingWork: missingWork.trim(), actionOutcome, outcomeEvidence: outcomeEvidence.trim(), acknowledgeLostWork: true, + }), onSuccess: onRecorded, + }); + return
+ Continue from saved sources +

The original workspace was destroyed before its edits were saved. Its last run will remain failed.

+ {canManage ?
+

Select task attachments to rebuild from and record what remains missing. Paperclip starts a fresh session through its normal execution checks. Existing work limits still apply.

+
+ Saved source attachments + {files.data?.map(file =>
+ + setSelected(current => checked === true ? [...current, file.id] : current.filter(value => value !== file.id))} /> + +
)} + {files.isPending &&

Loading saved attachments…

} + {files.data?.length === 0 &&

Attach the surviving source files to this task first.

} + +