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() }