mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 22:24:16 +01:00
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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
790b729086
commit
7c7d2d8d2c
23 files changed
+724
-23
No files matched your search
@@ -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));
|
||||
|
||||
@@ -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<CancelStaleQueuedRunOutcome> {
|
||||
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<T>(
|
||||
input: DispatchResolvedInteractionInput<T>,
|
||||
): Promise<DispatchResolvedInteractionOutcome<T>> {
|
||||
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
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<boolean> {
|
||||
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<ExecutionBlocker | null> {
|
||||
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 } : {}),
|
||||
};
|
||||
}
|
||||
@@ -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"),
|
||||
|
||||
@@ -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(() => {}) };
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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<typeof import("../../adapters/index.js")>("../../adapters/index.js"),
|
||||
getServerAdapter: () => ({ supportsLocalAgentJwt: false, execute: providerExecute }),
|
||||
}));
|
||||
|
||||
describe("reconstruction after confirmed native workspace loss", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let runningHeartbeat: ReturnType<typeof heartbeatService> | 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);
|
||||
});
|
||||
@@ -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<string, unknown> => value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
|
||||
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<typeof reconstructWorkspaceSchema>;
|
||||
}) {
|
||||
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 };
|
||||
}
|
||||
@@ -11,10 +11,10 @@ type Task = Pick<typeof issues.$inferSelect, "id" | "companyId" | "projectId" |
|
||||
type Intent = Pick<typeof issueThreadInteractions.$inferSelect, "id" | "companyId" | "issueId" | "kind" | "status" | "payload" | "addresseeUserId"> & Partial<Pick<typeof issueThreadInteractions.$inferSelect, "sourceRunId">>;
|
||||
|
||||
/** 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.
|
||||
|
||||
Reference in new issue
Block a user