From 3f1ad6b25b078c1e061dacff01440f07af17fa76 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 23 Oct 2026 07:51:34 -0700 Subject: [PATCH] fix(chat): resume task Retry after retained-source containment (#17317) ## Thinking Path > - Paperclip manages agent work through task and execution ownership. > - A task Retry must retain the user's request while the system verifies an old executor. > - Connected conversations also retain their external principal and accepted source batch. > - Chat Retry admission required containment before it could save a wait for that recovery. > - A saved chat receipt could also remain deferred after containment became available. > - This pull request lets the exact recovery-bound request wait and resume through its chat owner. > - Normal admission still checks containment, current access and policy before one successor starts. ## Linked Issues or Issue Description **What happened?** An ordinary task Retry on a connected conversation failed with `failed_chat_retry_not_authorized` while its retained remote executor still needed verification. The new recovery generation existed, but the chat action failed before it could save a dispatch wait. This occurred after the repairs in #17267 were confirmed live. The local regression found a second gap. A processed chat action with a deferred wake was not reconsidered after containment. Re-entering normal admission also tried to insert its existing durable receipt again. A historical hold for the same failed run could also hide the current Retry's workspace strategy, so the request still waited after the old executor was contained. Preparation could then read that sibling instead of the recovery record that owned the fresh successor, losing the fresh-workspace decision. **Expected behavior** Save the exact authorized Retry while its matching recovery generation verifies the executor. Once containment is recorded, reconsider the same chat receipt through normal admission. Start one successor with the original source batch. Keep publication, principal, pause and budget checks. **Steps to reproduce** 1. Admit a GitHub or Telegram request and record a failed native run with a retained remote lease. 2. Leave the executor's termination unconfirmed and exhaust its prior recovery generation. 3. Use ordinary task Retry to create a new recovery generation. 4. Process the connected Retry. The old admission code fails instead of retaining a wait. 5. With a deferred receipt, record exact containment and reconcile again. The old dispatch path leaves it stranded. **Paperclip version or commit** Reproduced from `412914cfa78d5abefcc34d3eb8d4bb86e461caac`. The branch is rebased onto current master. **Deployment mode** Hosted tenant for the reported failure. Isolated local PostgreSQL and the assembled heartbeat/chat services for the regression. Refs #17267 and #16965. This change does not add a Cloud dependency or change the absence protocol. ## What Changed - Permit a saved admission wait only for the exact current task Retry and active source recovery generation. - Keep promotion and execution dependent on independent containment proof. - Bind a Retry to active automatic recovery without replacing its generation, attempt count or deadline. - Share the current exact Retry strategy across historical holds for the same run. Retain each hold's stop and effect checks. - Select the canonical recovery record that proves the committed fresh successor during workspace preparation. - Let task receipt reconciliation return the same deferred action to its chat owner. - Recheck current board and external authority, then update the existing wake atomically with its successor. - Preserve the exact chat source batch. Keep generic board-comment filtering unchanged. - Add GitHub and Telegram cases for waiting, containment, one successor, stale generations, source edits, endpoint pause and supersession. - Update recovery documentation and capability maps. ## Verification - Red: real PostgreSQL tests reproduced the admission rejection for GitHub and Telegram. Six active-recovery cases then failed on the generation-ID assumption. Two same-run historical-hold cases remained deferred after containment on the old strategy lookup. A preparation regression then returned no readiness when the historical hold was stored first. - Green: 73 selected chat integration tests and 394 continuation, task Retry, workspace recovery, recovery-state and workspace-preparation tests passed. The chat tests use the real chat admission and heartbeat queue. Unrelated capacity stays occupied so no real provider starts. They assert no executor before containment, one successor afterward, unchanged source leases while waiting, preserved recovery budgets, and denial after authority changes. A separate source hold still blocks work. Both sibling insertion orders pass through the committed-readiness lookup and real workspace preparation: the new directory has independent Git storage and files, and unpublished old work remains intact. - `pnpm -r typecheck`, `node scripts/check-maps.mjs` and `git diff --check` passed after the review fixes. The source build passed on `f6660efa48e37a80cd9857217a537606de85b5c8`. - The local host reached its PostgreSQL semaphore limit. Successful runs used a test-only loader to reduce disposable cluster connection and worker limits. No repository database helper or application setting changed. - The full local `pnpm test:run` began before the final review edits, failed in its first server phase and was stopped after those failures. Later phases did not run. This is not a full local pass. The transient Retry-notice case passes after fixture isolation; the four Discord close-owned publication cases also pass in isolation. Other broad-run failure causes are not established. - Exact-head CI and security checks pass on `f6660efa48e37a80cd9857217a537606de85b5c8`; Greptile is 5/5 with zero unresolved review threads. Optional Paperclip Review completed neutral with a 5/5 static assessment of all 11 changed files and no findings; it did not run tests or hosted acceptance. All 57 current-head checks are terminal: 54 success, two skipped and the optional neutral review, with no failures or pending checks. - The real hosted workflow confirmed that the deployed absence repair recorded containment. A later ordinary UI Retry was accepted and saved a durable request, but it remains behind the historical source hold. The useful existing review remains unchanged. This follow-up is not deployed; a live successful successor and final task disposition remain unverified. No missing files are claimed recovered. ## Risks - The existing task reconciliation sweep can restage a deferred chat action. This reuses its receipt and generation; it does not grant new execution or publication authority. - Dispatch must keep the board Retry identity separate from the connected request's principal. The regression checks these bindings and preserves current chat authorization at each phase. - Historical holds reuse a strategy only from the locked same-run recovery that names the current Retry intent. A different run, malformed recovery record or independent source hold remains fenced. - Unknown executors without a matching active recovery still fail admission. Promotion and execution require containment even when admission saved a wait. - No schema or API shape changes. No credential changes, manual deployment or service restart. ## Model Used OpenAI GPT-6 through Codex, with repository tools, code execution and browser verification. The exact served model variant and context window are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/task-recovery.md | 14 ++ maps/agent-access-map/recovery.md | 6 +- maps/feature-map/recovery.md | 10 + .../chat-channels.integration.test.ts | 194 ++++++++++++++++++ .../chat-channels/failed-run-retry.ts | 40 +++- .../chat-channels/task-retry-authority.ts | 26 ++- .../services/explicit-native-continuation.ts | 13 +- .../services/fresh-workspace-continuation.ts | 9 +- server/src/services/heartbeat/queue.ts | 63 ++++-- .../heartbeat/workspace-preparation.test.ts | 44 +++- server/src/services/task-retry.ts | 14 ++ 11 files changed, 398 insertions(+), 35 deletions(-) diff --git a/doc/task-recovery.md b/doc/task-recovery.md index 6773c02b4be..870b8299a34 100644 --- a/doc/task-recovery.md +++ b/doc/task-recovery.md @@ -18,6 +18,20 @@ Connected conversations use their existing inbound authorization owner. A board Retry changes the workspace strategy, not the external principal, source batch, conversation access or publication authority. +A connected-conversation Retry can save its exact request while the matching +system recovery generation verifies the old executor. Saving that wait grants +no execution authority. The chat owner rechecks the same board intent, external +principal and source batch when it resumes the deferred receipt. Containment, +current access, task eligibility, pause and budget gates must still pass before +one successor starts. Restarts reuse the receipt; they do not create another +Retry generation or another external publication. + +Several retained recovery holds can describe the same failed execution. A +historical hold without a system recovery record uses the current exact Retry's +workspace strategy for that run. It does not borrow another run's strategy or +skip any hold's independent containment and effect checks. Attaching a Retry to +active automatic recovery preserves its generation, attempt count and deadline. + Agent Chat additionally requires the current conversation owner and enabled Agent Chat. Admission and dispatch both recheck these conditions under the issue lock. An unadmitted receipt from a former owner cannot own the current diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index 0cee692325b..6d619bafc90 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -69,7 +69,11 @@ board task-mutation authority, not agent retry authority. The only request fields are a UUID idempotency key and observed source run (or null). It accepts no paths, stop attestations or client-selected strategy. The source-scoped exact-run and maintenance APIs retain their existing contracts. Connected-chat -Retry reuses chat admission and its external identity. See the +Retry reuses chat admission and its external identity. It can persist a wait +only for the exact current system recovery generation. The chat owner resumes +that same deferred receipt through normal admission after containment. The +board intent and external principal remain separate authorities; current task, +connection, pause and budget gates still apply. See the [durable contract](../../doc/task-recovery.md). Agent Chat also requires its current conversation owner and feature enablement at admission and locked dispatch; generic board mutation permission does not replace chat ownership. diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 66be99d9dd4..86096436a72 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -67,6 +67,16 @@ Use the task's Retry action. It can create an isolated attempt after old executi is contained even when old-file preservation remains unresolved. Approval, credentials, budget and deliberate pause use their existing specific controls. +For a connected conversation, click Retry while an exact retained source still +needs containment. Observe a saved wait with no new executor. After verified +containment, the same chat receipt must admit one successor with the original +source batch. Reload and repeat reconciliation. Check that a paused endpoint, +changed source, superseded request or revoked access cannot start that work. +The connected-chat database regression in +[chat channel integration](../../server/src/__tests__/chat-channels.integration.test.ts) +exercises admission and queue continuation. It does not establish hosted +provider recovery or successful delivery. + ### `post-stop-message` After a stop, send a new instruction on the same task or inspect one already diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 855f6ee5714..4c253f0251c 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,5 +1,8 @@ import { idleWorkSnapshot } from "../services/task-admission.js"; +import { readFreshWorkspaceReadiness } from "../services/fresh-workspace-continuation.js"; import { requestTaskRetry } from "../services/task-retry.js"; +import { createSystemRecovery } from "../services/system-recovery-state.js"; +import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; import { stageReceiptReactionRemovals } from "../services/chat-receipt-reactions.js"; import type { Db } from "@paperclipai/db"; import * as activityLogModule from "../services/activity-log.js"; @@ -66,6 +69,7 @@ import { eq, inArray, isNotNull, + isNull, like, notInArray, or, @@ -60945,6 +60949,196 @@ 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 }))))( + "holds an ordinary $provider Retry behind its exact remote recovery: $gate", async ({ provider, gate }) => { + const context = await failedChatRetryFixture(provider, undefined, true); + try { + const companyId = context.fixture.companyId, issueId = context.issue.id; + const leaseId = randomUUID(); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, + nativePhase: "terminal_failure", runnerProfileJson: {}, + resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { + schema: "paperclip.workspace-restore-recovery.v1", runId: context.runId, leaseIds: [leaseId], + } }, + }).where(eq(heartbeatRuns.id, context.runId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, + runId: context.runId, phase: "terminal_failure" }); + await db.insert(environmentLeases).values({ id: leaseId, companyId, issueId, + heartbeatRunId: context.runId, provider: "daytona", providerLeaseId: randomUUID(), + status: "pending_cleanup", cleanupStatus: "failed", leasePolicy: "retain_on_failure", + metadata: { remoteCwd: "/home/daytona/paperclip-workspace", pendingCleanupInFlight: false }, + }); + const priorRecovery = { ...createSystemRecovery(context.runId), + state: gate === "automatic_scheduled" ? "scheduled" as const + : gate === "automatic_running" ? "running" as const + : gate === "automatic_awaiting_evidence" ? "awaiting_evidence" as const : "exhausted" as const, + attempts: 2, retriesUsed: 1, attemptId: randomUUID(), + firstFailureAt: Date.now() - 1000, deadlineAt: Date.now() + 60_000, + preservationOperationIds: [randomUUID()], + }; + const canResume = 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(), + nextAction: "Verify and preserve the exact retained source before continuing.", + evidence: { runId: context.runId, automaticRecovery: { replay: "blocked" }, + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] }, + historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: context.runId, leaseIds: [leaseId] }, + systemRecovery: priorRecovery, + }, + }); + const receipt = await requestTaskRetry({ db, companyId, issueId, actorId: "owner-user", + actorSource: "cloud_tenant", observedRunId: context.runId, idempotencyKey: randomUUID(), + prepareChatRetry: context.service.prepareFailedChatRunRetry }); + const [recoveryAction] = await db.select().from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId), + sql`${issueRecoveryActions.evidence}->>'runId' = ${context.runId}`, + )); + if (gate === "historical_sibling") await db.insert(issueRecoveryActions).values({ + companyId, sourceIssueId: issueId, kind: "active_run_watchdog", status: "resolved", ownerType: "system", + cause: "native_session_retry_exhausted", fingerprint: randomUUID(), + nextAction: "Preserve the historical execution outcome.", + evidence: { runId: context.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } }, + }); + if (gate.startsWith("automatic_")) expect(recoveryAction.evidence.systemRecovery).toMatchObject({ + generationId: priorRecovery.generationId, intentId: receipt.requestId, + attempts: priorRecovery.attempts, retriesUsed: priorRecovery.retriesUsed, + attemptId: priorRecovery.attemptId, deadlineAt: priorRecovery.deadlineAt, + preservationOperationIds: priorRecovery.preservationOperationIds, + }); + if (gate === "exhausted" || gate === "wrong_generation") await db.update(issueRecoveryActions).set({ + evidence: { ...recoveryAction.evidence, systemRecovery: { + ...recoveryAction.evidence.systemRecovery as Record, + ...(gate === "exhausted" ? { state: "exhausted" } : { intentId: randomUUID() }), + } }, + }).where(eq(issueRecoveryActions.id, recoveryAction.id)); + if (gate === "paused_endpoint") await db.update(chatEndpoints).set({ status: "paused" }) + .where(eq(chatEndpoints.id, context.endpoint.id)); + // Keep capacity occupied so a regression cannot start a real adapter. + await db.update(agents).set({ adapterType: "paperclip_runner", status: "running", + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } }) + .where(eq(agents.id, context.fixture.assignedAgentId)); + await db.insert(heartbeatRuns).values({ companyId, + agentId: context.fixture.assignedAgentId, status: "running", startedAt: new Date() }); + context.wakeup.mockImplementation(heartbeatService(db).wakeup); + const outcome = await context.service.processFailedChatRunRetry(receipt.chatActionId!); + if (["exhausted", "wrong_generation", "paused_endpoint"].includes(gate)) { + expect(outcome).toMatchObject({ status: "failed" }); + expect(await readAgentWakeupRequestsById(receipt.chatActionId!)).toHaveLength(0); + expect((await readHeartbeatRunsByCompanyId(companyId)) + .filter(run => run.contextSnapshot?.taskRetryRequestId === receipt.requestId)).toHaveLength(0); + return; + } + expect(outcome).toMatchObject({ status: "deferred" }); + const [wake] = await readAgentWakeupRequestsById(receipt.chatActionId!); + expect(wake).toMatchObject({ status: "deferred_issue_execution", runId: null, + payload: { taskRetryRequestId: receipt.requestId, retryOfRunId: context.runId } }); + const [intent] = await readAgentWakeupRequestsById(receipt.requestId!); + expect(intent).toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect((await readHeartbeatRunsByCompanyId(companyId)) + .filter(run => run.contextSnapshot?.taskRetryRequestId === receipt.requestId)).toHaveLength(0); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)))[0]) + .toMatchObject({ status: "pending_cleanup", cleanupStatus: "failed", metadata: { pendingCleanupInFlight: false } }); + const retryContext = context.wakeup.mock.calls.at(-1)![1].contextSnapshot!; + for (const phase of ["promotion", "execution"] as const) { + await expect(authorizeFailedChatRunRetryWake(db, db, { phase, + wakeupRequestId: receipt.chatActionId!, companyId, issueId, + agentId: context.fixture.assignedAgentId, contextSnapshot: retryContext, + })).rejects.toMatchObject({ details: { code: "chat_failed_run_retry_not_authorized" } }); + } + const heartbeat = heartbeatService(db); + await heartbeat.resumeQueuedRuns(); + await context.service.processFailedChatRunRetries(); + expect((await readAgentWakeupRequestsById(receipt.chatActionId!))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + // Model the recovery worker's validated absence receipt; this test + // exercises queue continuation, not the provider's absence signature. + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); + await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date(), + failureReason: "workspace_source_unavailable", metadata: { ...lease.metadata, + remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }), + } }).where(eq(environmentLeases.id, leaseId)); + await db.update(issueRecoveryActions).set({ evidence: { ...recoveryAction.evidence, + systemRecovery: { ...recoveryAction.evidence.systemRecovery as Record, state: "exhausted", + failure: { code: "source_unavailable", retryClass: "service_required", evidenceFingerprint: "missing-source" } }, + } }).where(eq(issueRecoveryActions.id, recoveryAction.id)); + if (gate === "foreign_source_hold") { + const foreignRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: foreignRunId, companyId, issueId, nativeIssueId: issueId, + agentId: context.fixture.assignedAgentId, runtimeMode: "native", nativePhase: "terminal_failure", + status: "failed", createdAt: new Date(Date.now() - 120_000), finishedAt: new Date(Date.now() - 60_000), + resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { + schema: "paperclip.workspace-restore-recovery.v1", runId: foreignRunId, leaseIds: [] } }, + }); + await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", status: "resolved", ownerType: "system", + cause: "native_session_retry_exhausted", fingerprint: randomUUID(), nextAction: "An independent source still needs recovery.", + evidence: { runId: foreignRunId, automaticRecovery: { replay: "blocked" } }, + }); + } + if (gate === "paused_after_wait") await db.update(chatEndpoints).set({ status: "paused" }) + .where(eq(chatEndpoints.id, context.endpoint.id)); + if (gate === "paused_agent_after_wait") await db.update(agents).set({ status: "paused" }) + .where(eq(agents.id, context.fixture.assignedAgentId)); + if (gate === "edited_after_wait") await db.update(issueComments).set({ body: "Changed after waiting", updatedAt: new Date() }) + .where(eq(issueComments.id, String(context.action.payload.commentId))); + if (gate === "superseded_after_wait") { + const replacement = await requestTaskRetry({ db, companyId, issueId, actorId: "owner-user", + actorSource: "cloud_tenant", observedRunId: context.runId, idempotencyKey: randomUUID(), + prepareChatRetry: context.service.prepareFailedChatRunRetry }); + expect(replacement.requestId).not.toBe(receipt.requestId); + // Reconsider only the old action, including a delayed worker after + // supersession. The replacement remains owned by its own request. + await context.service.processFailedChatRunRetry(receipt.chatActionId!); + } else { + await heartbeat.resumeQueuedRuns(); + await context.service.processFailedChatRunRetries(); + } + if (!canResume) { + if (gate === "foreign_source_hold") { + expect((await readAgentWakeupRequestsById(receipt.chatActionId!))[0]).toMatchObject({ + status: "deferred_issue_execution", payload: { executionWait: { reason: "workspace_repair_required" } }, + }); + } else if (gate !== "superseded_after_wait") { + await heartbeat.resumeQueuedRuns(); + expect((await db.select().from(chatActions).where(eq(chatActions.id, receipt.chatActionId!)))[0]) + .toMatchObject({ status: "failed" }); + expect((await readAgentWakeupRequestsById(receipt.requestId!))[0]).toMatchObject({ status: "failed" }); + } + expect((await readAgentWakeupRequestsById(receipt.chatActionId!))[0]).toMatchObject({ runId: null }); + expect((await readHeartbeatRunsByCompanyId(companyId)) + .filter(run => run.contextSnapshot?.taskRetryRequestId === receipt.requestId)).toHaveLength(0); + return; + } + const [admitted] = await readAgentWakeupRequestsById(receipt.chatActionId!); + expect(admitted).toMatchObject({ status: "queued" }); + expect(admitted.runId).toBeTruthy(); + const [successor] = await readHeartbeatRunsById(admitted.runId!); + expect(successor).toMatchObject({ status: "queued", retryOfRunId: context.runId, + contextSnapshot: { wakeCommentIds: [context.action.payload.commentId], taskRetryRequestId: receipt.requestId } }); + expect(await readFreshWorkspaceReadiness(db, successor)).toMatchObject({ + mode: "fresh_isolated_workspace", retryIntentId: receipt.requestId, sourceRunId: context.runId, + retainedSourceIds: [leaseId], + }); + await heartbeat.resumeQueuedRuns(); + await context.service.processFailedChatRunRetries(); + expect((await readHeartbeatRunsByCompanyId(companyId)) + .filter(run => run.contextSnapshot?.taskRetryRequestId === receipt.requestId)).toHaveLength(1); + + } finally { + await context.service.shutdown(); + // Remove this case's unadmitted synthetic Retry receipts. Global + // notice scans must not consume another case's injected read failure. + await db.delete(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, context.fixture.companyId), + inArray(agentWakeupRequests.reason, ["task_retry", "retry_failed_run"]), + isNull(agentWakeupRequests.runId), + )); + } + }, + ); + it.each(["stopped", "real_queue", "unknown_executor", "source_edited", "cancelled_stopped", "cancelled_unknown"] as const)( "admits ordinary chat task Retry with a missing checkpoint only after containment: %s", async kind => { const context = await failedChatRetryFixture("telegram", undefined, kind === "real_queue"); diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index e873bb16cca..2b5d2c9b026 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -1,5 +1,5 @@ import { continuingTeamsFilePublication, type ContinuingTeamsFileTransfer } from "./microsoft-teams/file-transfer-source.js"; -import { readTaskChatRetryIntent, taskChatRetryExecutionStopped } from "./task-retry-authority.js"; +import { readTaskChatRetryIntent, taskChatRetryCanAwaitRecovery, taskChatRetryExecutionStopped } from "./task-retry-authority.js"; import { notifyChatActionWork, notifyChatPublicationWork } from "../chat-work-notifications.js"; import { FAILED_RUN_RETRY_STALE_MS, dueChatAction } from "../chat-action-work.js"; @@ -549,6 +549,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti source: Pick, companyId: string, taskRetryIntentId?: string, + allowPendingRecovery = false, ) { const [run] = await tx .select() @@ -566,6 +567,9 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti if (!await readTaskChatRetryIntent(tx as unknown as Db, { companyId, issueId: source.issueId, agentId: source.agentId, sourceRunId: run.id, intentId: taskRetryIntentId })) throw failedChatRetryDenied(); if (await taskChatRetryExecutionStopped(tx as unknown as Db, run)) return; + if (allowPendingRecovery && await taskChatRetryCanAwaitRecovery( + tx as unknown as Db, run, source.issueId, taskRetryIntentId, + )) return; // A still-live/unknown executor remains held. The exact pre-provider // proof below can establish a never-started native attempt. if (run.runtimeMode !== "native") throw failedChatRetryDenied(); @@ -1437,7 +1441,8 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti retryActionId: input.wakeupRequestId, }); await assertFailedNativeRetryState(tx, source, input.companyId, - typeof source.intent.payload.taskRetryIntentId === "string" ? source.intent.payload.taskRetryIntentId : undefined); + typeof source.intent.payload.taskRetryIntentId === "string" ? source.intent.payload.taskRetryIntentId : undefined, + input.phase === "admission"); const context = input.contextSnapshot; const hint = context.chatFailedRunRetry as | Record @@ -1587,10 +1592,13 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti authorizeFailedChatRunRetry(tx, scope), ); const existing = await db - .select({ id: agentWakeupRequests.id }) + .select() .from(agentWakeupRequests) .where(eq(agentWakeupRequests.id, claimed.id)); - if (!existing.length) { + const resumeTaskRetry = typeof claimed.payload.taskRetryIntentId === "string" && + existing[0]?.status === "deferred_issue_execution" && !existing[0].runId && + existing[0].payload?.taskRetryRequestId === claimed.payload.taskRetryIntentId; + if (!existing.length || resumeTaskRetry) { const context = { issueId: source.issueId, ...(typeof claimed.payload.taskRetryIntentId === "string" ? { @@ -1641,6 +1649,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti current, claimed.companyId, typeof claimed.payload.taskRetryIntentId === "string" ? claimed.payload.taskRetryIntentId : undefined, + true, // The durable wake callback validates admission; later phases recheck containment. ); }, }), @@ -1662,11 +1671,6 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti .select() .from(agentWakeupRequests) .where(eq(agentWakeupRequests.id, claimed.id)); - if (receipt) - return settle("processed", { - code: "failed_chat_retry_durable", - attemptCount, - }); const denied = error && typeof error === "object" && @@ -1678,6 +1682,24 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti "chat_failed_run_retry_not_authorized", "chat_action_authorization_changed", ].includes(String(error.details.code)); + if (receipt) { + if (denied && typeof claimed.payload.taskRetryIntentId === "string" && + receipt.status === "deferred_issue_execution" && !receipt.runId && + receipt.payload?.taskRetryRequestId === claimed.payload.taskRetryIntentId) { + // Receipt existence proves the earlier wait, not current authority. + // Retire only this unadmitted wake; an admitted run keeps its history. + await db.update(agentWakeupRequests).set({ + status: "failed", error: "chat_failed_run_retry_not_authorized", + finishedAt: new Date(), updatedAt: new Date(), + }).where(and(eq(agentWakeupRequests.id, receipt.id), + eq(agentWakeupRequests.companyId, claimed.companyId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(agentWakeupRequests.runId))); + return settle("failed", { code: "failed_chat_retry_not_authorized", attemptCount }); + } + return settle(["skipped", "failed", "cancelled"].includes(receipt.status) ? "failed" : "processed", { + code: "failed_chat_retry_durable", attemptCount, + }); + } const retryable = !denied && attemptCount < 8; return settle(retryable ? "issued" : "failed", { code: retryable diff --git a/server/src/services/chat-channels/task-retry-authority.ts b/server/src/services/chat-channels/task-retry-authority.ts index f0bcc91afc8..340f2a67b54 100644 --- a/server/src/services/chat-channels/task-retry-authority.ts +++ b/server/src/services/chat-channels/task-retry-authority.ts @@ -1,9 +1,11 @@ import { and, eq, sql } from "drizzle-orm"; -import { agentWakeupRequests, environmentLeases, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { agentWakeupRequests, environmentLeases, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db"; 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 { readSystemRecovery } from "../system-recovery-state.js"; /** This receipt changes the workspace strategy, never the external principal, * accepted inbound batch, publication scope, or current conversation access. */ @@ -43,3 +45,25 @@ export async function taskChatRetryExecutionStopped(db: Db, run: typeof heartbea await hasNativeLocalProcessStop(db, run.companyId, run.id) || await hasHistoricalSuspendedNativeSession(db, run); } + +/** Admission may persist a wait for the exact recovery this user requested. + * This grants no execution authority: promotion and execution still require + * containment, and the queue retains the source's current execution blocker. */ +export async function taskChatRetryCanAwaitRecovery(db: Db, run: typeof heartbeatRuns.$inferSelect, + issueId: string, intentId: string) { + const retry = await readTaskChatRetryIntent(db, { companyId: run.companyId, issueId, + agentId: run.agentId, sourceRunId: run.id, intentId }); + if (!retry || retry.wake.status !== "deferred_issue_execution" || retry.wake.runId) return false; + const blocker = await getExecutionBlocker(db, run.companyId, issueId); + if (blocker?.runId !== run.id || !blocker.recoveryActionId) return false; + const [action] = await db.select({ evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId), + eq(issueRecoveryActions.id, blocker.recoveryActionId), + )).for("share", { noWait: true }); + const recovery = readSystemRecovery(action?.evidence); + // Attaching Retry to active automatic recovery preserves that recovery's + // generation and budget. Its current intentId binds this user request; the + // Retry receipt's generation is a different identity. + return Boolean(recovery && recovery.sourceRunId === run.id && recovery.intentId === intentId && + ["scheduled", "running", "awaiting_evidence"].includes(recovery.state)); +} diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index a33d421a24f..109efade684 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -194,8 +194,19 @@ 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") + ? actions.find(candidate => { + const recovery = readSystemRecovery(candidate.evidence); + return candidate.evidence.runId === sourceRunId && recovery?.sourceRunId === sourceRunId && + recovery.intentId === input.taskRetryRequestId; + })?.evidence ?? action.evidence + : action.evidence; const freshIntent = await canSelectFreshWorkspace({ db, companyId, issueId, agentId, actorId, - sourceRunId: run.id, evidence: action.evidence, taskRetryRequestId: input.taskRetryRequestId, + sourceRunId: run.id, evidence: strategyEvidence, taskRetryRequestId: input.taskRetryRequestId, queuedRequestId: input.queuedCommentRequestId }); const legacyUserTurn = run.runtimeMode === "legacy" && action.cause === "legacy_execution_requires_reconciliation" && diff --git a/server/src/services/fresh-workspace-continuation.ts b/server/src/services/fresh-workspace-continuation.ts index 9cb59b0a9ad..f37754acf9a 100644 --- a/server/src/services/fresh-workspace-continuation.ts +++ b/server/src/services/fresh-workspace-continuation.ts @@ -56,6 +56,10 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick>'{explicitUserContinuation,runId}' = ${run.id}`, sql`${issueRecoveryActions.evidence}#>>'{explicitUserContinuation,readiness,mode}' = 'fresh_isolated_workspace'`, + // Historical sibling holds also record the continuation, but only the + // recovery generation owns its committed fresh attempt. Select that proof. + sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,freshAttemptRunId}' = ${run.id}`, + sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,sourceRunId}' = ${issueRecoveryActions.evidence}#>>'{explicitUserContinuation,readiness,sourceRunId}'`, sql`${issueRecoveryActions.evidence}->>'runId' = ${issueRecoveryActions.evidence}#>>'{explicitUserContinuation,readiness,sourceRunId}'`, )).limit(1); if (!action) { @@ -113,8 +117,9 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick'queuedCommentInterrupt'->>'actorId' = ${opts.requestedByActorId ?? ""}` : opts.queuedCommentRequestId === executionWaitRequestId ? undefined @@ -2812,7 +2823,12 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende agent.companyId, issueId, pending.payload)))) { return { kind: "deferred" as const }; } - if (opts.taskRetryRequestId) { + if (opts.taskRetryRequestId && durableRequest?.failedRunRetry) { + if (pending.id !== durableRequest.id || pending.payload?.taskRetryRequestId !== opts.taskRetryRequestId) + return { kind: "deferred" as const }; + // The parent board Retry is checked separately below after the + // connected conversation reauthorizes this exact child receipt. + } else if (opts.taskRetryRequestId) { const retry = readTaskRetryIntent(pending.payload); if (pending.id !== opts.taskRetryRequestId || retry?.generationId !== pending.id || retry.chatActionId || retry.actorId !== opts.requestedByActorId || retry.sourceRunId !== enrichedContextSnapshot.previousRunId) @@ -4148,23 +4164,30 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende enrichedContextSnapshot.explicitUserContinuation = explicitContinuation; } - const wakeupRequest = await tx - .insert(agentWakeupRequests) - .values({ - ...durableReceiptFields, - companyId: agent.companyId, - agentId, - source, - triggerDetail, - reason, - payload, - status: "queued", - requestedByActorType: opts.requestedByActorType ?? null, - requestedByActorId: opts.requestedByActorId ?? null, - idempotencyKey: opts.idempotencyKey ?? null, - }) - .returning() - .then((rows) => rows[0]); + const wakeupValues = { + ...durableReceiptFields, + companyId: agent.companyId, + agentId, + source, + triggerDetail, + reason, + payload, + status: "queued" as const, + requestedByActorType: opts.requestedByActorType ?? null, + requestedByActorId: opts.requestedByActorId ?? null, + idempotencyKey: opts.idempotencyKey ?? null, + }; + // A deferred durable chat wake already owns this ID and source. + // Promote that receipt in the same transaction as its successor. + const wakeupRequest = durableRequest?.failedRunRetry && executionWaitRequestId === durableRequest.id + ? await tx.update(agentWakeupRequests).set({ + payload, status: "queued", error: null, finishedAt: null, updatedAt: new Date(), + }).where(and(eq(agentWakeupRequests.id, durableRequest.id), + eq(agentWakeupRequests.companyId, agent.companyId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(agentWakeupRequests.runId))) + .returning().then(rows => rows[0]) + : await tx.insert(agentWakeupRequests).values(wakeupValues).returning().then(rows => rows[0]); + if (!wakeupRequest) return { kind: "deferred" as const }; // A handoff changes the executor, not the owner of saved user input. // Validate its exact stopped source while the issue row is locked; @@ -4228,7 +4251,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ...queuedCommentIdsFromRunContext(enrichedContextSnapshot), ]), ]; - if (opts.queuedCommentRequestId || opts.taskRetryRequestId || handoffSource) { + if (opts.queuedCommentRequestId || (opts.taskRetryRequestId && !failedChatRetry) || handoffSource) { adoptedCommentIds = await undeliveredLegacyUserCommentIds(tx as unknown as Db, agent.companyId, issueId, agentId, adoptedCommentIds); } diff --git a/server/src/services/heartbeat/workspace-preparation.test.ts b/server/src/services/heartbeat/workspace-preparation.test.ts index b51d8713be5..156555e71a8 100644 --- a/server/src/services/heartbeat/workspace-preparation.test.ts +++ b/server/src/services/heartbeat/workspace-preparation.test.ts @@ -1,4 +1,5 @@ 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 { tmpdir } from "node:os"; import { join } from "node:path"; @@ -6,7 +7,7 @@ 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 { agents, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, environmentLeases, executionWorkspaces, executionWorkspaceRepositoryAccess, heartbeatRuns, issues, issueAccessGrants, plugins, projects, projectWorkspaces, type Db } from "@paperclipai/db"; +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 { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; @@ -21,6 +22,8 @@ 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 { readFreshWorkspaceReadiness } from "../fresh-workspace-continuation.js"; +import { createSystemRecovery } from "../system-recovery-state.js"; import { isolatedRemoteSourcePath } from "../isolated-remote-workspace-source.js"; import { cleanupExecutionWorkspaceArtifacts } from "../workspace-runtime.js"; import { buildProjectMentionHref } from "@paperclipai/shared"; @@ -262,6 +265,45 @@ describe.skipIf(!support.supported)("heartbeat workspace preparation boundary", expect((await readIssue(issue.id)).executionWorkspaceId).toBe(replacement.persistedExecutionWorkspace!.id); }, 30_000); + it.each(["historical_first", "recovery_first"] as const)( + "prepares the admitted fresh directory with sibling holds in either order: %s", async order => { + const { input, source, issue } = await fixture(true); + const original = await prepareHeartbeatWorkspace(db, input); + await writeFile(join(original.executionWorkspace.cwd, "unpublished.txt"), "keep the old work\n"); + const sourceRunId = input.run.id, retryIntentId = randomUUID(); + const [nextRun] = await db.insert(heartbeatRuns).values({ companyId: input.run.companyId, + agentId: input.agent.id, issueId: issue.id, scopeKind: "issue", status: "running", runtimeMode: "legacy", + contextSnapshot: { issueId: issue.id, taskRetryRequestId: retryIntentId } }).returning(); + await db.insert(agentWakeupRequests).values({ id: retryIntentId, companyId: input.run.companyId, + agentId: input.agent.id, reason: "task_retry", source: "on_demand", status: "coalesced", runId: nextRun.id, + payload: { issueId: issue.id } }); + const readiness = { mode: "fresh_isolated_workspace" as const, retryIntentId, sourceRunId, retainedSourceIds: [] }; + const evidence = { runId: sourceRunId, explicitUserContinuation: { runId: nextRun.id, readiness } }; + const hold = { companyId: input.run.companyId, sourceIssueId: issue.id, kind: "active_run_watchdog", + status: "resolved", ownerType: "system", nextAction: "Preserve the original source outcome." }; + const historical = { ...hold, cause: "native_session_retry_exhausted", fingerprint: randomUUID(), evidence }; + const recovery = { ...hold, cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(), + evidence: { ...evidence, systemRecovery: { ...createSystemRecovery(sourceRunId, retryIntentId), + state: "exhausted", freshAttemptRunId: nextRun.id } } }; + await db.insert(issueRecoveryActions).values(order === "historical_first" ? [historical, recovery] : [recovery, historical]); + // Use the same committed-receipt lookup as run preparation, not a + // hand-built freshWorkspace input. The historical hold has no generation. + input.task.freshWorkspace = await readFreshWorkspaceReadiness(db, nextRun); + expect(input.task.freshWorkspace).toEqual(readiness); + input.run = nextRun; + input.task.issueRef = { ...issue, executionWorkspaceId: null, executionWorkspacePreference: "create_new" }; + const replacement = await prepareHeartbeatWorkspace(db, input); + expect(replacement.executionWorkspace.cwd).not.toBe(original.executionWorkspace.cwd); + expect(replacement.executionWorkspace.cwd).not.toBe(source); + expect(await readFile(join(replacement.executionWorkspace.cwd, "README.md"), "utf8")).toBe("source content\n"); + await expect(access(join(replacement.executionWorkspace.cwd, "unpublished.txt"))).rejects.toThrow(); + expect(await readFile(join(original.executionWorkspace.cwd, "unpublished.txt"), "utf8")).toBe("keep the old work\n"); + const common = await execFile("git", ["-C", replacement.executionWorkspace.cwd, "rev-parse", "--path-format=absolute", "--git-common-dir"]); + expect(common.stdout.trim()).toBe(join(replacement.executionWorkspace.cwd, ".git")); + expect((await readIssue(issue.id)).executionWorkspaceId).toBe(replacement.persistedExecutionWorkspace!.id); + }, 30_000, + ); + it("starts a fresh clean task directory without treating it as a Git source", async () => { const { input, source, issue } = await fixture(); input.policy.useIsolatedTaskDirectory = true; diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index a443027f61d..1937712ac23 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -7,6 +7,7 @@ import { persistActivity } from "./activity-log.js"; import { createSystemRecovery, readSystemRecovery } from "./system-recovery-state.js"; import { updateSystemRecovery } from "./system-recovery.js"; import { instanceSettingsService } from "./instance-settings.js"; +import { notifyChatActionWork } from "./chat-work-notifications.js"; export async function assertTaskRetryConversationAllowed( db: Db, @@ -182,6 +183,19 @@ export async function reconcileTaskChatRetryReceipts(db: Db) { const [receipt] = chat ? await db.select().from(agentWakeupRequests).where(and( eq(agentWakeupRequests.companyId, wake.companyId), eq(agentWakeupRequests.id, chat.id), eq(agentWakeupRequests.agentId, wake.agentId))) : []; + // A deferred chat receipt is admitted input, not a completed dispatch. + // Re-stage that exact action through chat's owner; it rechecks authority + // and normal queue policy before updating the existing receipt. + if (chat?.status === "processed" && chat.payload.taskRetryIntentId === wake.id && + chat.payload.failedRunId === intent.sourceRunId && receipt?.status === "deferred_issue_execution" && + !receipt.runId && receipt.payload?.taskRetryRequestId === wake.id) { + await db.transaction(async tx => { + await notifyChatActionWork(tx, "failed_run_retry"); + await tx.update(chatActions).set({ status: "issued", updatedAt: new Date() }) + .where(and(eq(chatActions.id, chat.id), eq(chatActions.companyId, wake.companyId), + eq(chatActions.status, "processed"), eq(chatActions.updatedAt, chat.updatedAt))); + }); + } const failed = chat && ["failed", "cancelled"].includes(chat.status); await db.update(agentWakeupRequests).set({ ...(receipt?.runId ? { status: "coalesced", runId: receipt.runId, finishedAt: new Date() }