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" });