diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index c25c4e1449d..98efb8a882f 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -961,6 +961,8 @@ The handshake failure code is distinct from a session-identity mismatch. A timeo ### Explicit recovery actions +Connected task Retry can continue a failed legacy setup before provider dispatch. The bootstrap marker is insufficient by itself. The server requires its exact error and durable measurement events, a dispatch-only Runner profile, and no retained runtime, process, output, coordinator, result or environment lease. Current source authorization and every normal admission gate still apply. An inherited session display hint grants no resume authority. The original failed attempt remains unchanged, and an independent unfinished writer keeps its hold. + An explicit recovery action is a typed liveness repair path for a source issue. It is the recovery primitive; the action can be rendered directly on the source issue or backed by a separate recovery issue when the repair needs its own work item. A terminal native failure can retain a result accepted before checkpoint or cleanup failed. That result is historical evidence, not a live controller. New user input may start a fresh turn after the controller and execution environment have stopped and ordinary admission checks pass. Preserve the failed run, its result, and its recovery budget. Do not commit the old result, infer action outcomes, or replay the failed turn. A message saved while cleanup is pending must be reconsidered after verified cleanup and delivered once. External chat messages retain their original inbound action, wake receipt and actor. Cleanup delegates reconsideration to the live inbound owner, which revalidates the endpoint, conversation generation, principal and exact saved source before normal admission. The existing stranded-queue sweep uses the same owner after a lost callback or restart; it cannot substitute a generic assignment wake. A rejected source must not prevent other saved messages from being reconsidered. Admission must consume its deferred receipt in the same transaction that creates the new run, including in agent chat, where later messages keep their separate turns. Completing that run must not promote the consumed receipt again. Workspace-export repair retains its separate saved-result recovery path. diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index f68593c9b6a..4d1d4d23f3e 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -304,8 +304,8 @@ no external inbound authority or permission to replay a terminal result; an uncertifiable source directs the user to send a new task message. The real HTTP/queue regression proves this boundary locally, not hosted provider recovery. -For external chat, the same canonical Retry owner may certify the structured -isolated-directory binding failure that occurs before environment allocation. +For external chat, the same canonical Retry owner may certify an isolated-directory +binding failure or a fully recorded setup failure before provider dispatch. It requires exact task/source authority and trusted settlement with no process, lease, native result or unknown execution evidence. Other validation reasons and the older direct chat-retry entry point remain denied. Local HTTP/dispatch proof diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 7dc31fab4cb..4b3786861b9 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -64502,6 +64502,107 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return { ...f, input, result }; } + it.each((["github", "telegram"] as const).flatMap(provider => + (["writer_setup", "other_setup", "late_event", "retained_native", "accepted_result", "source_changed", "untrusted_metadata", "live_controller", "process", "environment", "prior_writer"] as const).map(mode => ({ provider, mode }))))( + "retries a fully recorded never-dispatched $provider setup through HTTP: $mode", async ({ provider, mode }) => { + const f = provider === "github" ? await currentGitHubTaskRetryFixture() : await failedChatRetryFixture("telegram", undefined, true); + const { issueRoutes } = await import("../routes/issues.js"); + const companyId = f.fixture.companyId, issueId = f.issue.id; + 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); + try { + if (provider === "telegram") { + await db.update(agents).set({ adapterType: "paperclip_runner", status: "running", + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } }) + .where(eq(agents.id, f.fixture.assignedAgentId)); + await db.insert(heartbeatRuns).values({ companyId, agentId: f.fixture.assignedAgentId, + status: "running", startedAt: new Date() }); + f.wakeup.mockImplementation(heartbeatService(db).wakeup); + } + await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.runId)); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const diagnostic = mode === "other_setup" ? "A different preparation failure" : "The workspace still has an unfinished writer or save operation."; + await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null, nativePhase: null, + errorCode: "setup_failed", error: diagnostic, sessionIdBefore: "inherited-display-hint", + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + usageJson: { ledgerScope: { issueId, projectId: null, billingCode: null } }, + resultJson: { stopReason: "adapter_failed", executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + timeoutFired: false, timeoutSource: "config", timeoutConfigured: false, effectiveTimeoutSec: 0, + ...(mode === "untrusted_metadata" ? { providerOutput: "Untrusted adapter data" } : {}), + }, + ...(mode === "live_controller" ? { controllerLeaseExpiresAt: new Date(Date.now() + 60_000) } : {}), + ...(mode === "process" ? { processPid: 999999999, processGroupId: 999999999 } : {}), + }).where(eq(heartbeatRuns.id, f.runId)); + await db.insert(heartbeatRunEvents).values([ + { companyId, agentId: f.fixture.assignedAgentId, runId: f.runId, seq: 1, + eventType: "error", stream: "system", level: "error", message: diagnostic }, + { companyId, agentId: f.fixture.assignedAgentId, runId: f.runId, seq: 2, + eventType: "lifecycle", stream: "system", message: "task.run.measured", + payload: { phase: "task.run.measured", outcome: "failed", durableSettlement: true, durationMs: 5 } }, + ]); + if (mode === "late_event") await db.insert(heartbeatRunEvents).values({ companyId, + agentId: f.fixture.assignedAgentId, runId: f.runId, seq: 3, eventType: "session.started", + sourceInstanceId: randomUUID(), payload: { prpEvent: { sourceKind: "runner" } } }); + if (mode === "retained_native") { + await db.update(heartbeatRuns).set({ nativeIssueId: issueId }).where(eq(heartbeatRuns.id, f.runId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: f.runId, phase: "terminal_failure" }); + } + if (mode === "environment") await db.insert(environmentLeases).values({ companyId, issueId, + heartbeatRunId: f.runId, provider: "daytona", providerLeaseId: randomUUID(), status: "active" }); + if (mode === "prior_writer") { + const previousId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: previousId, companyId, nativeIssueId: issueId, + agentId: f.fixture.assignedAgentId, status: "failed", runtimeMode: "native", nativePhase: "terminal_failure", + createdAt: new Date(Date.now() - 60_000), finishedAt: new Date(Date.now() - 30_000) }); + await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: previousId, + provider: "daytona", providerLeaseId: randomUUID(), status: "pending_cleanup", cleanupStatus: "failed" }); + await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", status: "resolved", ownerType: "system", outcome: "blocked", + cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(), nextAction: "Previous writer remains unknown", + evidence: { runId: previousId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }); + } + if (mode === "source_changed") await db.update(issueComments).set({ body: "Edited after admission", updatedAt: new Date() }) + .where(eq(issueComments.id, String(f.action.payload.commentId))); + if (mode === "accepted_result") { + const contractId = randomUUID(), resultId = randomUUID(); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: {}, + canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" }); + await db.update(heartbeatRuns).set({ nativeIssueId: issueId, completionContractId: contractId }) + .where(eq(heartbeatRuns.id, f.runId)); + await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId: f.runId, + completionContractId: contractId, serverFingerprint: resultId, schemaStatus: "accepted", canonicalSha256: resultId, + resultJson: { terminal: { runTerminalState: "succeeded" }, summary: "Existing output stays history" } }); + } + const original = (await readHeartbeatRunsById(f.runId))[0]; + const response = await request(app).post(`/api/issues/${issueId}/retry`).send({ observedRunId: f.runId, + idempotencyKey: randomUUID() }); + if (mode === "writer_setup" || mode === "other_setup") { + expect(response.status, JSON.stringify(response.body)).toBe(202); + 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, wakeCommentIds: [f.action.payload.commentId] } }); + expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(1); + } else { + expect([202, 409], JSON.stringify(response.body)).toContain(response.status); + if (response.status === 202) { + const outcome = await f.service.processFailedChatRunRetry(response.body.chatActionId); + expect(outcome.status).not.toBe("queued"); + } + expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(0); + } + expect((await readHeartbeatRunsById(f.runId))[0]).toEqual(original); + } finally { await f.service.shutdown(); } + }, + ); + it("retries saved GitHub input after a successful run without replaying that run", async () => { const f = await successfulGitHubQueuedInputFixture(); try { diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index d49bbb6aafe..aec7ce9a3fb 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -218,16 +218,19 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti const failedRetry = run.errorCode === "chat_failed_run_retry_not_authorized" && run.error === retryDiagnostic; const validation = run.resultJson?.workspaceValidation as Record | undefined; const recovery = run.resultJson?.executionRecovery as Record | undefined; - const workspacePreparation = allowWorkspacePreparation && run.errorCode === "workspace_validation_failed" && + const workspaceValidation = run.errorCode === "workspace_validation_failed" && validation?.reason === "isolated_task_directory_binding_mismatch" && typeof validation.issueId === "string" && validation.issueId === run.contextSnapshot?.issueId && (run.issueId === null || run.issueId === validation.issueId) && (run.contextSnapshot?.taskId === undefined || run.contextSnapshot.taskId === validation.issueId) && - Object.keys(validation).every(key => ["reason", "issueId"].includes(key)) && + Object.keys(validation).every(key => ["reason", "issueId"].includes(key)); + const setupPreparation = run.errorCode === "setup_failed" && typeof run.error === "string" && + run.error.trim().length > 0 && validation === undefined; + const workspacePreparation = allowWorkspacePreparation && (workspaceValidation || setupPreparation) && recovery?.kind === "bootstrap" && recovery.providerWorkStarted === false && Object.keys(recovery).every(key => ["kind", "providerWorkStarted"].includes(key)) && run.resultJson?.stopReason === "adapter_failed" && - Object.keys(run.resultJson).every(key => ["executionRecovery", "workspaceValidation", "stopReason", + Object.keys(run.resultJson).every(key => ["executionRecovery", ...(workspaceValidation ? ["workspaceValidation"] : []), "stopReason", "effectiveTimeoutSec", "timeoutConfigured", "timeoutFired", "timeoutSource"].includes(key)); const diagnostic = failedRetry ? retryDiagnostic : workspacePreparation ? run.error : "reviewed_chat_execution_binding_not_authorized"; const dispatchOnlyProfile = failedRetry || workspacePreparation;