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 => {