From 06c0566a087cbb04972f29b01b099603949a02fb Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 09:03:54 -0600 Subject: [PATCH] fix: make task Retry work after a refused automatic attempt (#19466) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Task Retry stores a durable request before it starts a new run. > - An automatic attempt can stop at the execution gate before a provider starts. > - That cancelled attempt becomes the source of the next Retry click. > - Admission rejected that source, or chose an older run that did not match the stored Retry authority. > - This pull request verifies the refused attempt and keeps it as the exact authority source. > - The user can start one new turn after the older execution holds pass their existing checks. ## Linked Issues or Issue Description **What happened?** Retry can remain deferred when the latest run was cancelled with `execution_reconciliation_required` before dispatch. An older failed preparation has an execution recovery hold. The Retry receipt names the cancelled child, but admission only accepts a failed preparation as an extra source. It can also choose the older run as `previousRunId`, which fails the exact source check at dispatch. **Expected behavior** Retry starts one new turn when the latest attempt has a verified pre-dispatch refusal and every older execution hold is ready. Provider activity, retained execution resources, missing receipts, or a live earlier process keep the hold. **Steps to reproduce** 1. Seed a failed native preparation and its recovery hold. 2. Let a later automatic attempt be cancelled by the stale queued execution gate before dispatch. 3. Click Retry against that latest cancelled attempt. 4. Before this change, continuation admission returns no receipt. After this change, one successor starts with authority bound to the clicked attempt. **Paperclip version or commit** Reproduced on `2f35eb1080`. Rebased onto `6cafcaf89b` for submission. **Deployment mode** Local source checkout with embedded PostgreSQL and the native Runner on macOS arm64. Refs #15383 and #16939. Those changes cover legacy conversation cancellation and queue promotion. This change covers the exact source of a task Retry receipt after a gate refused its automatic child. ## What Changed - Reuse the existing database-backed pre-dispatch refusal proof when admitting the exact Retry source. - Keep the clicked source as `previousRunId`. Continue to check every older execution hold. - Add database regressions for safe refusal, unsafe evidence, concurrent dispatch, and duplicate clicks. - Add an opt-in acceptance test that drives the authenticated Retry route through a real Codex turn and task finalization. - Document the admission contract. ## Verification - TDD: the new database regression returned `null` on the original code. It passes with the repair. - Final head `c43a84b1bf`: 487 tests pass across task Retry, explicit native continuation, and execution continuation. Server TypeScript and module boundary checks pass. - Live acceptance: `PAPERCLIP_LIVE_REFUSED_RETRY=true PAPERCLIP_RUNNER_BINARY= pnpm --dir server exec vitest run --config vitest.config.ts src/services/native-runtime/refused-task-retry.live.test.ts`. - The real HTTP request returned 202 on the final rebased head `c43a84b1bf`. Codex `gpt-6.1-sol` with high effort produced the requested file. The task became done, both task locks cleared, and finalization committed. A repeated click reused the same receipt and did not create another run. The source file stayed unchanged. - Server TypeScript check and module boundary check. - `pnpm verify --explain` returned 2 because the execution contract doc requires broad CI coverage. The required CI checks supply that coverage. ## Risks This changes execution admission. The existing proof requires the stored refusal receipt and no provider work or retained execution resources. Earlier recovery holds, process stop checks, decisions, actor authority, and task ownership still apply. No schema change or migration. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, tool use, and code execution. The host does not expose an exact author model ID or context window. The live acceptance provider was `gpt-6.1-sol` with high reasoning effort. ## 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 — all GitHub Actions checks pass on `c43a84b1bf`; Paperclip Review completed at 5/5 with no findings (neutral check conclusion). - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups — reviewed `c43a84b1bf` with no actionable findings - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- doc/execution-semantics.md | 2 + .../services/explicit-native-continuation.ts | 10 +- .../refused-task-retry.live.test.ts | 116 ++++++++++++++++++ .../services/pre-dispatch-execution-wait.ts | 2 +- server/src/services/task-retry.test.ts | 90 +++++++++++++- 5 files changed, 214 insertions(+), 6 deletions(-) create mode 100644 server/src/services/native-runtime/refused-task-retry.live.test.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 9d9321e017f..69ca5f0d3f3 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1256,6 +1256,8 @@ A new user message can continue a terminal native run whose process fields were The task thread exposes the guarded Retry action for failed or timed-out conversation runs and native preparation cancelled before provider startup when the never-started proof is verified. The server projects this eligibility on the recovery notice and rechecks it on Retry. Cleanup quarantine retains its inspection path, and non-conversation reconciliation gates remain enforced. Pending decisions, active execution, pause, budget, dependency, and ownership gates remain in force. A refused Retry reports its reason inline and remains available for another attempt. When a newer linked-human GitHub message was saved but its admission was skipped by a budget or agent-availability gate, ordinary task Retry selects that unstarted message before considering the older failed run. The existing chat task-Retry owner snapshots the complete pending batch, including eligible skipped input, into a new admission generation bound to the canonical processed deliveries, comments, principal, session generation, admitted instruction snapshots, and task Retry intent. It retains the original skipped receipt and failed result unchanged. A terminal failed publication of that old result also remains history for a distinct selected request; pending or uncertain publication still holds work, and the old output is never replayed. Board Retry refusals expose only stable allowlisted authority categories, never source payloads, identifiers, or credential fingerprints. Admission rechecks source integrity and current access, then uses the normal new-user-turn continuation proof: a committed failed result can remain history only after controller, successor, process, workspace, and exact remote-termination guards pass. Unknown ownership stays deferred; Retry cannot certify it. The existing chat retry queue owns delivery and bounded infrastructure retries; a committed receipt settles a lost acknowledgment without another execution. An accepted saved-input receipt discharges only its matching unadmitted-input gate, allowing later genuine conversation or tree-resume wakes without replaying the skipped request. +When the latest attempt was refused before dispatch, Retry keeps that exact attempt as its authority source. Admission verifies its stored refusal receipt and the absence of provider execution, sessions, leases, and output. Every earlier execution hold must also pass its normal stop and recovery checks. A refused attempt cannot replace that evidence or authorize replay of the earlier turn. + A deliberate Retry also binds an already completed file-preservation record to that intent. Other recovery holds for the same source run can then use its saved snapshot when selecting a fresh workspace. This binding preserves the recovery diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 9f6a96545ef..a7ccbeeb638 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -4,6 +4,7 @@ import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { canSelectFreshWorkspace, type ContinuationReadiness } from "./fresh-workspace-continuation.js"; import { verifiedWorkspaceSnapshotReceipt } from "./workspace-recovery-snapshot.js"; import { readTaskRetryIntent } from "./task-retry.js"; +import { isPreDispatchExecutionWaitVerified } from "./pre-dispatch-execution-wait.js"; import { readSystemRecovery } from "./system-recovery-state.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; @@ -269,7 +270,8 @@ export async function admitExplicitNativeContinuation(input: { eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, taskRetry.sourceRunId), eq(heartbeatRuns.agentId, agentId), )).for("update"); if (!source || (source.nativeIssueId ?? source.issueId ?? source.contextSnapshot?.issueId) !== issueId - || adapterExecutionControls.has(source.id) || !await isFailedNativePreparation(db, source, coordinator)) return null; + || adapterExecutionControls.has(source.id) || + !(await isFailedNativePreparation(db, source, coordinator) || await isPreDispatchExecutionWaitVerified(db, source))) return null; sourceChecks.push({ cause: "legacy_execution_requires_reconciliation", evidence: { runId: source.id } }); } if (!sourceChecks.length) return null; @@ -317,7 +319,8 @@ export async function admitExplicitNativeContinuation(input: { const failedPreparation = await isFailedNativePreparation(db, run, coordinator); const unusedAdmission = run.status === "cancelled" && !run.startedAt && run.errorCode === "execution_reconciliation_required" && - !run.processPid && !run.processGroupId && !run.nativeSessionId; + !run.processPid && !run.processGroupId && !run.nativeSessionId && + (!taskRetry || await isPreDispatchExecutionWaitVerified(db, run)); const sourceRunId = run.id; const strategyRequestId = input.taskRetryRequestId ?? input.queuedCommentRequestId ?? savedChatRequestId; const strategyEvidence = strategyRequestId && !Object.hasOwn(action.evidence, "systemRecovery") @@ -492,7 +495,8 @@ export async function admitExplicitNativeContinuation(input: { ne(heartbeatRuns.id, input.successorRunId), )).limit(1); if (active) return blocked("execution_active", "Waiting for the current run. Your message is saved."); - const previous = executedSources.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime())[0]!; + const previous = taskRetry ? sources.find(run => run.id === taskRetry.sourceRunId)! + : executedSources.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime())[0]!; // Prove required task history is available before retiring any hold. await buildExecutionContinuation({ db, companyId, issueId, agentId, context: { previousRunId: previous.id, wakeCommentId: commentId }, diff --git a/server/src/services/native-runtime/refused-task-retry.live.test.ts b/server/src/services/native-runtime/refused-task-retry.live.test.ts new file mode 100644 index 00000000000..db131e057ee --- /dev/null +++ b/server/src/services/native-runtime/refused-task-retry.live.test.ts @@ -0,0 +1,116 @@ +import { randomUUID } from "node:crypto"; +import { copyFile, mkdir, readFile, writeFile } from "node:fs/promises"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import { and, eq } from "drizzle-orm"; +import { agents, agentWakeupRequests, companies, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { expect, it } from "vitest"; +import { startRunnerApiTestServer } from "../../__tests__/helpers/runner-api-server.js"; +import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js"; +import { heartbeatService } from "../heartbeat.js"; +import { instanceSettingsService } from "../instance-settings.js"; +import { closeIdleWarmNativeSessionsForRestart } from "./native-session-executor.js"; + +it.skipIf(process.env.PAPERCLIP_LIVE_REFUSED_RETRY !== "true")("completes work through HTTP Retry after a refused automatic attempt", async () => { + const home = createIsolatedTestHome(); + const keys = ["PAPERCLIP_HOME", "CODEX_HOME", "PAPERCLIP_API_URL", "PAPERCLIP_AGENT_JWT_SECRET", "PAPERCLIP_TELEMETRY_DISABLED", "PAPERCLIP_RUNNER_STATE_DIR"]; + const saved = Object.fromEntries(keys.map(key => [key, process.env[key]])); + let server: Awaited> | undefined; + let heartbeat: ReturnType | undefined; + let companyId: string | undefined; + try { + const codexHome = join(home.home, "codex"); + await mkdir(codexHome, { mode: 0o700 }); + await copyFile(join(process.env.PAPERCLIP_LIVE_CODEX_HOME ?? join(homedir(), ".codex"), "auth.json"), join(codexHome, "auth.json")); + process.env.CODEX_HOME = codexHome; + process.env.PAPERCLIP_HOME = join(home.home, "instance"); + process.env.PAPERCLIP_RUNNER_STATE_DIR = join(home.home, "runner-state"); + process.env.PAPERCLIP_AGENT_JWT_SECRET = randomUUID(); + process.env.PAPERCLIP_TELEMETRY_DISABLED = "1"; + server = await startRunnerApiTestServer({ deploymentMode: "authenticated", humanSessions: true }); + process.env.PAPERCLIP_API_URL = server.apiUrl; + const f = await server.fixture({ connectionScenario: "fresh" }), db = server.db; + companyId = f.companyId; + heartbeat = heartbeatService(db); + const preserved = randomUUID(), appended = randomUUID(); + await writeFile(join(f.workspace, "retry-proof.txt"), `${preserved}\n`); + await db.update(companies).set({ defaultResponsibleUserId: f.userId }).where(eq(companies.id, f.companyId)); + await db.update(agents).set({ status: "idle", adapterConfig: { provider: "codex", model: "gpt-6.1-sol", modelReasoningEffort: "high", + cwd: f.workspace, workspaceStrategy: { type: "project_primary" }, env: { CODEX_HOME: codexHome }, timeoutSec: 180, lifecycleMode: "per_turn" }, + runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true, maxConcurrentRuns: 1 } }, + }).where(eq(agents.id, f.agentId)); + await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true, enableIsolatedWorkspaces: true }); + await db.update(issues).set({ status: "blocked", executionRunId: null, checkoutRunId: null, responsibleUserId: f.userId, + executionWorkspaceSettings: { mode: "shared_workspace" }, title: "Complete one Retry turn", + description: `Write exactly ${appended} and a newline once to retry-result.txt in the current workspace. Read the complete file. This is the only requested work. Finish with paperclip_finish, reportedWorkDisposition done. No external writes are needed.`, + }).where(eq(issues.id, f.issueId)); + const [workspace] = await db.insert(executionWorkspaces).values({ companyId: f.companyId, projectId: f.projectId, + projectWorkspaceId: f.projectWorkspaceId, name: "Saved task workspace", mode: "shared_workspace", + strategyType: "project_primary", cwd: f.workspace }).returning(); + await db.update(issues).set({ executionWorkspaceId: workspace.id, executionWorkspacePreference: "reuse_existing" }) + .where(eq(issues.id, f.issueId)); + const now = Date.now(); + await db.update(heartbeatRuns).set({ status: "failed", runtimeMode: "legacy", nativeIssueId: null, + issueId: f.issueId, errorCode: "setup_failed", executionStage: "preparing", createdAt: new Date(now - 3000), + finishedAt: new Date(now - 2000), runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).where(eq(heartbeatRuns.id, f.runId)); + const [hold] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: f.runId, + status: "resolved", outcome: "blocked", nextAction: "Continue after recovery.", + evidence: { runId: f.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }).returning(); + const [refused] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + issueId: f.issueId, status: "cancelled", errorCode: "execution_reconciliation_required", retryOfRunId: f.runId, + createdAt: new Date(now - 1000), finishedAt: new Date(now - 500), contextSnapshot: { issueId: f.issueId, retryOfRunId: f.runId }, + resultJson: { stopReason: "execution_reconciliation_required", timeoutSource: "stale_queued_run_gate", + executionWait: { issueId: f.issueId, recoveryActionId: hold.id } } }).returning(); + const cookie = await server.humanSession(f.userId!); + const idempotencyKey = randomUUID(); + const retry = () => fetch(`${server!.apiUrl}/api/issues/${f.issueId}/retry`, { method: "POST", + headers: { "content-type": "application/json", cookie, origin: server!.apiUrl }, + body: JSON.stringify({ observedRunId: refused.id, idempotencyKey }) }); + const response = await retry(); + const receipt = await response.json() as { requestId: string; accepted: boolean }; + expect(response.status, JSON.stringify(receipt)).toBe(202); + expect(receipt.accepted).toBe(true); + await expect.poll(async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId)))[0].runId, + { timeout: 30_000, interval: 250 }).toBeTruthy(); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId)); + await expect.poll(async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!)))[0].status, + { timeout: 240_000, interval: 500 }).toBe("succeeded"); + await heartbeat.drainActiveRunExecutions(); + expect(await readFile(join(f.workspace, "retry-proof.txt"), "utf8")).toBe(`${preserved}\n`); + const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!)); + expect(successor.contextSnapshot).toMatchObject({ previousRunId: refused.id, + explicitUserContinuation: { previousRunId: refused.id, taskRetryRequestId: receipt.requestId } }); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(task).toMatchObject({ status: "done", executionRunId: null, checkoutRunId: null }); + const [completedWorkspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, task.executionWorkspaceId!)); + expect(await readFile(join(completedWorkspace.cwd!, "retry-result.txt"), "utf8")).toBe(`${appended}\n`); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, successor.id)); + expect(coordinator.phase).toBe("committed"); + const repeated = await retry(); + expect(await repeated.json()).toMatchObject({ accepted: true, requestId: receipt.requestId }); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.runtimeMode, "native")))) + .toHaveLength(1); + const evidence = { passed: true, platform: `${process.platform}-${process.arch}`, model: "gpt-6.1-sol", effort: "high", + sourceRunId: f.runId, refusedRunId: refused.id, successorRunId: successor.id, sourceFilePreserved: true, outputFileLines: 1, + taskStatus: task.status, locksCleared: true, finalization: coordinator.phase, duplicateClickReused: true }; + if (process.env.PAPERCLIP_LIVE_EVIDENCE_PATH) await writeFile(process.env.PAPERCLIP_LIVE_EVIDENCE_PATH, JSON.stringify(evidence, null, 2) + "\n"); + console.log(JSON.stringify(evidence)); + } catch (error) { + if (server && companyId && process.env.PAPERCLIP_LIVE_EVIDENCE_PATH) { + const runs = await server.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)); + await writeFile(`${process.env.PAPERCLIP_LIVE_EVIDENCE_PATH}.failure.json`, JSON.stringify({ error: String(error), + runs: runs.map(run => ({ id: run.id, status: run.status, error: run.error, result: run.resultJson })), + }, null, 2)); + } + throw error; + } finally { + await heartbeat?.drainActiveRunExecutions(); + await closeIdleWarmNativeSessionsForRestart(); + await server?.close(); + for (const key of keys) { if (saved[key] === undefined) delete process.env[key]; else process.env[key] = saved[key]; } + home.cleanup(); + } +}, 360_000); diff --git a/server/src/services/pre-dispatch-execution-wait.ts b/server/src/services/pre-dispatch-execution-wait.ts index b91bbc878be..3d17320ec49 100644 --- a/server/src/services/pre-dispatch-execution-wait.ts +++ b/server/src/services/pre-dispatch-execution-wait.ts @@ -21,7 +21,7 @@ function preDispatchExecutionWaitPredicate() { AND ${noRetainedExecutionPredicate()}, false)`; } -export async function isPreDispatchExecutionWaitVerified(db: Db, run: typeof heartbeatRuns.$inferSelect): Promise { +export async function isPreDispatchExecutionWaitVerified(db: Db, run: Pick): Promise { if (run.errorCode !== "execution_reconciliation_required" || run.runtimeMode !== "legacy") return false; const [verified] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id), preDispatchExecutionWaitPredicate(), diff --git a/server/src/services/task-retry.test.ts b/server/src/services/task-retry.test.ts index c1a29ddc465..995923a4e77 100644 --- a/server/src/services/task-retry.test.ts +++ b/server/src/services/task-retry.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { agents, agentWakeupRequests, chatConversations, chatEndpoints, companies, createDb, type Db, environmentLeases, environments, executionWorkspaces, heartbeatRuns, issueComments, issues, issueRecoveryActions, nativeRunFinalizations, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db"; +import { agents, agentWakeupRequests, chatConversations, chatEndpoints, companies, createDb, type Db, environmentLeases, environments, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issueComments, issues, issueRecoveryActions, nativeRunFinalizations, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { priorRecoveryApiReceipt, unresolvedRecoveryToolInvocation } from "./recovery-effect-receipts.js"; import { requestTaskRetry } from "./task-retry.js"; @@ -189,6 +189,93 @@ describe("durable task Retry admission", () => { expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.input.issueId))) .toHaveLength(0); }); + async function seedRefusedRetry() { + const input = await seed(); + const now = Date.now(); + const [source] = await db.update(heartbeatRuns).set({ + runtimeMode: "legacy", errorCode: "setup_failed", executionStage: "preparing", + createdAt: new Date(now - 3000), finishedAt: new Date(now - 2000), + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).where(eq(heartbeatRuns.id, input.observedRunId)).returning(); + const [hold] = await db.insert(issueRecoveryActions).values({ companyId: input.companyId, sourceIssueId: input.issueId, + kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: source.id, + status: "resolved", outcome: "blocked", nextAction: "Continue after recovery.", + evidence: { runId: source.id, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }).returning(); + const [refused] = await db.insert(heartbeatRuns).values({ companyId: input.companyId, agentId: source.agentId, + issueId: input.issueId, status: "cancelled", errorCode: "execution_reconciliation_required", + retryOfRunId: source.id, createdAt: new Date(now - 1000), finishedAt: new Date(now - 500), + contextSnapshot: { issueId: input.issueId, retryOfRunId: source.id }, + resultJson: { stopReason: "execution_reconciliation_required", timeoutSource: "stale_queued_run_gate", + executionWait: { issueId: input.issueId, recoveryActionId: hold.id } } }).returning(); + return { input: { ...input, actorSource: "local_implicit", observedRunId: refused.id }, source, refused, hold }; + } + + it.each(["ready", "provider_event", "provider_event_with_hold", "retained_lease", "started", "missing_receipt", "foreign_receipt", "live_ancestor"] as const)( + "admits Retry after a refused automatic attempt only with no provider work: %s", async gate => { + const { input, source, refused, hold } = await seedRefusedRetry(); + if (gate === "provider_event_with_hold") await db.insert(issueRecoveryActions).values({ companyId: input.companyId, sourceIssueId: input.issueId, + kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: refused.id, + status: "resolved", outcome: "blocked", nextAction: "Continue after recovery.", evidence: { runId: refused.id } }); + if (gate === "provider_event" || gate === "provider_event_with_hold") await db.insert(heartbeatRunEvents).values({ companyId: input.companyId, + agentId: source.agentId, runId: refused.id, seq: 1, eventType: "tool.execution.started" }); + if (gate === "retained_lease") await db.insert(environmentLeases).values({ companyId: input.companyId, + issueId: input.issueId, heartbeatRunId: refused.id, provider: "local", status: "released", leasePolicy: "ephemeral", + releasedAt: new Date(), cleanupStatus: "success" }); + if (gate === "started") await db.update(heartbeatRuns).set({ startedAt: refused.createdAt }).where(eq(heartbeatRuns.id, refused.id)); + if (gate === "missing_receipt") await db.update(heartbeatRuns).set({ resultJson: null }).where(eq(heartbeatRuns.id, refused.id)); + if (gate === "foreign_receipt") await db.update(heartbeatRuns).set({ resultJson: { ...refused.resultJson, + executionWait: { issueId: randomUUID(), recoveryActionId: hold.id } } }).where(eq(heartbeatRuns.id, refused.id)); + if (gate === "live_ancestor") await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, source.id)); + const receipt = await requestTaskRetry(input); + expect(receipt).toMatchObject({ accepted: true, observedRunId: refused.id }); + const successorRunId = randomUUID(); + const continuation = await db.transaction(async tx => { + await tx.select().from(issues).where(eq(issues.id, input.issueId)).for("update"); + const continuation = await admitExplicitNativeContinuation({ db: tx as unknown as Db, companyId: input.companyId, + issueId: input.issueId, agentId: source.agentId, actorId: input.actorId, actorType: "user", + reason: "task_retry", commentId: null, successorRunId, taskRetryRequestId: receipt.requestId! }); + if (continuation) { + await tx.insert(heartbeatRuns).values({ id: successorRunId, companyId: input.companyId, agentId: source.agentId, + issueId: input.issueId, status: "queued", contextSnapshot: { issueId: input.issueId, explicitUserContinuation: continuation } }); + await tx.update(agentWakeupRequests).set({ status: "coalesced", runId: successorRunId }).where(eq(agentWakeupRequests.id, receipt.requestId!)); + } + return continuation; + }); + if (gate === "ready") { + expect(continuation).toMatchObject({ previousRunId: refused.id, taskRetryRequestId: receipt.requestId }); + await expect(buildExecutionContinuation({ db, companyId: input.companyId, issueId: input.issueId, + agentId: source.agentId, runId: successorRunId, context: { issueId: input.issueId, explicitUserContinuation: continuation }, + summary: null, exposeLowTrustRaw: false })).resolves.toMatchObject({ interruptedRunId: refused.id }); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, hold.id)))[0].evidence.automaticRecovery) + .toMatchObject({ actionOutcome: "unknown", replay: "explicit_user_continuation" }); + } else { + expect(continuation).toBeNull(); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, hold.id)))[0].evidence.explicitUserContinuation) + .toBeUndefined(); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, successorRunId))).toHaveLength(0); + } + }); + + it("dispatches one Retry continuation after a refused automatic attempt", async () => { + const { input, source, refused } = await seedRefusedRetry(); + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, source.agentId)); + await db.insert(heartbeatRuns).values({ companyId: input.companyId, agentId: source.agentId, status: "running" }); + const receipt = await requestTaskRetry(input); + const duplicate = await requestTaskRetry({ ...input, idempotencyKey: randomUUID() }); + expect(duplicate.requestId).toBe(receipt.requestId); + const heartbeat = heartbeatService(db); + await Promise.all([heartbeat.resumeTaskRetry(input.companyId, receipt.requestId!), + heartbeat.resumeTaskRetry(input.companyId, receipt.requestId!)]); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!)); + const queued = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.status, "queued"))); + expect(queued, JSON.stringify(wake.payload)).toHaveLength(1); + expect(wake).toMatchObject({ status: "coalesced", runId: queued[0].id }); + expect(queued[0].contextSnapshot).toMatchObject({ previousRunId: refused.id, + explicitUserContinuation: { previousRunId: refused.id, taskRetryRequestId: receipt.requestId } }); + await expect(dispatchRun(input, queued[0])).resolves.toMatchObject({ interruptedRunId: refused.id }); + }); + it("serializes concurrent clicks and remembers each key after completion", async () => { const input = await seed(); const second = { ...input, idempotencyKey: randomUUID() }; @@ -403,7 +490,6 @@ describe("durable task Retry admission", () => { const [issue] = await db.select().from(issues).where(eq(issues.id, input.issueId)); await db.update(issues).set({ conversationAgentId: issue.assigneeAgentId, conversationUserId: input.actorId, conversationState: "waiting" }).where(eq(issues.id, input.issueId)); - // Keep admitted runs queued: this test asserts admission, not provider behavior. await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }) .where(eq(agents.id, issue.assigneeAgentId!)); await db.insert(heartbeatRuns).values({ companyId: input.companyId, agentId: issue.assigneeAgentId!, status: "running" });