diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 2c6fc2218d0..c555895265b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1343,6 +1343,14 @@ ordinary Retry can continue. Finalized workspaces, mismatched sources and accepted results keep their existing settlement paths. Reassignment does not discard the stopped run's files or grant authority over the new assignee. +For older failed native runs that lack this bookkeeping, a deliberate Retry +registers preservation from the exact prepared workspace descriptor and assigns +the existing bounded maintenance owner. A live finalization owner or mismatched +source cannot be adopted. Registration does not prove termination or authorize +replay: the provider stop receipt, saved-file receipt, current execution authority +and effect fences still control successor admission. Repeated clicks retain the +same recovery generation and request. + Every continuation carries the triggering request, ordered user direction, interaction outcomes, completed work, and explicit history coverage. A delivered message remains part of the task's request after its connection or approval resolves. The original title is background; a completed Notion read does not satisfy a later Gmail request. Author and source-trust boundaries survive rendering into both native and legacy prompts. Missing required history must be fetched before dispatch rather than described as complete. ### Interrupted conversation continuation diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index 7f212f4a3d2..23927ec4db3 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -243,6 +243,13 @@ replace the owner's generation or revision. Resolved files-only recovery retains its no-replay restriction. The local PostgreSQL regression checks the producer, worker, and physical source receipt; it is not deployed provider acceptance. +An authorized deliberate Retry also registers missing historical native +preservation from an exact prepared descriptor and an unowned terminal +finalization. It adopts only the existing bounded maintenance capability. A live +owner or mismatched allocation cannot supply this proof. No termination receipt +or execution authority is created by registration; connected Retry still waits +behind recovery and rechecks its existing external identity. + The internal preservation resume path grants no board or agent execution API. It requires the exact source action, company, run, lease and environment, a held physical workspace owner, and its current stop-only cleanup claim. The diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 8514cd446be..307479e38b9 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -35,6 +35,8 @@ Agent access: [companion](../agent-access-map/recovery.md). hosted provider recovery remains separate acceptance. Native finalization deadline failures also retain the exact prepared source for the same stop-and-save workflow before Retry admits another executor. + Retry reconstructs missing preservation for historical failures only from an + exact prepared source without a live finalization owner. - `unknown-outcome`: `recovery/unknown-outcome` — uncertain external side effects require inspection and an explicit decision, not automatic replay. - `user-continuation`: `recovery/user-continuation` — saved user messages after a stop are reconsidered once @@ -269,6 +271,10 @@ completed obligations followed by quiet paused input. The physical export regression holds one export while another completes, then verifies a third Retry can use the free slot and shutdown joins the held work. These local checks do not prove recovery of a deployed allocation. +The historical deadline case starts without preservation bookkeeping, holds the +provider stop reply, and verifies that HTTP Retry cannot admit a successor until +the real temporary files are restored. Duplicate requests then admit one queued +turn. GitHub and Telegram admission also retain the exact pending recovery. Automated: [disposition notices](../../ui/src/components/DispositionRecoveryNotice.test.tsx) tests pending/error/gated states and waiting for the real retry acknowledgement. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 095ca9aa105..82f108ec69e 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -9,7 +9,7 @@ import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@ import { githubResponseAuthorityFields } from "../services/chat-channels/github/response-authority-read.js"; import * as activityLogModule from "../services/activity-log.js"; import * as connectionCredentialBindings from "../services/connection-credential-bindings.js"; -import { chatSlackManagerGrants, chatSlackRegistrations, toolOauthStates } from "@paperclipai/db"; +import { chatSlackManagerGrants, chatSlackRegistrations, toolOauthStates, environments } from "@paperclipai/db"; import type { PaperclipCloudConnector, SealedConnectorCredentials } from "../services/paperclip-cloud-connector.js"; import { buildSlackAppManifest, SLACK_MANAGER_SCOPES } from "@paperclipai/shared"; import { registerChatActionWork, dueChatAction } from "../services/chat-action-work.js"; @@ -61435,7 +61435,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ); it.each((["github", "telegram"] as const).flatMap(provider => - (["pending", "historical_sibling", "foreign_source_hold", "automatic_scheduled", "automatic_running", "automatic_awaiting_evidence", "exhausted", "wrong_generation", "paused_endpoint", "paused_after_wait", "paused_agent_after_wait", "edited_after_wait", "superseded_after_wait"] as const).map(gate => ({ provider, gate }))))( + (["unregistered", "pending", "historical_sibling", "foreign_source_hold", "automatic_scheduled", "automatic_running", "automatic_awaiting_evidence", "exhausted", "wrong_generation", "paused_endpoint", "paused_after_wait", "paused_agent_after_wait", "edited_after_wait", "superseded_after_wait"] as const).map(gate => ({ provider, gate }))))( "holds an ordinary $provider Retry behind its exact remote recovery: $gate", async ({ provider, gate }) => { const context = await failedChatRetryFixture(provider, undefined, true); try { @@ -61462,7 +61462,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { firstFailureAt: Date.now() - 1000, deadlineAt: Date.now() + 60_000, preservationOperationIds: [randomUUID()], }; - const canResume = gate === "pending" || gate === "historical_sibling" || gate.startsWith("automatic_"); + const canResume = gate === "unregistered" || gate === "pending" || gate === "historical_sibling" || gate.startsWith("automatic_"); await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, kind: "active_run_watchdog", status: "resolved", ownerType: "system", cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(), @@ -61473,6 +61473,19 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { systemRecovery: priorRecovery, }, }); + if (gate === "unregistered") { + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); + const [environment] = await db.insert(environments).values({ name: context.runId, driver: "sandbox" }).returning(); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)); + await db.update(heartbeatRuns).set({ resultJson: null, 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: issueId, leaseId, providerLeaseId: lease.providerLeaseId, + remoteCwd: lease.metadata.remoteCwd, resourceDisposition: "keep_running" } }, + }).where(eq(heartbeatRuns.id, context.runId)); + await db.update(environmentLeases).set({ environmentId: environment.id, status: "active", cleanupStatus: null, leasePolicy: "reuse_by_environment" }) + .where(eq(environmentLeases.id, leaseId)); + } const receipt = await requestTaskRetry({ db, companyId, issueId, actorId: "owner-user", actorSource: "cloud_tenant", observedRunId: context.runId, idempotencyKey: randomUUID(), prepareChatRetry: context.service.prepareFailedChatRunRetry }); diff --git a/server/src/services/task-retry.test.ts b/server/src/services/task-retry.test.ts index 693ee1a4e8d..d760071806b 100644 --- a/server/src/services/task-retry.test.ts +++ b/server/src/services/task-retry.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { agents, agentWakeupRequests, companies, createDb, heartbeatRuns, issues, issueRecoveryActions, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db"; +import { agents, agentWakeupRequests, companies, createDb, environmentLeases, environments, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { priorRecoveryApiReceipt, unresolvedRecoveryToolInvocation } from "./recovery-effect-receipts.js"; import { requestTaskRetry } from "./task-retry.js"; @@ -10,6 +10,7 @@ import { settleUnrecoverableExecutions } from "./execution-recovery-resolution.j import { createSystemRecovery, failSystemRecovery, readSystemRecovery, SystemRecoveryFailure } from "./system-recovery-state.js"; import { instanceSettingsService } from "./instance-settings.js"; import { heartbeatService } from "./heartbeat.js"; +import { taskChatRetryCanAwaitRecovery, taskChatRetryExecutionStopped } from "./chat-channels/task-retry-authority.js"; describe("durable task Retry admission", () => { let postgres: Awaited>; @@ -56,6 +57,53 @@ describe("durable task Retry admission", () => { expect(await requestTaskRetry(input)).toMatchObject({ accepted: false, stale: true }); expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, input.companyId))).toHaveLength(0); }); + it.each(["prepared", "mismatched", "owned"] as const)("reconstructs missing native preservation on deliberate Retry only for an exact unowned source: %s", async state => { + const input = await seed(); + const [issue] = await db.select().from(issues).where(eq(issues.id, input.issueId)); + const [environment] = await db.insert(environments).values({ name: input.observedRunId, driver: "sandbox" }).returning(); + const [lease] = await db.insert(environmentLeases).values({ companyId: input.companyId, issueId: input.issueId, + heartbeatRunId: input.observedRunId, environmentId: environment.id, provider: "daytona", providerLeaseId: randomUUID(), + status: "active", leasePolicy: "reuse_by_environment", metadata: { remoteCwd: "/work" }, + }).returning(); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: input.issueId, + nativePhase: "terminal_failure", 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: input.observedRunId, leaseId: lease.id, + providerLeaseId: state === "mismatched" ? randomUUID() : lease.providerLeaseId, + remoteCwd: "/work", resourceDisposition: "keep_running" }, + } }).where(eq(heartbeatRuns.id, input.observedRunId)); + await db.insert(nativeRunFinalizations).values({ companyId: input.companyId, issueId: input.issueId, + runId: input.observedRunId, phase: "terminal_failure", leaseOwner: state === "owned" ? "live-controller" : null }); + await db.insert(issueRecoveryActions).values({ companyId: input.companyId, sourceIssueId: input.issueId, + kind: "active_run_watchdog", status: "resolved", ownerType: "board", + cause: "execution_finalization_deadline_exceeded", fingerprint: randomUUID(), outcome: "blocked", + nextAction: "Inspect the failed run.", evidence: { runId: input.observedRunId, + providerOwnership: "unverified", automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }); + + const receipt = await requestTaskRetry(input); + expect((await requestTaskRetry(input)).requestId).toBe(receipt.requestId); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.observedRunId)); + const [retained] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)); + const repairs = (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, input.issueId))) + .filter(action => action.evidence.workspaceRestoreRecovery); + if (state === "prepared") { + expect(source.resultJson).toMatchObject({ workspaceRestoreRecovery: { leaseIds: [lease.id] } }); + expect(retained).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure" }); + expect(repairs).toHaveLength(1); + expect(readSystemRecovery(repairs[0].evidence)).toMatchObject({ state: "scheduled", intentId: receipt.requestId }); + expect(await taskChatRetryCanAwaitRecovery(db, source, input.issueId, receipt.requestId!)).toBe(true); + } else { + expect(source.resultJson).toBeNull(); + expect(retained).toMatchObject({ status: "active", leasePolicy: "reuse_by_environment" }); + expect(repairs).toHaveLength(0); + } + expect(await taskChatRetryExecutionStopped(db, source)).toBe(false); + expect(retained.metadata).not.toHaveProperty("remoteExecutionTermination"); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, input.companyId))).toHaveLength(1); + expect((await db.select().from(issues).where(eq(issues.id, input.issueId)))[0]) + .toMatchObject({ status: "blocked", assigneeAgentId: issue.assigneeAgentId, executionRunId: null, checkoutRunId: null }); + }); it("requires current Agent Chat ownership and enablement even when replaying a retry receipt", async () => { const input = await seed(); const settings = instanceSettingsService(db); diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index afbfbb34d08..f1ec119386f 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -7,10 +7,13 @@ import { agentWakeupRequests, chatActions, chatConversations, chatEndpoints, hea import { conflict, forbidden, notFound } from "../errors.js"; import { persistActivity } from "./activity-log.js"; import { createSystemRecovery, readSystemRecovery } from "./system-recovery-state.js"; -import { updateSystemRecovery } from "./system-recovery.js"; +import { adoptSystemRecovery, updateSystemRecovery } from "./system-recovery.js"; import { instanceSettingsService } from "./instance-settings.js"; import { notifyChatActionWork } from "./chat-work-notifications.js"; import { authorizationService } from "./authorization.js"; +import { preserveNativeFailedWorkspace } from "./native-runtime/native-failed-workspace-recovery.js"; +import { NativeWorkspaceFinalizationBusyError } from "./native-runtime/native-workspace-finalization-ownership.js"; +import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; export async function taskRetryMutationAllowed(db: Db, input: { companyId: string; issueId: string; actorId: string; actorSource: TaskRetryIntent["actorSource"]; @@ -116,9 +119,22 @@ export async function requestTaskRetry(input: { active = undefined; } const busy = ["queued", "running", "scheduled_retry"].includes(latest.status); + if (!busy) { + // Historical deadline failures can predate their preservation record. + // Register the exact prepared source; this proves neither a stopped + // executor nor permission to replay its work. + try { await preserveNativeFailedWorkspace(db, latest); } + catch (error) { if (!(error instanceof NativeWorkspaceFinalizationBusyError)) throw error; } + } const records = await tx.select().from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, issue.companyId), eq(issueRecoveryActions.sourceIssueId, issue.id), sql`${issueRecoveryActions.evidence}->>'runId' = ${latest.id}`)).for("update"); + for (let index = 0; !busy && index < records.length; index++) { + const action = records[index]; + if (!hasRequiredWorkspaceRecovery(action.evidence) || readSystemRecovery(action.evidence)) continue; + await adoptSystemRecovery(db, { companyId: issue.companyId, issueId: issue.id, runId: latest.id, actionId: action.id }); + [records[index]] = await tx.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)); + } // An unadmitted request can outlive its exhausted repair. A new deliberate // click replaces that generation; a replay of the same key returned above. if (active && !busy && !active.runId && records.some(action => diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index e9519384bab..4eb74906cb3 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -72,7 +72,7 @@ describe("automatic retained workspace recovery", () => { if (priorHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = priorHome; if (priorInstance === undefined) delete process.env.PAPERCLIP_INSTANCE_ID; else process.env.PAPERCLIP_INSTANCE_ID = priorInstance; }); - async function seed(mode = "native", nativeDescriptor = true) { + async function seed(mode = "native", nativeDescriptor = true, resourceDisposition: "destroy" | "keep_running" = "destroy") { const companyId = randomUUID(), issueId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), actionId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Workspace recovery", defaultResponsibleUserId: "board", issuePrefix: `R${companyId.slice(0, 6)}` }); await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner", status: "idle", runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }); @@ -101,7 +101,7 @@ describe("automatic retained workspace recovery", () => { targets.set(lease.id, target); if (nativeDescriptor) { const prepared = await prepareNativeWorkspaceSync({ db, companyId, workspaceId: issueId, workspaceLocalDir: host, - runId, lease: lease as EnvironmentLease, target, resourceDisposition: "destroy" }); + runId, lease: lease as EnvironmentLease, target, resourceDisposition }); expect(prepared).not.toBeNull(); } else { // Provider boundary only: real temporary files, no native baseline. The @@ -137,6 +137,64 @@ describe("automatic retained workspace recovery", () => { const action = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!; const run = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; + it("repairs an unregistered historical deadline failure through HTTP Retry without admitting work before verified stop", async () => { + const f = await seed("native", true, "keep_running"); + const source = await run(f.runId); + await db.update(heartbeatRuns).set({ resultJson: null, errorCode: "execution_finalization_deadline_exceeded" }) + .where(eq(heartbeatRuns.id, f.runId)); + await db.update(environmentLeases).set({ status: "active", leasePolicy: "reuse_by_environment", + releasedAt: null, cleanupStatus: null, metadata: f.lease.metadata }).where(eq(environmentLeases.id, f.lease.id)); + await db.update(issueRecoveryActions).set({ status: "resolved", outcome: "blocked", + cause: "execution_finalization_deadline_exceeded", evidence: { runId: f.runId, providerOwnership: "unverified", + automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }).where(eq(issueRecoveryActions.id, f.actionId)); + // Leave admission observable without substituting a fake model turn. + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: source.agentId, status: "running" }); + let releaseStop!: () => void; + const stopped = new Promise(resolve => { releaseStop = resolve; }); + f.runtime.retryPendingSandboxTeardown.mockImplementation(async () => { + await stopped; + return { providerLeaseId: f.providerLeaseId, state: "stopped" }; + }); + const errors = vi.fn(), heartbeat = heartbeatService(db); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: errors }); + const worker = registerWorkspaceRecoveryWork(db, coordinator, f.environmentRuntime, heartbeat.resumeRemoteStopComments); + const successors = () => db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + try { + await worker.ready; + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { req.actor = { type: "board", source: "local_implicit", userId: "board" }; next(); }); + app.use("/api", issueRoutes(db, {} as Parameters[1])); + app.use(errorHandler); + const body = { idempotencyKey: randomUUID(), observedRunId: f.runId }; + const retry = await request(app).post(`/api/issues/${f.issueId}/retry`).send(body).expect(202); + expect((await request(app).post(`/api/issues/${f.issueId}/retry`).send(body).expect(202)).body.requestId).toBe(retry.body.requestId); + await expect.poll(() => f.runtime.retryPendingSandboxTeardown.mock.calls.length, { timeout: 20_000 }).toBe(1); + await heartbeat.resumeTaskRetry(f.companyId, retry.body.requestId); + expect(await successors()).toHaveLength(0); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].metadata) + .not.toHaveProperty("remoteExecutionTermination"); + await expect(readFile(path.join(f.host, "saved.txt"), "utf8")).rejects.toMatchObject({ code: "ENOENT" }); + releaseStop(); + await expect.poll(async () => (await successors()).length, { timeout: 20_000 }).toBe(1); + const [successor] = await successors(); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n"); + expect(await readFile(path.join(f.host, "newer.txt"), "utf8")).toBe("newer local work\n"); + expect(await readFreshWorkspaceReadiness(db, successor)).toMatchObject({ mode: "resume_preserved_workspace" }); + await Promise.all([heartbeat.resumeTaskRetry(f.companyId, retry.body.requestId), heartbeat.resumeRemoteStopComments(await run(f.runId))]); + expect(await successors()).toHaveLength(1); + expect(f.runtime.resumeRunLease).toHaveBeenCalledOnce(); + expect((await run(f.runId)).status).toBe("failed"); + const reloaded = await request(app).get(`/api/issues/${f.issueId}/recovery-actions`).expect(200); + expect(reloaded.body.active).toBeNull(); + const records = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + expect(records.some(record => readSystemRecovery(record.evidence)?.state === "complete")).toBe(true); + expect(errors).not.toHaveBeenCalled(); + } finally { releaseStop(); await coordinator.stop(); await worker.stop(); } + }, 40_000); + it("wakes physical preservation from HTTP Retry before passive history without replaying the provider", async () => { for (let index = 0; index < 5; index++) { const old = await seed();