From ac22b1b1f9ce7d40feb540188ed4d799521c4e76 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 27 Oct 2026 11:02:43 -0500 Subject: [PATCH] fix: preserve recovered input and task directory continuity (#17878) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit https://entire.com/gh/paperclipai/-/trails/60/changes/paperclip/60 ## Thinking Path > - Paperclip lets people manage AI agents and their work. > - Recovery must preserve files and the authority of each human request. > - Some failed native runs preserve a snapshot without completing native copyback. > - The saved input could remain blocked after recovery settled. > - A later turn could also treat the new task directory as a Git base because it inherited the agent default. > - This change continues the same authorized input once and keeps its saved directory usable on later turns and Retry after a pre-provider preparation failure. ## Linked Issues or Issue Description Refs #14905. Related: #17724 covers cleanup handoff after native copyback. This change covers snapshot-only preservation and later turns in its directory. #17814 covers a separate same-run checkpoint restore defect. The open legacy-adapter change in #13291 does not cover this native saved-input path. ## What Changed - Forward the trusted deferred inbound request to native continuation admission. Verify the exact unconsumed receipt and its source, actor, company, agent, and task. - Let historical holds for the same failed run use its settled recovery strategy. Keep each controller and termination check. - Keep the old native sync proof unchanged. Attach recovered files through the verified snapshot path in a fresh attempt. - Recognize only the current same-task, active local directory in the canonical task or attempt namespace. Keep it on later turns instead of applying inherited Git defaults. - Keep the current healthy directory when canonical Retry follows the exact pre-allocation Git-base failure. Require the committed source link, strict server-produced failure events, no execution evidence or unfinished owner, and the current canonical binding. Recognize the known configuration warning without accepting arbitrary or provider-origin warnings. - Keep owned-directory readiness separate from repository-free strategy selection. A repaired real Git checkout with explicit task/project/assignee Git policy uses normal Git validation and preparation. Preserve explicit fresh, new sources, immutable native bindings, and locked path/binding checks. - Add real PostgreSQL and filesystem regressions for admission, file preservation, and the complete later-turn heartbeat policy chain. ## Verification - Real PostgreSQL callback/queue regressions are RED on unchanged production and GREEN with this repair: one successor from the same authenticated saved request, with filesystem snapshot preservation. The 24-case source/Retry matrix passes and retains termination, allocation, generation and revocation negatives. - Actual browser use exposed both later-turn Git-base failure and canonical Retry selecting an empty sibling. The real heartbeat selection/configuration/preparation/Retry regressions reproduce each. All 81 workspace-preparation cases passed before the final repaired-Git edge; all 17 final affected phase-chain/safety cases pass, including that new case. Its finalized oracle checks effective Git policy, an actual Git checkout and saved-file bytes. Known preparation warnings pass; unknown/provider warnings and execution contradictions do not certify reuse. - Full workspace typecheck, full build with matching `5342bd52da` stamp, maps and diff checks pass. Runtime commit `5342bd52da` passed CI with 53 successful checks and 2 conditional Storybook skips, and Greptile 5/5 with zero unresolved threads. The follow-up `9096fe8f62` removes only the feature/access map additions as requested; runtime and test files are byte-identical. Map structural and diff checks pass. Greptile reviewed this exact final commit at 5/5 with no unresolved threads. All test and build jobs passed. The final required `ci / verify` and `ci / e2e` aggregate gates are queued without runners; one bounded gate-only requeue retained every successful job. GitHub auto-merge is enabled and waits for those gates. - Normal UI acceptance on exact final source with actual OpenCode, configured DeepSeek model and Daytona passes both turns: snapshot continuation writes a useful review/report; reload and a second ordinary message retain the same workspace and logical native/provider session. All six files remain byte-identical, the report's computed SHA and unpredictable marker are confirmed, the concrete result is actual 250 versus correct 350, native finalization/copyback commit, and normal lease cleanup succeeds. A separate normal UI canonical Retry on preceding `9ac1d29147` preserves the current directory/report/marker and writes a continuity note; the final Git-policy split leaves that exercised non-Git path unchanged. - Boundaries: local app/server source with real remote native execution; seeded historical failure/stop receipts and actual stored snapshot, not a replay of the original staging allocation or external review publication. Prior local verification rewrote installed dependency metadata; its generated lock diff was retained and reverted. Final launch/end source and dependency metadata hashes match; no installed dependency byte-identity claim beyond that observed freeze. - Required full local suite was attempted but remains incomplete: embedded PostgreSQL bootstrap failures prevented assertions in several suites, so only its owned test tree was stopped. A prior workspace-recovery test timed out and the following seed collided with an active hold; both unchanged tests passed in isolation. Runtime-head `5342bd52da` hosted CI passed the full gate; the documentation-only follow-up waits for the aggregate jobs noted above. Its [first Rust attempt](https://github.com/paperclipai/paperclip/actions/runs/40528918372/job/122276159874) failed an unchanged ACPX shutdown cleanup assertion; the preceding same-Rust head passed. [One bounded Rust-only retry](https://github.com/paperclipai/paperclip/actions/runs/40528918372/job/122281990961) passed. No test assertion or production guard was weakened. ## Risks - Incorrect input authority could repeat work. The existing owner reauthorizes the same durable source; admission verifies its receipt under the task lock. - Directory recognition could select unrelated files. It requires the current scoped binding and canonical namespace, then preparation rechecks the stored binding and path under the source lock. Missing or symlinked retained roots are not recreated. - Snapshot preservation does not certify native copyback or old external outcomes. The old run and its unknown outcomes stay unchanged. Unconfirmed controllers or termination remain blocked. - There is no schema migration or new public API. ## Model Used OpenAI `gpt-6.1-sol` through Codex, with reasoning, code execution, and tool use. The host does not expose the 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 — targeted suites; the full local attempt limitation and final CI proof are documented 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 - [ ] 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/execution-semantics.md | 17 +++ .../chat-channels.integration.test.ts | 117 +++++++++++++--- .../services/explicit-native-continuation.ts | 27 +++- .../services/fresh-workspace-continuation.ts | 99 ++++++++++++- .../heartbeat/environment-selection.ts | 29 +++- server/src/services/heartbeat/queue.ts | 2 + .../heartbeat/workspace-preparation.test.ts | 131 +++++++++++++++++- .../src/services/isolated-task-directory.ts | 18 +++ .../src/services/workspace-recovery.test.ts | 30 +++- 9 files changed, 430 insertions(+), 40 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index b1408b10266..485cb5e9ad1 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1609,6 +1609,23 @@ ownership, active controllers, cleanup leases, pause, approval, budget, and norm execution gates still apply. Dependency-blocked interaction mode remains limited to its existing answer/triage contract. +Snapshot preservation and native copyback are distinct receipts. A historical +workspace recovered as an immutable snapshot does not make its old native sync +reference finalized. Once recovery settles, the current inbound owner can +reauthorize its same unconsumed durable human request and select a fresh isolated +workspace. The saved snapshot attaches to that successor through the existing +verified attachment path. A saved board-message queue uses the same fresh +strategy. Historical holds for the same failed run share that recovery strategy; +each still requires its own execution and termination proofs. Neither path +replays the stopped turn, adopts a chat request through the generic board queue, +nor invents a copyback hash or resets the predecessor's attempt count. +A later ordinary turn keeps the current source-free task directory, including a +fresh attempt directory. Inherited agent Git defaults do not turn that saved +directory into a Git base. Recognition requires the current same-task binding, +active local directory realization, and verified canonical path. Explicit new +source or Git requirements and immutable recovery bindings keep their existing +selection rules. Preparation rechecks the binding and path under the source lock. + The hold retirement, audit record, and new run commit together under the task lock. The new turn uses a fresh provider session and retains the latest user request, task history, completed work, and the interruption notice. It receives diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 05bca7807a0..d8d9720f2ca 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -2434,8 +2434,10 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await db.update(chatEndpoints).set({ botUsername: "gonzo[bot]", status: "active" }).where(eq(chatEndpoints.id, peer.id)); return peer; } - it.each(["callback", "restart", "race", "unknown_stop", "wrong_allocation", "paused_endpoint", "revoked_principal", "deleted_comment", "changed_generation", "foreign_scope", "changed_source", "revoked_before_valid"])( + it.each(["snapshot_callback", "snapshot_restart", "snapshot_race", "snapshot_unknown_stop", "snapshot_wrong_allocation", "snapshot_revoked_principal", "snapshot_changed_generation", "snapshot_changed_source", "callback", "restart", "race", "unknown_stop", "wrong_allocation", "paused_endpoint", "revoked_principal", "deleted_comment", "changed_generation", "foreign_scope", "changed_source", "revoked_before_valid"])( "reconsiders the same GitHub reply after stopped native workspace cleanup (%s)", async scenario => { + const snapshotScenario = scenario.startsWith("snapshot_"); + const safetyScenario = snapshotScenario ? scenario.slice("snapshot_".length) : scenario; const f = await subscribedGitHubFixture("github:paperclipai/paperclip:419"); const issueId = f.conversation.issueId!; const heartbeat = heartbeatService(db); @@ -2472,6 +2474,38 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { systemRecovery: { ...createSystemRecovery(sourceRunId), state: "complete", phase: "reconcile_effects" }, }, }); + if (snapshotScenario) { + const { workspaceOperations } = await import("@paperclipai/db"); + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "execution_finalization_deadline_exceeded", + runnerProfileJson: { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared", + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, + workspaceId: randomUUID(), leaseId, providerLeaseId, remoteCwd: "/workspace", resourceDisposition: "destroy" } }, + }).where(eq(heartbeatRuns.id, sourceRunId)); + const [recovery] = await db.select().from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, f.companyId), eq(issueRecoveryActions.sourceIssueId, issueId), + eq(issueRecoveryActions.cause, "native_continuation_requires_reconciliation"), + )); + const prefix = `${f.companyId}/workspace-recovery/${sourceRunId}/${leaseId}/`; + const [snapshot] = await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId, heartbeatRunId: sourceRunId, + phase: "workspace_recovery_snapshot", status: "succeeded", finishedAt: new Date(), metadata: { + schema: "paperclip.workspace-recovery-snapshot.v1", recoveryActionId: recovery.id, leaseId, providerLeaseId, + archive: { objectKey: `${prefix}${"c".repeat(64)}.tar`, sha256: "c".repeat(64), byteSize: 1024 }, + manifest: { objectKey: `${prefix}${"d".repeat(64)}.jsonl`, sha256: "d".repeat(64), byteSize: 512 }, + } }).returning(); + await db.update(issueRecoveryActions).set({ evidence: { ...recovery.evidence, systemRecovery: { + ...recovery.evidence.systemRecovery as Record, preservationOperationIds: [snapshot.id], + } } }).where(eq(issueRecoveryActions.id, recovery.id)); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", status: "resolved", ownerType: "system", outcome: "blocked", + cause: "execution_finalization_deadline_exceeded", fingerprint: randomUUID(), nextAction: "Inspect prior failure", + evidence: { runId: sourceRunId, automaticRecovery: { replay: "blocked" } }, + }); + const previousLease = { id: randomUUID(), companyId: f.companyId, issueId, heartbeatRunId: sourceRunId, + provider: "daytona", providerLeaseId: randomUUID(), status: "expired", cleanupStatus: "success", + releasedAt: new Date(), leasePolicy: "ephemeral" }; + await db.insert(environmentLeases).values({ ...previousLease, metadata: { remoteExecutionTermination: + remoteTerminationReceipt(previousLease, { providerLeaseId: previousLease.providerLeaseId, state: "destroyed" }) } }); + } f.wakeup.mockImplementation(heartbeat.wakeup); await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread: f.thread, trigger: "subscribed_message", message: makeMessage({ id: "41903", text: "A genuinely new author reply after Stop", @@ -2481,9 +2515,13 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { .filter(row => row.id !== initial.id); expect(receipt).toMatchObject({ status: "deferred_issue_execution", runId: null, payload: { executionWait: { reason: "remote_cleanup" } } }); + if (snapshotScenario) await db.update(heartbeatRuns).set({ + resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { + schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] } }, + }).where(eq(heartbeatRuns.id, sourceRunId)); let nextReceipt: typeof receipt | undefined; let nextCommentId: string | undefined; - if (scenario === "revoked_before_valid") { + if (safetyScenario === "revoked_before_valid") { const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:419:rc:4332012267" }).thread; const { id: _id, ...conversation } = f.conversation; await db.insert(chatConversations).values({ ...conversation, externalThreadId: thread.id }); @@ -2497,21 +2535,21 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, comment.id)); } const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); - if (scenario !== "unknown_stop") await db.update(environmentLeases).set({ status: "expired", cleanupStatus: "success", + if (safetyScenario !== "unknown_stop") await db.update(environmentLeases).set({ status: "expired", cleanupStatus: "success", releasedAt: new Date(), metadata: { remoteExecutionTermination: remoteTerminationReceipt(lease, - { providerLeaseId: scenario === "wrong_allocation" ? randomUUID() : providerLeaseId, state: "destroyed" }) } }) + { providerLeaseId: safetyScenario === "wrong_allocation" ? randomUUID() : providerLeaseId, state: "destroyed" }) } }) .where(eq(environmentLeases.id, leaseId)); - if (scenario === "paused_endpoint") await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, f.endpoint.id)); - if (scenario === "revoked_principal") await db.update(chatIdentityLinks).set({ status: "revoked", revokedAt: new Date() }) + if (safetyScenario === "paused_endpoint") await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, f.endpoint.id)); + if (safetyScenario === "revoked_principal") await db.update(chatIdentityLinks).set({ status: "revoked", revokedAt: new Date() }) .where(eq(chatIdentityLinks.principalId, f.principal.id)); - if (scenario === "deleted_comment") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, comment.id)); - if (scenario === "changed_generation") await db.update(chatConversations).set({ sessionGeneration: 2 }).where(eq(chatConversations.id, f.conversation.id)); - if (scenario === "changed_source") { + if (safetyScenario === "deleted_comment") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, comment.id)); + if (safetyScenario === "changed_generation") await db.update(chatConversations).set({ sessionGeneration: 2 }).where(eq(chatConversations.id, f.conversation.id)); + if (safetyScenario === "changed_source") { const [action] = await db.select().from(chatActions).where(eq(chatActions.id, receipt.id)); await db.update(chatActions).set({ payload: { ...action.payload, commentId: initial.payload?.commentId } }) .where(eq(chatActions.id, action.id)); } - if (scenario === "restart") { + if (safetyScenario === "restart") { await f.service.shutdown(); // Fresh application owner and scheduler must recover from saved data, // with no in-memory request object left in the retired chat service. @@ -2519,14 +2557,14 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } const source = (await readHeartbeatRunsByCompanyId(f.companyId)).find(row => row.id === sourceRunId)!; const { resumeDeferredChatWake } = await import("../services/durable-chat-wakeup.js"); - if (scenario === "foreign_scope") await resumeDeferredChatWake(db, { companyId: randomUUID(), agentId: f.assignedAgentId, + if (safetyScenario === "foreign_scope") await resumeDeferredChatWake(db, { companyId: randomUUID(), agentId: f.assignedAgentId, issueId, requestId: receipt.id }); - else if (scenario === "callback" || scenario === "revoked_before_valid") await heartbeat.resumeRemoteStopComments(source); - else if (scenario === "race") await Promise.all([heartbeat.resumeRemoteStopComments(source), heartbeat.resumeQueuedRuns()]); + else if (safetyScenario === "callback" || safetyScenario === "revoked_before_valid") await heartbeat.resumeRemoteStopComments(source); + else if (safetyScenario === "race") await Promise.all([heartbeat.resumeRemoteStopComments(source), heartbeat.resumeQueuedRuns()]); else await heartbeatService(db).resumeQueuedRuns(); const [settled] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.id)); const successors = (await readHeartbeatRunsByCompanyId(f.companyId)).filter(row => row.wakeupRequestId === receipt.id); - const admitted = ["callback", "restart", "race"].includes(scenario); + const admitted = ["callback", "restart", "race"].includes(safetyScenario); expect(successors).toHaveLength(admitted ? 1 : 0); expect(settled).toMatchObject({ id: receipt.id, idempotencyKey: receipt.idempotencyKey, requestedByActorId: receipt.requestedByActorId, requestedAt: receipt.requestedAt, @@ -2535,6 +2573,10 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(extractWakeCommentIds(successors[0].contextSnapshot)).toEqual([comment.id]); expect(successors[0].contextSnapshot?.chatConversationId).toBe(f.conversation.id); expect(successors[0].retryOfRunId).toBeNull(); + if (snapshotScenario) expect(successors[0].contextSnapshot?.explicitUserContinuation).toMatchObject({ + previousRunId: sourceRunId, commentId: comment.id, + readiness: { mode: "fresh_isolated_workspace", retryIntentId: receipt.id, sourceRunId }, + }); await heartbeat.resumeRemoteStopComments(source); await heartbeat.resumeQueuedRuns(); expect((await readHeartbeatRunsByCompanyId(f.companyId)).filter(row => row.wakeupRequestId === receipt.id)).toHaveLength(1); @@ -2545,7 +2587,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(extractWakeCommentIds(admittedBehindRevoked[0].contextSnapshot)).toEqual([nextCommentId]); } expect((await readHeartbeatRunsByCompanyId(f.companyId)).find(row => row.id === sourceRunId)) - .toMatchObject({ status: "cancelled", resultJson: { workspaceRestoreFailure: "restore_failed" } }); + .toMatchObject({ status: snapshotScenario ? "failed" : "cancelled", resultJson: { workspaceRestoreFailure: "restore_failed" } }); }, 40_000, ); @@ -61065,17 +61107,52 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return { ...f, pending, originalDelivery }; } - it.each(["reconnected", "newer_input", "reconnected_newer_input"] as const)( + it.each(["reconnected", "newer_input", "reconnected_newer_input", "snapshot_newer_input"] as const)( "admits current GitHub task Retry without replaying obsolete input: %s", async kind => { const f = await currentGitHubTaskRetryFixture(kind.includes("newer_input")); const { pending, originalDelivery } = f; try { const companyId = f.fixture.companyId, issueId = f.issue.id; if (kind.includes("reconnected")) await f.service.configure(f.endpoint.id, { action: "reconnect" }, "owner-user"); + let snapshotActionId: string | undefined; + if (kind === "snapshot_newer_input") { + const leaseId = randomUUID(), providerLeaseId = randomUUID(); + await db.update(heartbeatRuns).set({ errorCode: "execution_finalization_deadline_exceeded", + resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { + schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] } }, + runnerProfileJson: { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared", + descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, + workspaceId: randomUUID(), leaseId, providerLeaseId, remoteCwd: "/workspace", resourceDisposition: "destroy" } }, + }).where(eq(heartbeatRuns.id, f.runId)); + for (const current of [{ id: leaseId, providerLeaseId }, { id: randomUUID(), providerLeaseId: randomUUID() }]) { + const lease = { ...current, companyId, issueId, heartbeatRunId: f.runId, + provider: "daytona", status: "released", cleanupStatus: "success", releasedAt: new Date() }; + await db.insert(environmentLeases).values({ ...lease, metadata: { remoteExecutionTermination: + remoteTerminationReceipt(lease, { providerLeaseId: current.providerLeaseId, state: "stopped" }) } }); + } + for (const cause of ["native_continuation_requires_reconciliation", "execution_finalization_deadline_exceeded"]) { + const id = randomUUID(); + if (cause === "native_continuation_requires_reconciliation") snapshotActionId = id; + await db.insert(issueRecoveryActions).values({ id, companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", status: "resolved", ownerType: "system", outcome: "blocked", cause, + fingerprint: randomUUID(), nextAction: "Preserved snapshot needs fresh continuation", evidence: { + runId: f.runId, automaticRecovery: { replay: "blocked" }, + ...(cause === "native_continuation_requires_reconciliation" ? { + systemRecovery: { ...createSystemRecovery(f.runId), state: "complete", phase: "reconcile_effects" }, + } : {}), + } }); + } + } const input = { db, companyId, issueId, actorId: "owner-user", actorSource: "cloud_tenant", observedRunId: f.runId, idempotencyKey: randomUUID(), prepareChatRetry: f.service.prepareFailedChatRunRetry }; const request = await requestTaskRetry(input); expect(request.accepted).toBe(true); + if (snapshotActionId) { + const [saved] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, snapshotActionId)); + await db.update(issueRecoveryActions).set({ evidence: { ...saved.evidence, + systemRecovery: { ...saved.evidence.systemRecovery as Record, intentId: request.requestId }, + } }).where(eq(issueRecoveryActions.id, snapshotActionId)); + } expect((await requestTaskRetry(input)).requestId).toBe(request.requestId); expect(await f.service.processFailedChatRunRetry(request.chatActionId!)).toMatchObject({ status: "queued" }); const [wake] = await readAgentWakeupRequestsById(request.chatActionId!); @@ -61084,6 +61161,14 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const [run] = await readHeartbeatRunsById(wake.runId!); expect(run).toMatchObject({ status: "queued", retryOfRunId: f.runId, contextSnapshot: { wakeCommentIds: [expectedComment], taskRetryRequestId: request.requestId } }); + if (snapshotActionId) { + const { readFreshWorkspaceReadiness } = await import("../services/fresh-workspace-continuation.js"); + expect(await readFreshWorkspaceReadiness(db, run)).toMatchObject({ + mode: "fresh_isolated_workspace", sourceRunId: f.runId, retryIntentId: request.requestId, + }); + expect((await readHeartbeatRunsById(f.runId))[0].runnerProfileJson?.nativeWorkspaceSync) + .toMatchObject({ state: "prepared", finalHostSha256: null }); + } if (pending) expect((await readAgentWakeupRequestsById(pending.id))[0]) .toMatchObject({ status: "coalesced", runId: wake.runId }); await f.service.processFailedChatRunRetry(request.chatActionId!); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 904bf270097..4ede5b10afc 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -1,3 +1,4 @@ +import { assertDurableChatWakeupRequest, assertDurableChatWakeupReceipt, type DurableChatWakeupRequest } from "./durable-chat-wakeup.js"; import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { canSelectFreshWorkspace, type ContinuationReadiness } from "./fresh-workspace-continuation.js"; import { readTaskRetryIntent } from "./task-retry.js"; @@ -92,6 +93,7 @@ export async function admitExplicitNativeContinuation(input: { queuedCommentInterruptId?: string; /** Internal delivery of an unconsumed, user-authored saved queue entry. */ queuedCommentRequestId?: string; + durableChatRequest?: DurableChatWakeupRequest; dryRun?: boolean; onBlocked?: (reason: string, message: string) => void; }): Promise<{ previousRunId: string; commentId: string | null; failedRunId?: string; taskRetryRequestId?: string; readiness?: ContinuationReadiness } | null> { @@ -136,6 +138,21 @@ export async function admitExplicitNativeContinuation(input: { !savedQueue.idempotencyKey?.startsWith("chat-inbound:") && queuedCommentIdsFromWakePayload(savedQueue.payload).includes(commentId)); if (input.queuedCommentRequestId && !queuedRequest) return null; + const chatRequest = input.durableChatRequest; + let savedChatRequestId: string | undefined; + if (chatRequest) { + assertDurableChatWakeupRequest(chatRequest, { companyId, issueId, agentId, commentId, + requestedByActorType: input.actorType ?? undefined, requestedByActorId: actorId }); + if (chatRequest.failedRunRetry || !chatRequest.resumeDeferred) return null; + const [receipt] = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.id, chatRequest.id), eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"), + isNull(agentWakeupRequests.runId), sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + )); + if (!receipt || !commentId || !queuedCommentIdsFromWakePayload(receipt.payload).includes(commentId)) return null; + assertDurableChatWakeupReceipt(chatRequest, receipt); + savedChatRequestId = receipt.id; + } let partiallyDeliveredQueue = false; if (queuedRequest) { const ids = queuedCommentIdsFromWakePayload(savedQueue!.payload); @@ -199,20 +216,18 @@ export async function admitExplicitNativeContinuation(input: { const unusedAdmission = run.status === "cancelled" && !run.startedAt && run.errorCode === "execution_reconciliation_required" && !run.processPid && !run.processGroupId && !run.nativeSessionId; - // Several retained holds may describe this same failed execution. Reuse - // its current explicit Retry strategy for a historical hold that predates - // system recovery. Every hold still passes the stop/effect checks below. const sourceRunId = run.id; - const strategyEvidence = input.taskRetryRequestId && !Object.hasOwn(action.evidence, "systemRecovery") + const strategyRequestId = input.taskRetryRequestId ?? input.queuedCommentRequestId ?? savedChatRequestId; + const strategyEvidence = strategyRequestId && !Object.hasOwn(action.evidence, "systemRecovery") ? actions.find(candidate => { const recovery = readSystemRecovery(candidate.evidence); return candidate.evidence.runId === sourceRunId && recovery?.sourceRunId === sourceRunId && - recovery.intentId === input.taskRetryRequestId; + (input.taskRetryRequestId ? recovery.intentId === input.taskRetryRequestId : true); })?.evidence ?? action.evidence : action.evidence; const freshIntent = await canSelectFreshWorkspace({ db, companyId, issueId, agentId, actorId, sourceRunId: run.id, evidence: strategyEvidence, taskRetryRequestId: input.taskRetryRequestId, - queuedRequestId: input.queuedCommentRequestId }); + queuedRequestId: input.queuedCommentRequestId ?? savedChatRequestId }); const legacyUserTurn = run.runtimeMode === "legacy" && action.cause === "legacy_execution_requires_reconciliation" && (isConversationAdapter(agent.adapterType) || diff --git a/server/src/services/fresh-workspace-continuation.ts b/server/src/services/fresh-workspace-continuation.ts index f37754acf9a..25ac8556975 100644 --- a/server/src/services/fresh-workspace-continuation.ts +++ b/server/src/services/fresh-workspace-continuation.ts @@ -1,13 +1,18 @@ -import { and, eq, sql } from "drizzle-orm"; -import { agentWakeupRequests, environmentLeases, heartbeatRuns, issueRecoveryActions, type Db } from "@paperclipai/db"; +import { and, asc, eq, sql } from "drizzle-orm"; +import { agentWakeupRequests, environmentLeases, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; import { verifiedWorkspaceRepairReceipt } from "./workspace-recovery-receipt.js"; import { persistActivity } from "./activity-log.js"; import { readSystemRecovery } from "./system-recovery-state.js"; import { readTaskRetryIntent } from "./task-retry.js"; import { z } from "zod"; +import { adapterExecutionControls } from "./adapter-execution-control.js"; +import { hasPendingRunCleanup } from "./execution-blocker.js"; +import { toExecutionWorkspace } from "./execution-workspaces.js"; +import { isBoundIsolatedTaskDirectory } from "./isolated-task-directory.js"; export type ContinuationReadiness = | { mode: "resume_preserved_workspace"; preservationReceiptId: string } + | { mode: "reuse_current_workspace"; retryIntentId: string; sourceRunId: string; workspaceId: string; cwd: string } | { mode: "fresh_isolated_workspace"; retryIntentId: string; sourceRunId: string; retainedSourceIds: string[] }; const freshReadinessSchema = z.object({ mode: z.literal("fresh_isolated_workspace"), retryIntentId: z.string().uuid(), @@ -47,6 +52,76 @@ export async function canSelectFreshWorkspace(input: { return ["complete", "exhausted"].includes(system.state) ? wake.id : null; } +async function reusableDirectoryAfterPreparationFailure(db: Db, source: typeof heartbeatRuns.$inferSelect, issueId: string) { + const validation = source.resultJson?.workspaceValidation as Record | undefined; + const recovery = source.resultJson?.executionRecovery as Record | undefined; + if (source.status !== "failed" || !source.finishedAt || source.runtimeMode !== "legacy" || + source.errorCode !== "workspace_validation_failed" || source.executionStage !== "preparing" || + validation?.reason !== "git_worktree_base_not_git_checkout" || validation.issueId !== issueId || + validation.resolvedWorkspaceSource !== "task_session" || validation.requestedExecutionWorkspaceMode !== "isolated_workspace" || + validation.issueExecutionWorkspacePreference !== "reuse_existing" || recovery?.kind !== "bootstrap" || + recovery.providerWorkStarted !== false || Object.keys(recovery).some(key => !["kind", "providerWorkStarted"].includes(key)) || + source.resultJson?.stopReason !== "adapter_failed" || + adapterExecutionControls.has(source.id) || (source.controllerLeaseExpiresAt && source.controllerLeaseExpiresAt > new Date()) || + [source.runtimeModeResolvedAt, source.runtimeModeResolverVersion, source.runtimeModeReason, source.runnerInstanceId, + source.nativeIssueId, source.nativePhase, source.nativeSessionId, source.driverKind, source.driverVersion, + source.completionContractId, source.completionContractSha256, source.sessionIdAfter, source.externalRunId, + source.processPid, source.processGroupId, source.processStartedAt, source.logStore, source.logRef, + source.logBytes, source.logSha256, source.stdoutExcerpt, source.stderrExcerpt, source.lastOutputAt, + source.lastOutputStream, source.lastOutputBytes, source.exitCode, source.signal].some(value => value !== null) || + source.lastOutputSeq !== 0 || source.logCompressed || + Object.keys(source.usageJson ?? {}).some(key => key !== "ledgerScope") || + Object.keys(source.runnerProfileJson ?? {}).some(key => !["adapterDispatch", "chatControlRecoveryAdmission", "aiConnectionNonAssigneeCommentWake"].includes(key)) || + Object.keys(source.resultJson ?? {}).some(key => !["effectiveTimeoutSec", "executionRecovery", "stopReason", "timeoutConfigured", "timeoutFired", "timeoutSource", "workspaceValidation"].includes(key))) return null; + const events = await db.select().from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, source.companyId), eq(heartbeatRunEvents.runId, source.id), + )).orderBy(asc(heartbeatRunEvents.seq)).limit(4); + let offset = 0; + const warning = events[0]; + if (warning?.eventType === "agent_observability.unavailable") { + if (warning.seq !== 1 || warning.level !== "warn" || + warning.message !== "Company observability was not applied: this harness is not supported." || + warning.payload?.reason !== "unsupported_harness" || !z.string().uuid().safeParse(warning.payload.revision).success || + Object.keys(warning.payload).some(key => !["reason", "revision"].includes(key))) return null; + offset = 1; + } + const diagnostic = events[offset], measured = events[offset + 1]; + if (events.length !== 2 + offset || !diagnostic || diagnostic.seq !== 1 + offset || + diagnostic.eventType !== "error" || diagnostic.level !== "error" || diagnostic.message !== source.error || + diagnostic.payload !== null || !measured || measured.seq !== 2 + offset || + measured.eventType !== "lifecycle" || measured.level !== "info" || measured.message !== "task.run.measured" || + measured.payload?.phase !== "task.run.measured" || measured.payload?.outcome !== "failed" || + measured.payload?.durableSettlement !== true || typeof measured.payload.durationMs !== "number" || + !Number.isFinite(measured.payload.durationMs) || measured.payload.durationMs < 0 || + Object.keys(measured.payload).some(key => !["phase", "outcome", "durableSettlement", "durationMs"].includes(key)) || + events.some(event => event.stream !== "system" || event.agentId !== source.agentId || + event.sourceInstanceId !== null || event.sourceEventId !== null || event.sourceSeq !== null || + event.sourcePayloadSha256 !== null || event.protocolSchemaVersion !== null)) return null; + const [executionEvidence] = await db.execute<{ present: boolean }>(sql`select + exists(select 1 from ${environmentLeases} where ${environmentLeases.companyId} = ${source.companyId} and ${environmentLeases.heartbeatRunId} = ${source.id}) + or exists(select 1 from ${nativeRunResults} where ${nativeRunResults.companyId} = ${source.companyId} and ${nativeRunResults.runId} = ${source.id}) + or exists(select 1 from ${nativeRunFinalizations} where ${nativeRunFinalizations.companyId} = ${source.companyId} and ${nativeRunFinalizations.runId} = ${source.id}) as present`); + if (executionEvidence?.present !== false) return null; + const [task] = await db.select().from(issues).where(and(eq(issues.companyId, source.companyId), eq(issues.id, issueId))); + if (!task || task.assigneeAgentId !== source.agentId || task.executionWorkspacePreference !== "reuse_existing" || + task.executionWorkspaceId !== validation.issueExecutionWorkspaceId || task.projectId !== validation.issueProjectId || + task.projectWorkspaceId !== validation.issueProjectWorkspaceId) return null; + const [workspace] = await db.select().from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, source.companyId), eq(executionWorkspaces.id, task.executionWorkspaceId!), + )); + if (!workspace || workspace.cwd !== validation.resolvedWorkspaceCwd || !await isBoundIsolatedTaskDirectory({ + companyId: source.companyId, issueId, workspaceId: task.executionWorkspaceId, workspace: toExecutionWorkspace(workspace) })) return null; + const prior = await db.select({ runId: environmentLeases.heartbeatRunId }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, source.companyId), eq(environmentLeases.executionWorkspaceId, workspace.id), + )); + for (const { runId } of prior) if (runId && await hasPendingRunCleanup(db, source.companyId, runId)) return null; + const [operation] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, source.companyId), eq(workspaceOperations.executionWorkspaceId, workspace.id), + eq(workspaceOperations.status, "running"), + )).limit(1); + return operation ? null : workspace; +} + /** Preparation trusts the committed admission receipt, not a raw wake payload. */ export async function readFreshWorkspaceReadiness(db: Db, run: Pick): Promise { @@ -63,10 +138,6 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick>'runId' = ${issueRecoveryActions.evidence}#>>'{explicitUserContinuation,readiness,sourceRunId}'`, )).limit(1); if (!action) { - // Normal dispatch can admit a retry after a pre-provider failure without a - // recovery hold. The committed wake is its authority. It still needs an - // independent directory: falling back to the agent's configured cwd would - // reopen the original source after a failed fresh attempt. const [wake] = await db.select().from(agentWakeupRequests).where(and( eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.reason, "task_retry"), @@ -96,6 +167,22 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick { + const changed = await tx.update(agentWakeupRequests).set({ + payload: sql`jsonb_set(${agentWakeupRequests.payload}, '{taskRetry,strategy}', '"continue"'::jsonb)`, + }).where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.runId, run.id), + sql`${agentWakeupRequests.payload}#>>'{taskRetry,strategy}' = 'evaluate'`)).returning({ id: agentWakeupRequests.id }); + if (changed.length) await persistActivity(tx as unknown as Db, { companyId: run.companyId, + actorType: "system", actorId: "task-recovery", action: "issue.retry_workspace_reused", + entityType: "issue", entityId: issueId, runId: run.id, details: readiness }); + }); + return readiness; + } } const leases = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, source.id), diff --git a/server/src/services/heartbeat/environment-selection.ts b/server/src/services/heartbeat/environment-selection.ts index 9910ab0b5b7..484a6937907 100644 --- a/server/src/services/heartbeat/environment-selection.ts +++ b/server/src/services/heartbeat/environment-selection.ts @@ -17,7 +17,7 @@ import { type ParsedExecutionWorkspaceMode, } from "../execution-workspace-policy.js"; import { assertTaskWorkspaceExecutionAccess, assertTaskWorkspaceSourceProjectAccess } from "../task-workspace-access.js"; -import { shouldUseIsolatedTaskDirectory } from "../isolated-task-directory.js"; +import { isBoundIsolatedTaskDirectory, shouldUseIsolatedTaskDirectory } from "../isolated-task-directory.js"; import { readTaskDirectoryRetention } from "../isolated-task-directory-retention.js"; import { findNativeChatWorkspaceScope, @@ -32,7 +32,7 @@ import type { TrustPresetResolution } from "../trust-preset-resolver.js"; import { WORKSPACE_BUSY_RETRY_REASON } from "../../modules/run-dispatch/index.js"; import { WorkspaceBusyDeferral, type createHeartbeatRetries } from "./retries.js"; import { ConfigurationIncompleteFailure } from "./run-preparation.js"; -import { resolveNativeRecoveryExecutionWorkspaceBinding, resolveExecutionWorkspaceReuseRequestForIssue } from "./workspaces.js"; +import { resolveNativeRecoveryExecutionWorkspaceBinding, resolveExecutionWorkspaceReuseRequestForIssue, WorkspaceValidationFailure } from "./workspaces.js"; type Issue = typeof issues.$inferSelect; @@ -138,6 +138,13 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron } existingExecutionWorkspace = await executionWorkspacesSvc.getStoredById(agent.companyId, existingExecutionWorkspace.id); } + const retainedDirectory = input.task.freshWorkspace?.mode === "reuse_current_workspace" ? input.task.freshWorkspace : null; + if (retainedDirectory && (issueRef?.executionWorkspaceId !== retainedDirectory.workspaceId || + existingExecutionWorkspace?.id !== retainedDirectory.workspaceId || existingExecutionWorkspace.cwd !== retainedDirectory.cwd)) { + throw new WorkspaceValidationFailure("The saved Retry workspace binding changed.", { + workspaceValidation: { reason: "isolated_task_directory_binding_mismatch", issueId }, + }); + } const nativeRecoveryExecutionWorkspaceId = resolveNativeRecoveryExecutionWorkspaceBinding({ bindingId: persistedNativeExecutionWorkspaceId, @@ -471,7 +478,23 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron preserveImplicitDefaultSettings: ["project", "task_default"].includes(taskWorkspaceIntent.source) }); } const explicitTaskDirectory = !isDotRun && taskWorkspaceIntent?.selection.kind === "task_directory"; - const useIsolatedTaskDirectory = !isDotRun && issueRef !== null && ((fresh && !workspaceProjectContext?.hasWorkspace && !issueRef.projectWorkspaceId) || shouldUseIsolatedTaskDirectory({ + const explicitDirectoryStrategies = [issueAssigneeOverrides?.adapterConfig?.workspaceStrategy, + projectExecutionWorkspacePolicy?.workspaceStrategy, issueExecutionWorkspaceSettings?.workspaceStrategy]; + const boundTaskDirectory = !isDotRun && !fresh && !nativeRecoveryExecutionWorkspaceId && issueId !== null && issueRef !== null && + requestedShouldReuseExisting && requestedExecutionWorkspaceMode === "isolated_workspace" && + !workspaceProjectContext?.hasWorkspace && !issueRef.projectWorkspaceId && + taskWorkspaceIntent?.selection.kind !== "configured_source" && + (taskWorkspaceIntent?.selection.kind !== "existing" || taskWorkspaceIntent.selection.workspaceId === issueRef.executionWorkspaceId) && + await isBoundIsolatedTaskDirectory({ companyId: agent.companyId, issueId, + workspaceId: issueRef.executionWorkspaceId, workspace: reusableExistingExecutionWorkspace }); + const reuseTaskDirectory = boundTaskDirectory && explicitDirectoryStrategies.every(value => { + const strategy = parseObject(value); + return Object.keys(strategy).length === 0 || (Object.keys(strategy).length === 1 && strategy.type === "project_primary"); + }); + if (retainedDirectory && !boundTaskDirectory) throw new WorkspaceValidationFailure("The saved Retry task directory is unavailable.", { + workspaceValidation: { reason: "isolated_task_directory_binding_mismatch", issueId }, + }); + const useIsolatedTaskDirectory = !isDotRun && issueRef !== null && (reuseTaskDirectory || (fresh && !workspaceProjectContext?.hasWorkspace && !issueRef.projectWorkspaceId) || shouldUseIsolatedTaskDirectory({ trustPreset: trustPreset.kind, environmentDriver: selectedEnvironmentForConfig?.driver ?? null, mode: requestedExecutionWorkspaceMode, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index debcd1c18e0..17590f7be12 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -3382,6 +3382,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende queuedCommentInterruptId: opts.queuedCommentInterruptId, queuedCommentRequestId: opts.queuedCommentRequestId, taskRetryRequestId: opts.taskRetryRequestId, + ...(durableRequest?.resumeDeferred && !failedChatRetry ? { durableChatRequest: durableRequest } : {}), dryRun: true, onBlocked: (reason, message) => { continuationWait = { reason, message }; }, }))) return deferBlockedExecution(executionBlocker); @@ -4182,6 +4183,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende queuedCommentInterruptId: opts.queuedCommentInterruptId, queuedCommentRequestId: opts.queuedCommentRequestId, taskRetryRequestId: opts.taskRetryRequestId, + ...(durableRequest?.resumeDeferred && !failedChatRetry ? { durableChatRequest: durableRequest } : {}), }); if (!explicitContinuation && executionBlocker && !ownsReconstructionContinuation) return deferBlockedExecution(executionBlocker); if (explicitContinuation) { diff --git a/server/src/services/heartbeat/workspace-preparation.test.ts b/server/src/services/heartbeat/workspace-preparation.test.ts index d3c9944abd5..6192dccbae3 100644 --- a/server/src/services/heartbeat/workspace-preparation.test.ts +++ b/server/src/services/heartbeat/workspace-preparation.test.ts @@ -1,16 +1,19 @@ import { execFile as execFileCallback } from "node:child_process"; import { randomUUID } from "node:crypto"; -import { mkdtemp, mkdir, readFile, readdir, rm, writeFile, access, rename } from "node:fs/promises"; +import { mkdtemp, mkdir, readFile, readdir, rm, writeFile, access, rename, symlink } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { pathToFileURL } from "node:url"; import { promisify } from "node:util"; import { captureDirectorySnapshot, directorySnapshotSha256, serializeDirectorySnapshot } from "@paperclipai/adapter-utils/workspace-restore-merge"; import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; -import { agentWakeupRequests, issueRecoveryActions, agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environmentLeases, executionWorkspaces, executionWorkspaceRepositoryAccess, heartbeatRuns, issues, issueAccessGrants, plugins, projects, projectWorkspaces, type Db } from "@paperclipai/db"; +import { agentWakeupRequests, heartbeatRunEvents, companyObservability, issueRecoveryActions, agents, authUsers, companyMemberships, companies, createDb, applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environmentLeases, executionWorkspaces, executionWorkspaceRepositoryAccess, heartbeatRuns, issues, issueAccessGrants, plugins, projects, projectWorkspaces, type Db } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { heartbeatService } from "../heartbeat.js"; +import { registerServerAdapter, getServerAdapter, type AdapterExecutionContext } from "../../adapters/index.js"; import { environmentRunOrchestrator } from "../environment-run-orchestrator.js"; import { environmentService } from "../environments.js"; import { executionWorkspaceService, toExecutionWorkspace } from "../execution-workspaces.js"; @@ -22,6 +25,7 @@ import { prepareHeartbeatWorkspace, type HeartbeatWorkspacePreparationInput } fr import { buildNativeExecutionInput } from "../native-runtime/native-execution-input.js"; import { nativeRuntimeContextFixture } from "../native-runtime/runtime-context.test-fixture.js"; import * as managedRepositoryCheckout from "../managed-repository-checkout.js"; +import { requestTaskRetry } from "../task-retry.js"; import { readFreshWorkspaceReadiness } from "../fresh-workspace-continuation.js"; import { createSystemRecovery } from "../system-recovery-state.js"; import { isolatedRemoteSourcePath } from "../isolated-remote-workspace-source.js"; @@ -35,9 +39,10 @@ import { createStorageService } from "../../storage/service.js"; import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace"; const execFile = promisify(execFileCallback); -const support = await getEmbeddedPostgresTestSupport(); +const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_WORKSPACE_PREPARATION_TEST_DATABASE_URL"); +const support = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); -describe.skipIf(!support.supported)("heartbeat workspace preparation boundary", () => { +describe.skipIf(!support.supported && !externalDatabaseUrl)("heartbeat workspace preparation boundary", () => { let database: Awaited>; let db: Db; let home: string; @@ -45,7 +50,10 @@ describe.skipIf(!support.supported)("heartbeat workspace preparation boundary", beforeAll(async () => { home = await mkdtemp(join(tmpdir(), "paperclip-workspace-preparation-")); vi.stubEnv("PAPERCLIP_HOME", home); - database = await startEmbeddedPostgresTestDatabase("paperclip-workspace-preparation"); + const external = externalDatabaseUrl; + if (external) await applyPendingMigrations(external); + database = external ? { connectionString: external, cleanup: async () => {} } + : await startEmbeddedPostgresTestDatabase("paperclip-workspace-preparation"); db = createDb(database.connectionString); await db.execute(sql`set client_min_messages = warning`); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); @@ -320,6 +328,119 @@ describe.skipIf(!support.supported)("heartbeat workspace preparation boundary", expect((await readIssue(issue.id)).executionWorkspaceId).toBe(prepared.persistedExecutionWorkspace!.id); }); + it.each(["reuse", "explicit_git", "foreign_path", "missing_path", "symlink_path", "binding_changed", "canonical_retry", "canonical_warning", "canonical_unknown_warning", "canonical_provider_warning", "canonical_fresh", "canonical_lease", "canonical_result", "canonical_controller", "canonical_output", "canonical_measurement", "canonical_repaired_git"] as const)("recomputes ordinary-turn policy for a saved source-free fresh directory: %s", async scenario => { + const { input, issue } = await fixture(); + const userId = `directory-owner-${randomUUID()}`; + await db.insert(authUsers).values({ id: userId, name: "Directory owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId: input.agent.companyId, principalId: userId, principalType: "user", membershipRole: "owner", status: "active" }); + await db.update(companies).set({ defaultResponsibleUserId: userId }).where(eq(companies.id, input.agent.companyId)); + await db.update(issues).set({ projectId: null, projectWorkspaceId: null, status: "in_progress", responsibleUserId: userId }).where(eq(issues.id, issue.id)); + input.task.issueRef = await readIssue(issue.id); + input.task.context = { issueId: issue.id }; + input.policy.useIsolatedTaskDirectory = true; + input.policy.effectiveExecutionWorkspaceMode = "isolated_workspace"; + input.policy.requestedExecutionWorkspaceMode = "isolated_workspace"; + input.config.mergedConfig = { workspaceStrategy: { type: "git_worktree" } }; + input.task.freshWorkspace = { mode: "fresh_isolated_workspace", retryIntentId: input.run.id, + sourceRunId: input.run.id, retainedSourceIds: [] }; + const original = await prepareHeartbeatWorkspace(db, input); + const workspace = original.persistedExecutionWorkspace!; + await writeFile(join(workspace.cwd!, "saved-review.txt"), "completed snapshot review\n"); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, input.run.id)); + await db.update(environmentLeases).set({ status: "expired", cleanupStatus: "success", releasedAt: new Date() }).where(eq(environmentLeases.heartbeatRunId, input.run.id)); + const adapterType = "process"; + const originalAdapter = getServerAdapter(adapterType); + let preservedCwd = workspace.cwd!; + if (scenario === "explicit_git" || scenario.startsWith("canonical_")) await db.update(issues).set({ executionWorkspaceSettings: { mode: "isolated_workspace", workspaceStrategy: { type: "git_worktree" } } }).where(eq(issues.id, issue.id)); + if (scenario === "foreign_path") { + const outside = join(home, "unowned-directory"); await mkdir(outside); + await db.update(executionWorkspaces).set({ cwd: outside }).where(eq(executionWorkspaces.id, workspace.id)); + } + if (scenario === "missing_path" || scenario === "symlink_path") { + preservedCwd = join(home, `retained-files-${issue.id}`); await rename(workspace.cwd!, preservedCwd); + if (scenario === "symlink_path") await symlink(preservedCwd, workspace.cwd!); + } + if (scenario === "binding_changed") await db.update(issues).set({ executionWorkspaceId: null }).where(eq(issues.id, issue.id)); + const freshExpected = ["canonical_unknown_warning", "canonical_provider_warning", "canonical_fresh", "canonical_lease", "canonical_result", "canonical_controller", "canonical_output", "canonical_measurement"].includes(scenario); + const executed = vi.fn(async (execution: AdapterExecutionContext) => { + const [lease] = await leases(execution.runId); + if (freshExpected) { + expect(lease.executionWorkspaceId).not.toBe(workspace.id); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, issue.id)); + return { exitCode: 0, signal: null, timedOut: false, summary: "Fresh workspace required" }; + } + if (scenario !== "canonical_repaired_git") expect(lease.executionWorkspaceId).toBe(workspace.id); + const activeWorkspace = await input.services.executionWorkspacesSvc.getStoredById(input.agent.companyId, lease.executionWorkspaceId!); + if (scenario === "canonical_repaired_git") { + expect(execution.config.workspaceStrategy).toMatchObject({ type: "git_worktree" }); + expect((await execFile("git", ["-C", activeWorkspace!.cwd!, "rev-parse", "--is-inside-work-tree"])).stdout.trim()).toBe("true"); + } + else expect(activeWorkspace?.cwd).toBe(workspace.cwd); + expect(await readFile(join(activeWorkspace!.cwd!, "saved-review.txt"), "utf8")).toBe("completed snapshot review\n"); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, issue.id)); + return { exitCode: 0, signal: null, timedOut: false, summary: "Retained review confirmed" }; + }); + registerServerAdapter({ ...originalAdapter, type: adapterType, supportsLocalAgentJwt: false, execute: executed, + testEnvironment: async () => ({ adapterType, status: "pass", checks: [], testedAt: new Date().toISOString() }) }); + await db.update(agents).set({ adapterType, status: "idle", adapterConfig: { workspaceStrategy: { type: "git_worktree" } } }).where(eq(agents.id, input.agent.id)); + if (scenario.includes("warning")) await db.insert(companyObservability).values({ companyId: input.agent.companyId, revision: randomUUID(), enabled: true, + destination: { provider: "otlp", endpointBaseUrl: "https://collector.example/otel" }, + capture: { userPrompts: false, assistantResponses: false, toolDetails: false, toolContent: false } }); + const heartbeat = heartbeatService(db); + try { + const next = await heartbeat.wakeup(input.agent.id, { source: "on_demand", triggerDetail: "manual", reason: "manual", + requestedByActorType: "user", requestedByActorId: userId, payload: { issueId: issue.id }, contextSnapshot: { issueId: issue.id } }); + await heartbeat.drainActiveRunExecutions(); + let finalRunId = next?.id; + if (scenario.startsWith("canonical_")) { + expect(await readRun(next!.id)).toMatchObject({ status: "failed", errorCode: "workspace_validation_failed" }); + expect(executed).not.toHaveBeenCalled(); + if (scenario.includes("warning")) { + const warning = (await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, next!.id))).find(event => event.seq === 1)!; + expect(warning).toMatchObject({ eventType: "agent_observability.unavailable", payload: { reason: "unsupported_harness" } }); + if (scenario === "canonical_unknown_warning") await db.update(heartbeatRunEvents).set({ payload: { ...warning.payload, reason: "unknown" } }).where(eq(heartbeatRunEvents.id, warning.id)); + if (scenario === "canonical_provider_warning") await db.update(heartbeatRunEvents).set({ sourcePayloadSha256: "a".repeat(64), protocolSchemaVersion: 1 }).where(eq(heartbeatRunEvents.id, warning.id)); + } + if (scenario === "canonical_lease") await db.insert(environmentLeases).values({ companyId: input.agent.companyId, issueId: issue.id, heartbeatRunId: next!.id, provider: "local_fs", status: "expired", cleanupStatus: "success" }); + if (scenario === "canonical_result") { const failed = await readRun(next!.id); await db.update(heartbeatRuns).set({ resultJson: { ...failed.resultJson, acceptedResult: { outcome: "unknown" } } }).where(eq(heartbeatRuns.id, next!.id)); } + if (scenario === "canonical_controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(heartbeatRuns.id, next!.id)); + if (scenario === "canonical_output") await db.update(heartbeatRuns).set({ lastOutputSeq: 1 }).where(eq(heartbeatRuns.id, next!.id)); + if (scenario === "canonical_measurement") { const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, next!.id)); const measured = events.find(event => event.eventType === "lifecycle")!; await db.update(heartbeatRunEvents).set({ payload: { ...measured.payload, durationMs: -1 } }).where(eq(heartbeatRunEvents.id, measured.id)); } + if (scenario === "canonical_repaired_git") { + await execFile("git", ["init", "--initial-branch=main", workspace.cwd!]); + await execFile("git", ["-C", workspace.cwd!, "config", "user.email", "test@example.com"]); + await execFile("git", ["-C", workspace.cwd!, "config", "user.name", "Directory repair"]); + await execFile("git", ["-C", workspace.cwd!, "add", "."]); + await execFile("git", ["-C", workspace.cwd!, "commit", "-m", "Retain saved review"]); + } else await db.update(issues).set({ executionWorkspaceSettings: { mode: "isolated_workspace" } }).where(eq(issues.id, issue.id)); + const intent = await requestTaskRetry({ db, companyId: input.agent.companyId, issueId: issue.id, + actorId: userId, actorSource: "local_implicit", observedRunId: next!.id, idempotencyKey: randomUUID() }); + expect(intent.requestId).toBeTruthy(); + if (scenario === "canonical_fresh") await db.update(agentWakeupRequests).set({ payload: sql`jsonb_set(${agentWakeupRequests.payload}, '{taskRetry,strategy}', '"fresh_isolated_workspace"'::jsonb)` }).where(eq(agentWakeupRequests.id, intent.requestId!)); + await heartbeat.resumeTaskRetry(input.agent.companyId, intent.requestId!); + await heartbeat.drainActiveRunExecutions(); + const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, intent.requestId!)); + expect(receipt.runId).toBeTruthy(); + finalRunId = receipt.runId!; + } + const continued = scenario === "reuse" || scenario.startsWith("canonical_"); + if (scenario === "binding_changed") { + expect(next).toBeNull(); + const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, input.agent.id)); + expect(receipt).toMatchObject({ status: "skipped", reason: "workspace_worktree_requires_project", runId: null }); + } else { + expect(next).not.toBeNull(); + expect(await readRun(finalRunId!)).toMatchObject(continued + ? { status: "succeeded", errorCode: null } : { status: "failed", errorCode: "workspace_validation_failed" }); + } + expect(executed).toHaveBeenCalledTimes(continued ? 1 : 0); + if (freshExpected) expect((await readIssue(issue.id)).executionWorkspaceId).not.toBe(workspace.id); + else if (scenario !== "canonical_repaired_git") expect((await readIssue(issue.id)).executionWorkspaceId).toBe(scenario === "binding_changed" ? null : workspace.id); + expect(await readFile(join(preservedCwd, "saved-review.txt"), "utf8")).toBe("completed snapshot review\n"); + if (scenario === "missing_path") await expect(access(workspace.cwd!)).rejects.toThrow(); + } finally { await heartbeat.drainActiveRunExecutions(); registerServerAdapter(originalAdapter); } + }, 30_000); + it("reuses a successful fresh task attempt for the next ordinary user turn", async () => { const { input, issue } = await fixture(); input.policy.useIsolatedTaskDirectory = true; diff --git a/server/src/services/isolated-task-directory.ts b/server/src/services/isolated-task-directory.ts index 26238976e08..65d0748b85e 100644 --- a/server/src/services/isolated-task-directory.ts +++ b/server/src/services/isolated-task-directory.ts @@ -1,5 +1,6 @@ import { lstat, mkdir, realpath } from "node:fs/promises"; import path from "node:path"; +import type { ExecutionWorkspace } from "@paperclipai/shared"; import { parseObject } from "../adapters/utils.js"; import { resolvePaperclipInstanceRoot } from "../home-paths.js"; @@ -63,3 +64,20 @@ export async function materializeIsolatedTaskDirectory(input: { } return cwd; } + +export async function isBoundIsolatedTaskDirectory(input: { + companyId: string; issueId: string; workspaceId: string | null; + workspace: Pick | null; +}): Promise { + const workspace = input.workspace; + if (!workspace || workspace.id !== input.workspaceId || workspace.companyId !== input.companyId || + workspace.sourceIssueId !== input.issueId || workspace.status !== "active" || + workspace.mode !== "isolated_workspace" || workspace.strategyType !== "project_primary" || + workspace.providerType !== "local_fs" || !workspace.cwd || workspace.repoUrl || workspace.branchName) return false; + try { + await materializeIsolatedTaskDirectory({ companyId: input.companyId, issueId: input.issueId, reuseCwd: workspace.cwd }); + return true; + } catch { + return false; + } +} diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index b5ec6de2a25..91b5e3e5b6f 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -775,8 +775,19 @@ describe("automatic retained workspace recovery", () => { expect(await readFile(path.join(f.host, "newer.txt"), "utf8")).toBe("newer local work\n"); }); - it.each(["native", "legacy", "native_export", "legacy_snapshot"])("delivers only saved user input once through ordinary admission: %s", async mode => { - const f = await seed(mode.startsWith("legacy") ? "legacy" : "native", mode !== "legacy_snapshot"), commentId = randomUUID(), queueId = randomUUID(); + it.each(["native", "legacy", "native_export", "legacy_snapshot", "native_snapshot"])("delivers only saved user input once through ordinary admission: %s", async mode => { + const f = await seed(mode.startsWith("legacy") ? "legacy" : "native", !mode.endsWith("_snapshot")), commentId = randomUUID(), queueId = randomUUID(); + if (mode === "native_snapshot") { + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { + schema: "paperclip.native-workspace-sync/v3", state: "prepared", descriptorSha256: "a".repeat(64), + baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId: f.issueId, + leaseId: f.lease.id, providerLeaseId: f.providerLeaseId, remoteCwd: f.remote, resourceDisposition: "destroy", + } } }).where(eq(heartbeatRuns.id, f.runId)); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", status: "resolved", ownerType: "system", outcome: "blocked", + cause: "execution_finalization_deadline_exceeded", fingerprint: randomUUID(), nextAction: "Inspect original failure", + evidence: { runId: f.runId, automaticRecovery: { replay: "blocked" } } }); + } if (mode === "native_export") { await db.update(issueRecoveryActions).set({ cause: "native_workspace_sync_out_retry_exhausted", evidence: { runId: f.runId } }).where(eq(issueRecoveryActions.id, f.actionId)); await db.update(heartbeatRuns).set({ resultJson: {} }).where(eq(heartbeatRuns.id, f.runId)); @@ -798,8 +809,19 @@ describe("automatic retained workspace recovery", () => { expect(turns[0].contextSnapshot).toMatchObject({ forceFreshSession: true, previousRunId: f.runId, wakeCommentIds: [commentId] }); expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId)))[0]).toMatchObject({ status: "coalesced", runId: turns[0].id }); expect((await run(f.runId)).status).toBe("failed"); - if (mode === "legacy_snapshot") expect(await readFreshWorkspaceReadiness(db, turns[0])).toMatchObject({ - mode: "fresh_isolated_workspace", retryIntentId: queueId, sourceRunId: f.runId, retainedSourceIds: [f.lease.id] }); + if (mode.endsWith("_snapshot")) { + expect(await readFreshWorkspaceReadiness(db, turns[0])).toMatchObject({ + mode: "fresh_isolated_workspace", retryIntentId: queueId, sourceRunId: f.runId, retainedSourceIds: [f.lease.id] }); + const fresh = path.join(root, randomUUID()); await mkdir(fresh); + const attached = await attachPreservedWorkspaceSnapshots({ db, companyId: f.companyId, issueId: f.issueId, + sourceRunIds: [turns[0].id], cwd: fresh }); + expect(attached).toHaveLength(1); + expect(await readFile(path.join(attached[0].path, "files", "saved.txt"), "utf8")).toBe("finished work\n"); + expect(await readFile(path.join(f.host, "newer.txt"), "utf8")).toBe("newer local work\n"); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); + if (mode === "native_snapshot") expect((await run(f.runId)).runnerProfileJson?.nativeWorkspaceSync) + .toMatchObject({ state: "prepared", finalHostSha256: null }); + } }, 30_000); it.each(["copyback_unknown", "source_unavailable", "executor_unknown"])("ordinary task Retry evaluates fresh isolation independently: %s", async failure => {