diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index a51bbcdd42f..fc3720f0edc 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2897,6 +2897,13 @@ Paperclip submits saved input through normal task admission, once, with the original user's authority. Pauses, task ownership, budgets, approvals, and execution recovery holds still apply. Unconfirmed cleanup does not start work. +An admitted connected-task Retry keeps its existing successor queued while the +exact native predecessor still owes cleanup, even when the task already points +to the successor. A stop-and-retain receipt alone does not settle that cleanup. +Cleanup writes and exact workspace-owner releases wake scoped dispatch after +commit; the same run then passes the normal source, identity, Stop, approval and +budget checks again. + A Stop from a bound Slack session records the verified user's identity, just as a board Stop does. Native cancellation can then recognize the acknowledged operator Stop instead of creating an execution recovery hold. An AI login request diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index e4e6fc9e798..4abc378949d 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -65177,6 +65177,161 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } finally { await f.service.shutdown(); } }, ); + it.each(["settled", "pending_at_intake", "source_edited", "operator_stopped", "unknown_stop", "live_owner", "absent_sandbox"] as const)( + "keeps the admitted chat Retry queued until its source lease cleanup settles: %s", async mode => { + const f = await currentGitHubTaskRetryFixture(); + const companyId = f.fixture.companyId, issueId = f.issue.id; + const heartbeat = heartbeatService(db); + const errors: unknown[] = []; + let coordinator: ReturnType | undefined; + let releaseWorkspaceOwner: () => void = () => undefined; + let workspaceOwner: Promise | undefined; + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { req.actor = boardActor(companyId); next(); }); + app.use("/api", issueRoutes(db, createStorageService().storage, { chatRunRetries: f.service })); + app.use(errorHandler); + const lease = { id: randomUUID(), companyId, issueId, heartbeatRunId: f.runId, + provider: "daytona", providerLeaseId: randomUUID(), status: "released", cleanupStatus: "success", + releasedAt: new Date(), leasePolicy: "retain_on_failure" }; + const termination = remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" })!; + const metadata = { remoteExecutionTermination: termination, sandboxStopAndRetainReceipt: { + schema: "paperclip.sandbox-stop-and-retain-receipt.v1", companyId, runId: f.runId, + leaseId: lease.id, provider: lease.provider, providerLeaseId: lease.providerLeaseId, + requestId: randomUUID(), confirmedAt: termination.confirmedAt, builtinProvider: "daytona", method: "builtin.stopLease", + } }; + await db.insert(environmentLeases).values({ ...lease, metadata, + ...(mode === "pending_at_intake" ? { status: "pending_cleanup", cleanupStatus: "failed" } : {}) }); + await db.update(nativeRunFinalizations).set({ failureCode: "native_session_interrupted", + failureDetail: { recoveryMode: "ambiguous_state", providerSessionEstablished: false, + providerEventsExist: false, checkpointExists: false } }).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", status: "resolved", outcome: "blocked", ownerType: "system", + cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(), nextAction: "Verify source cleanup", + evidence: { runId: f.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" }, + systemRecovery: { ...createSystemRecovery(f.runId), state: mode === "pending_at_intake" ? "running" : "complete" } } }); + const adapterType = `retained-chat-retry-${randomUUID()}`; + const execute = vi.fn(async () => ({ exitCode: 0, signal: null, timedOut: false, errorMessage: null, + summary: "Completed the exact saved request after source cleanup", provider: "test", model: "test-model" })); + registerServerAdapter({ type: adapterType, supportsLocalAgentJwt: false, execute, + testEnvironment: async () => ({ adapterType, status: "pass", checks: [], testedAt: new Date().toISOString() }) }); + try { + const body = { observedRunId: f.runId, idempotencyKey: randomUUID() }; + const response = await request(app).post(`/api/issues/${issueId}/retry`).send(body); + expect(response.status, JSON.stringify(response.body)).toBe(202); + if (mode === "pending_at_intake") { + await expect.poll(async () => (await readAgentWakeupRequestsById(response.body.chatActionId))[0]?.status) + .toBe("deferred_issue_execution"); + expect((await readAgentWakeupRequestsById(response.body.requestId))[0]).toMatchObject({ runId: null }); + expect(execute).not.toHaveBeenCalled(); + await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success" }) + .where(eq(environmentLeases.id, lease.id)); + await reconcileTaskChatRetryReceipts(db, { companyId }); + await f.service.processFailedChatRunRetry(response.body.chatActionId); + } + await expect.poll(async () => (await readAgentWakeupRequestsById(response.body.requestId))[0]?.runId).toBeTruthy(); + const [intent] = await readAgentWakeupRequestsById(response.body.requestId); + const [successor] = await readHeartbeatRunsById(intent.runId!); + expect(successor).toMatchObject({ status: "queued", retryOfRunId: f.runId, + contextSnapshot: { taskRetryRequestId: intent.id, explicitUserContinuation: { + previousRunId: f.runId, readiness: { mode: "fresh_isolated_workspace" } } } }); + await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", + metadata: { ...metadata, pendingCleanupInFlight: true } }).where(eq(environmentLeases.id, lease.id)); + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "running"), isNull(heartbeatRuns.wakeupRequestId))); + await db.update(agents).set({ status: "idle" }).where(eq(agents.id, f.fixture.assignedAgentId)); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + expect((await readHeartbeatRunsById(successor.id))[0]).toMatchObject({ status: "queued", startedAt: null }); + expect(execute).not.toHaveBeenCalled(); + expect((await readAgentWakeupRequestsById(response.body.requestId))[0]).toMatchObject({ runId: successor.id }); + if (mode === "operator_stopped") await heartbeat.cancelRun(successor.id); + if (mode === "source_edited") await db.update(issueComments).set({ body: "Changed after admission" }) + .where(eq(issueComments.id, String(f.action.payload.commentId))); + if (mode === "live_owner") await db.update(nativeRunFinalizations).set({ leaseOwner: "another-controller", + leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(agents).set({ adapterType }).where(eq(agents.id, f.fixture.assignedAgentId)); + const { registerRunDispatchWork } = await import("../services/heartbeat/dispatch-work.js"); + const { writeEnvironmentCleanup } = await import("../services/environment-cleanup-notifications.js"); + let dispatchAttempts = 0; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + onError: error => errors.push(error) }); + await registerRunDispatchWork(coordinator, { promoteDueScheduledRetries: async () => undefined, + nextScheduledRetryAt: async () => null, schedulingSuppressed: async () => false, + resumeQueuedRunsAfterDrain: async () => { await heartbeat.resumeQueuedRunsAfterDrain({ companyId }); dispatchAttempts++; }, + }, () => true).ready; + await expect.poll(() => dispatchAttempts).toBeGreaterThan(0); + await expect.poll(() => coordinator!.nextWakeAt(), { timeout: 10_000 }).toBeNull(); + const beforeSettlement = dispatchAttempts; + const settledMetadata = { ...metadata, pendingCleanupInFlight: false, + remoteExecutionTermination: mode === "unknown_stop" ? { ...termination, providerLeaseId: randomUUID() } : termination }; + if (mode === "absent_sandbox") { + const { withNativeWorkspaceFinalizationOwnership } = await import("../services/native-runtime/native-workspace-finalization-ownership.js"); + const { settleAbsentSandbox } = await import("../services/sandbox-stop-and-retain.js"); + let entered!: (token: string) => void; + const ownerEntered = new Promise(resolve => { entered = resolve; }); + const ownerMayFinish = new Promise(resolve => { releaseWorkspaceOwner = resolve; }); + workspaceOwner = withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: f.runId }, async ownership => { + entered(ownership.token); + await ownerMayFinish; + }); + const ownerToken = await ownerEntered; + const [currentLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)); + expect(await settleAbsentSandbox(db, currentLease, { receipt: "verified-upstream", proof: { + companyId, issueId, runId: f.runId, leaseId: lease.id, provider: "daytona", providerLeaseId: lease.providerLeaseId, + ownerToken, v: 1, iss: "paperclip-cloud", aud: "paperclip-sandbox-absence/v1", sub: "fixture", absent: true, + iat: Math.floor(Date.now() / 1000), exp: Math.floor(Date.now() / 1000) + 60, + } })).toBe(true); + expect((await readHeartbeatRunsById(successor.id))[0]).toMatchObject({ status: "queued", startedAt: null }); + expect(execute).not.toHaveBeenCalled(); + releaseWorkspaceOwner(); + await workspaceOwner; + } else { + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "released", cleanupStatus: "success", + metadata: settledMetadata }).where(eq(environmentLeases.id, lease.id)), { companyId, runId: f.runId, leaseId: lease.id }); + } + await expect.poll(() => dispatchAttempts, { timeout: 10_000 }).toBeGreaterThan(beforeSettlement); + if (mode === "source_edited") await expect.poll(async () => + (await readHeartbeatRunsById(successor.id))[0]?.status, { timeout: 10_000 }).toBe("failed"); + await heartbeat.drainActiveRunExecutions(); + expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(1); + expect((await readHeartbeatRunsById(f.runId))[0]).toMatchObject({ status: "failed" }); + expect((await request(app).post(`/api/issues/${issueId}/retry`).send(body)).body.requestId).toBe(intent.id); + expect(errors).toEqual([]); + if (mode === "operator_stopped") { + expect(execute).not.toHaveBeenCalled(); + expect((await readHeartbeatRunsById(successor.id))[0]).toMatchObject({ status: "cancelled" }); + return; + } + if (mode === "source_edited" || mode === "unknown_stop") { + expect(execute).not.toHaveBeenCalled(); + const [withheld] = await readHeartbeatRunsById(successor.id); + if (mode === "unknown_stop") expect(["queued", "failed"]).toContain(withheld.status); + if (withheld.status === "failed") expect(withheld.errorCode).toBe("chat_failed_run_retry_not_authorized"); + return; + } + if (mode === "live_owner") { + await expect.poll(() => coordinator!.nextWakeAt(), { timeout: 10_000 }).toBeNull(); + expect(execute).not.toHaveBeenCalled(); + expect((await readHeartbeatRunsById(successor.id))[0]).toMatchObject({ status: "queued", startedAt: null }); + const { writeExecutionControl } = await import("../services/execution-control-notifications.js"); + const { DELIVERY_QUEUES } = await import("../services/delivery-work-notifications.js"); + await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(nativeRunFinalizations) + .set({ leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId)), { companyId, runId: f.runId }); + } + await expect.poll(() => execute.mock.calls.length, { timeout: 10_000 }).toBe(1); + await heartbeat.drainActiveRunExecutions(); + expect(execute).toHaveBeenCalledTimes(1); + expect((await readHeartbeatRunsById(successor.id))[0]).toMatchObject({ status: "succeeded" }); + expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(1); + expect((await readHeartbeatRunsById(f.runId))[0]).toMatchObject({ status: "failed" }); + expect((await request(app).post(`/api/issues/${issueId}/retry`).send(body)).body.requestId).toBe(intent.id); + expect(execute).toHaveBeenCalledTimes(1); + await expect.poll(() => coordinator!.nextWakeAt(), { timeout: 10_000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { releaseWorkspaceOwner(); await workspaceOwner?.catch(() => undefined); await coordinator?.stop(); + await heartbeat.drainActiveRunExecutions(); unregisterServerAdapter(adapterType); await f.service.shutdown(); } + }); it("retries saved GitHub input after a successful run without replaying that run", async () => { const f = await successfulGitHubQueuedInputFixture(); diff --git a/server/src/services/chat-channels/task-retry-authority.ts b/server/src/services/chat-channels/task-retry-authority.ts index 8bb532d9c0e..d17bcc4fc04 100644 --- a/server/src/services/chat-channels/task-retry-authority.ts +++ b/server/src/services/chat-channels/task-retry-authority.ts @@ -5,7 +5,7 @@ import { readTaskRetryIntent } from "../task-retry.js"; import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, recordedLocalExecutionStopped } from "../native-local-process-stop.js"; import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js"; import { adapterExecutionControls } from "../adapter-execution-control.js"; -import { getExecutionBlocker } from "../execution-blocker.js"; +import { getExecutionBlocker, hasPendingRunCleanup } from "../execution-blocker.js"; import { readSystemRecovery } from "../system-recovery-state.js"; /** This receipt changes the workspace strategy, never the external principal, @@ -34,6 +34,23 @@ export async function taskChatRetryExecutionStopped(db: Db, run: typeof heartbea return taskChatRetryExecutionContained(db, run, false); } +export async function taskChatRetrySourceCleanupPending(db: Db, run: typeof heartbeatRuns.$inferSelect, issueId: string) { + const hint = run.contextSnapshot?.chatFailedRunRetry as Record | undefined; + const intentId = run.contextSnapshot?.taskRetryRequestId; + if (typeof intentId !== "string" || !run.retryOfRunId || !run.wakeupRequestId || + run.contextSnapshot?.issueId !== issueId || hint?.version !== 1 || + hint.actionId !== run.wakeupRequestId || hint.failedRunId !== run.retryOfRunId) return false; + const retry = await readTaskChatRetryIntent(db, { companyId: run.companyId, issueId, agentId: run.agentId, + sourceRunId: run.retryOfRunId, intentId, chatActionId: run.wakeupRequestId }); + if (retry?.wake.status !== "coalesced" || retry.wake.runId !== run.id) return false; + const [source] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId), + eq(heartbeatRuns.nativeIssueId, issueId), eq(heartbeatRuns.id, run.retryOfRunId), + eq(heartbeatRuns.runtimeMode, "native"), + )); + return Boolean(source && await hasPendingRunCleanup(db, run.companyId, source.id)); +} + async function taskChatRetryExecutionContained(db: Db, run: typeof heartbeatRuns.$inferSelect, admittedFailure: boolean) { if (adapterExecutionControls.has(run.id) || (run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date())) return false; diff --git a/server/src/services/environment-cleanup-notifications.ts b/server/src/services/environment-cleanup-notifications.ts index 76ba2521ea1..958ec68666b 100644 --- a/server/src/services/environment-cleanup-notifications.ts +++ b/server/src/services/environment-cleanup-notifications.ts @@ -4,6 +4,7 @@ import { notifyLiveResourceRecovery } from "./live-run-recovery-notifications.js import { notifyNativeEnvironmentSettlement, type NativeEnvironmentSettlementScope } from "./native-environment-settlement-notifications.js"; import type { Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import { notifyQueuedRunDispatch } from "./run-dispatch-notifications.js"; export type EnvironmentCleanupWorkScope = NativeEnvironmentSettlementScope & { leaseId?: string }; @@ -28,6 +29,7 @@ export async function notifyEnvironmentCleanup(transaction: object, scope?: Envi } } await notifyNativeEnvironmentSettlement(transaction, scope); + await notifyQueuedRunDispatch(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.environmentCleanup); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup); } diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 78812e8c239..aa0977e5659 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -51,6 +51,7 @@ import { import { canRetryStoppedRun } from "../cancelled-native-startup.js"; import { getExecutionBlocker, hasPendingRunCleanup } from "../execution-blocker.js"; import { nativeManualAdmissionHoldsExecution } from "../native-runtime/native-runner-ownership.js"; +import { taskChatRetrySourceCleanupPending } from "../chat-channels/task-retry-authority.js"; import { validateNativeWorkspaceReconstruction } from "../native-runtime/native-workspace-reconstruction.js"; import { isConversationAdapter } from "../conversation-continuation.js"; import { recordExecutionWait } from "../execution-wait.js"; @@ -1075,6 +1076,9 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende owner.executionRunId !== run.id && owner.executionRunId !== run.retryOfRunId) { return { ownsIssue, blocked: true }; } + if (ownsIssue && issueId && await taskChatRetrySourceCleanupPending(tx, run, issueId)) { + return { ownsIssue, blocked: true }; + } const previousRunId = run.scheduledRetryReason === "native_provider_overloaded" ? run.retryOfRunId : ownsIssue ? owner?.executionRunId : null; if (previousRunId && previousRunId !== run.id) { diff --git a/server/src/services/native-runtime/native-workspace-finalization-ownership.ts b/server/src/services/native-runtime/native-workspace-finalization-ownership.ts index c908a7d004b..62415c69f3e 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-ownership.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-ownership.ts @@ -1,6 +1,7 @@ import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization, releaseNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; +import { notifyQueuedRunDispatch } from "../run-dispatch-notifications.js"; import { beginRunPhase } from "../execution-preparation-timing.js"; import { logActivity } from "../activity-log.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; @@ -302,6 +303,7 @@ export async function withNativeWorkspaceFinalizationOwnership( await notifyNativeResultCommit(tx, { companyId: input.companyId, runId: input.runId }); await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, runId: input.runId }); await notifyNativeStatusReassessment(tx, { companyId: input.companyId, runId: input.runId }); + await notifyQueuedRunDispatch(tx, { companyId: input.companyId }); await tx.update(heartbeatRuns).set({ runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) - ${OWNER_KEY}`, }).where(owned); diff --git a/server/src/services/native-runtime/native-workspace-finalizer.test.ts b/server/src/services/native-runtime/native-workspace-finalizer.test.ts index 53063f345e9..0d57c171400 100644 --- a/server/src/services/native-runtime/native-workspace-finalizer.test.ts +++ b/server/src/services/native-runtime/native-workspace-finalizer.test.ts @@ -219,12 +219,19 @@ describe("native workspace finalization recovery", () => { observed.push(db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns) .where(eq(heartbeatRuns.id, seed.runId)).then(rows => rows[0].profile)); }); + const queued: Array> = []; + const stopQueued = subscribeDeliveryWork(db, DELIVERY_QUEUES.queuedRunDispatch, () => { + queued.push(db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns) + .where(eq(heartbeatRuns.id, seed.runId)).then(rows => rows[0].profile)); + }); let receipt: Awaited>; try { receipt = await resumeNativeWorkspaceAfterOwnerStop(seed); await vi.waitFor(() => expect(observed).toHaveLength(1)); expect(await Promise.all(observed)).toEqual([{ preserved: "workspace descriptor" }]); - } finally { stop(); } + await vi.waitFor(() => expect(queued).toHaveLength(1)); + expect(await Promise.all(queued)).toEqual([{ preserved: "workspace descriptor" }]); + } finally { stop(); stopQueued(); } expect(receipt).toEqual({ runId: seed.runId, resultId: before.resultId, status: "queued" }); const [after] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seed.runId)); expect(after).toEqual(before); diff --git a/server/src/services/native-runtime/native-workspace-owner-recovery.ts b/server/src/services/native-runtime/native-workspace-owner-recovery.ts index 60a92a72b89..346ee79e885 100644 --- a/server/src/services/native-runtime/native-workspace-owner-recovery.ts +++ b/server/src/services/native-runtime/native-workspace-owner-recovery.ts @@ -2,6 +2,7 @@ import { notifyNativeResultCommit } from "../native-result-commit-notifications. import { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js"; import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; +import { notifyQueuedRunDispatch } from "../run-dispatch-notifications.js"; import os from "node:os"; import { and, eq, sql } from "drizzle-orm"; import { heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db"; @@ -71,6 +72,7 @@ export async function resumeNativeWorkspaceAfterOwnerStop(input: { await notifyNativeResultCommit(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId }); await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId }); await notifyNativeStatusReassessment(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId }); + await notifyQueuedRunDispatch(tx, { companyId: input.companyId }); await tx.update(heartbeatRuns).set({ runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} - ${OWNER_KEY}`, updatedAt: now, }).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, run.id), diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts index 1b013f4f304..3cd4f00cbe1 100644 --- a/server/src/services/run-dispatch-notifications.ts +++ b/server/src/services/run-dispatch-notifications.ts @@ -3,11 +3,16 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type RunDispatchScope = { companyId: string }; +export async function notifyQueuedRunDispatch(transaction: object, scope?: RunDispatchScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, + scope ? JSON.stringify({ companyId: scope.companyId }) : "*"); +} + export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) { await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined); const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativeProcessSessions, hint); - await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint); + await notifyQueuedRunDispatch(transaction, scope); } /** Only a retry row or deadline change needs to re-read the saved retry schedule. */ diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 68e0e7bdad5..f1b3fb758f9 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -361,14 +361,48 @@ describe("automatic retained workspace recovery", () => { return app; } - async function acceptedRemoteExport() { + type RecoverySourceLayout = "retained" | "published" | "raw"; + + async function acceptedRemoteExport(sourceLayout: RecoverySourceLayout = "retained") { const f = await seed("native", false); - const source = await run(f.runId), workspaceId = randomUUID(); - const publishedRoot = path.join(path.dirname(f.remote), ".paperclip-workspaces", workspaceId); - await mkdir(path.dirname(publishedRoot), { recursive: true }); - await rename(f.remote, publishedRoot); - f.remote = publishedRoot; - const remoteCwd = await realpath(f.remote); + const source = await run(f.runId); + let workspaceId: string = randomUUID(), remoteCwd = await realpath(f.remote); + let preparedReference: Awaited>["reference"] | undefined; + if (sourceLayout === "published") { + // Fresh modern sources retain the actual publisher's path and authority. + // The explicit raw fixture remains available for refusal coverage. + targets.get(f.lease.id)!.remoteCwd = remoteCwd; + const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + await db.update(issues).set({ status: "in_progress", executionRunId: f.runId, executionLockedAt: new Date() }) + .where(eq(issues.id, f.issueId)); + await db.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null }) + .where(eq(heartbeatRuns.id, f.runId)); + const leaseMetadata = { ...f.lease.metadata }; + delete leaseMetadata.remoteExecutionTermination; + const [activeLease] = await db.update(environmentLeases).set({ status: "active", releasedAt: null, + cleanupStatus: null, metadata: leaseMetadata }).where(eq(environmentLeases.id, f.lease.id)).returning(); + const reserved = await reserveRemoteWorkspace({ db, companyId: f.companyId, issueId: f.issueId, + runId: f.runId, agentId: source.agentId, environmentId: activeLease.environmentId!, + projectId: null, projectWorkspaceId: null, existingWorkspaceId: null, source: { kind: "empty" } }); + const prepared = await prepareRemoteWorkspace({ db, workspace: reserved, runId: f.runId, + lease: activeLease as EnvironmentLease, target: targets.get(f.lease.id)! }); + workspaceId = prepared.workspace.id; remoteCwd = prepared.remoteRoot; preparedReference = prepared.reference; + for (const name of ["existing.txt", "saved.txt"]) await cp(path.join(f.remote, name), path.join(remoteCwd, name)); + await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date(), + metadata: { ...prepared.lease.metadata, + remoteExecutionTermination: remoteTerminationReceipt(prepared.lease, { providerLeaseId: f.providerLeaseId, state: "stopped" }) }, + }).where(eq(environmentLeases.id, f.lease.id)); + await db.update(heartbeatRuns).set({ status: source.status, finishedAt: source.finishedAt, + error: source.error, errorCode: source.errorCode }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(issues).set({ status: issue.status, executionRunId: issue.executionRunId, + executionLockedAt: issue.executionLockedAt }).where(eq(issues.id, f.issueId)); + } else if (sourceLayout === "retained") { + const publishedRoot = path.join(path.dirname(f.remote), ".paperclip-workspaces", workspaceId); + await mkdir(path.dirname(publishedRoot), { recursive: true }); + await rename(f.remote, publishedRoot); + f.remote = publishedRoot; + remoteCwd = await realpath(f.remote); + } const git = (args: string[], cwd = remoteCwd) => new Promise((resolve, reject) => execFile("git", args, { cwd }, (error, stdout) => error ? reject(error) : resolve(stdout.trim()))); await git(["init"]); @@ -380,7 +414,7 @@ describe("automatic retained workspace recovery", () => { const authority = { schema: "paperclip.remote-workspace.v1", epoch: 1, snapshotId: null, source: { kind: "empty" }, writerRunId: f.runId, dirty: true, location: { environmentId: f.lease.environmentId!, leaseId: f.lease.id, providerLeaseId: f.providerLeaseId, remoteCwd } }; - await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: f.companyId, sourceIssueId: f.issueId, + if (sourceLayout !== "published") await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: f.companyId, sourceIssueId: f.issueId, name: "Completed remote work", cwd: remoteCwd, providerType: "cloud_sandbox", mode: "isolated_workspace", strategyType: "project_primary", status: "active", metadata: { remoteWorkspace: authority } }); await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, f.issueId)); @@ -396,7 +430,7 @@ describe("automatic retained workspace recovery", () => { canonicalSha256: "contract", createdByActorType: "system", createdByActorId: "test" }); await db.update(heartbeatRuns).set({ completionContractId: contractId, resultJson: {}, runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" }, - remoteWorkspace: { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId: lease.id, + remoteWorkspace: preparedReference ?? { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId: lease.id, providerLeaseId: f.providerLeaseId, remoteCwd, epoch: 1, state: "prepared", snapshotId: null }, } }).where(eq(heartbeatRuns.id, f.runId)); await db.insert(nativeRunResults).values({ id: resultId, companyId: f.companyId, issueId: f.issueId, runId: f.runId, @@ -407,11 +441,11 @@ describe("automatic retained workspace recovery", () => { await db.update(issueRecoveryActions).set({ cause: "native_workspace_sync_out_retry_exhausted", evidence: { runId: f.runId } }) .where(eq(issueRecoveryActions.id, f.actionId)); await adoptSystemRecovery(db, f); - return { ...f, source, workspaceId, resultId, git, commit }; + return { ...f, ...(sourceLayout === "published" ? { remote: remoteCwd } : {}), source, workspaceId, resultId, git, commit }; } - async function interruptedRemotePreparation(interruption: "shutdown" | "lease" = "shutdown") { - const f = await acceptedRemoteExport(); + async function interruptedRemotePreparation(interruption: "shutdown" | "lease" = "shutdown", sourceLayout: RecoverySourceLayout = "retained") { + const f = await acceptedRemoteExport(sourceLayout); await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); await db.delete(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)); const recovery = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] }; @@ -437,8 +471,8 @@ describe("automatic retained workspace recovery", () => { return f; } - async function failedRemotePreparation() { - const f = await interruptedRemotePreparation(); + async function failedRemotePreparation(sourceLayout: RecoverySourceLayout = "retained") { + const f = await interruptedRemotePreparation("shutdown", sourceLayout); const source = await run(f.runId); const error = "Repository access is no longer available; refresh the repository catalog"; await terminalizeLegacyExecution({ db, run: source, status: "failed", reconcileIfNeeded: false, @@ -691,9 +725,9 @@ describe("automatic retained workspace recovery", () => { expect(await readFile(path.join(cwd, "saved.txt"), "utf8")).toBe("existing task work\n"); }); - it.each(["complete", "catalog", "catalog_warning", "catalog_unknown_warning", "catalog_provider_warning"] as const)( + it.each(["complete", "catalog", "catalog_warning", "catalog_unknown_warning", "catalog_provider_warning", "invalid_published_root"] as const)( "dispatches the recovered failed-preparation Retry once through the native provider boundary (%s)", async mode => { - const unavailableCatalog = mode !== "complete", withWarning = mode.includes("warning"); + const unavailableCatalog = mode !== "complete" && mode !== "invalid_published_root", withWarning = mode.includes("warning"); const localStorage = getStorageService(); const databaseUrl = process.env.DATABASE_URL, masterKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY; const experimental = await instanceSettingsService(db).getExperimental(); @@ -705,7 +739,7 @@ describe("automatic retained workspace recovery", () => { storage = vi.spyOn(storageBackend, "getStorageService").mockReturnValue({ ...localStorage, provider: "s3" }); process.env.DATABASE_URL = connectionString; process.env.PAPERCLIP_SECRETS_MASTER_KEY = "63".repeat(32); await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true, enableIsolatedWorkspaces: true }); - const f = await failedRemotePreparation(); + const f = await failedRemotePreparation(mode === "invalid_published_root" ? "raw" : "published"); await recoverRetainedWorkspace(f.input); const source = await run(f.runId), nextDirectory = path.join(root, randomUUID(), "next"); await mkdir(nextDirectory, { recursive: true }); @@ -809,6 +843,15 @@ describe("automatic retained workspace recovery", () => { eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "running"))); await heartbeat.resumeQueuedRuns(); await heartbeat.drainActiveRunExecutions(); const completed = await run(successors[0].id); + if (mode === "invalid_published_root") { + expect(completed).toMatchObject({ status: "failed", runtimeMode: "legacy", errorCode: "setup_failed", + error: "remote_workspace_published_path_invalid" }); + expect(inputs).toHaveLength(0); expect(startTurn).not.toHaveBeenCalled(); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, completed.id))).toHaveLength(0); + expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n"); + expect((await run(f.runId)).error).toBe(source.error); + return; + } if (unavailableCatalog) { expect(completed).toMatchObject({ status: "failed", errorCode: "setup_failed", error: "Repository access is no longer available; refresh the repository catalog" }); @@ -855,6 +898,7 @@ describe("automatic retained workspace recovery", () => { expect({ status: completed.status, error: completed.error, runtimeMode: completed.runtimeMode }) .toEqual({ status: "succeeded", error: null, runtimeMode: "native" }); expect(inputs).toHaveLength(1); expect(startTurn).toHaveBeenCalledOnce(); + expect(path.basename(path.dirname(inputs[0].workspace.cwd!))).toBe(".paperclip-workspaces"); expect(completed.runnerProfileJson?.nativeExecutionInput).toMatchObject({ provider: { kind: "codex", model: "gpt-6.1-sol" } }); expect(await readFile(path.join(inputs[0].workspace.cwd!, "continued.txt"), "utf8")).toBe("continued from preserved work\n"); expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, completed.id)))