From 7e8901aa60a3a633ce71c553f09b3606a95ff6ec Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 11 Nov 2026 13:42:36 -0600 Subject: [PATCH] fix(recovery): let Retry and the person's answer continue a held task with a pending decision (#20872) 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 > - A native run on a remote sandbox can lose its Runner process. Recovery then saves the sandbox files and puts an execution hold on the task > - The hold offers Retry. A person can also continue the task with a new message > - Many held tasks also have a pending decision: the question that a parked turn asked, or an older connection request > - Continuation admission refuses every continuation while a decision is pending ("decision_pending"). The answer to that decision is not a message, so admission defers it behind the hold ("execution_recovery") > - So Retry waits for the answer and the answer waits for the hold. The task shows a Retry that does nothing, and nothing continues it > - This pull request lets a person's Retry pass an open question, and lets the person's own answer continue the task through the hold > - The benefit is that a task with a lost Runner and a pending decision always has a control that works ## Linked Issues or Issue Description **What happened?** On staging, the same-run retry of six remote native runs found the Runner process gone (`process_not_running`, lifecycle `ready`). Each run failed with `runner_remote_process_identity_unavailable`. Saved-work recovery then recovered the sandbox files, and the task showed "Saved work recovered. Retry or send a message to continue." with `canRetry: true`. Four of these tasks also had a pending decision: the question that the parked turn asked, or an older pending connection request. For those tasks, Retry is deferred with `decision_pending`, and an answer to the decision is deferred with `execution_recovery`. No control continues the task. **What did you expect to happen?** Retry starts one new run. The person's answer to the pending decision also continues the task. **Steps to reproduce** 1. A remote native run asks the user a question and parks the turn. 2. The Runner process exits. The same-run retry fails with `runner_remote_process_identity_unavailable`. 3. Saved-work recovery recovers the files. The task offers Retry. 4. Click Retry: the Retry waits with "A pending approval or question must be resolved before this message can start." 5. Answer the question: the answer waits with "Waiting for execution recovery. Your message is saved." Refs #20838, #20853 ## What Changed - `admitExplicitNativeContinuation` (`server/src/services/explicit-native-continuation.ts`): a task Retry is no longer held by a plain pending question (`ask_user_questions`) or connection request (`connection_intent`). Any interaction with approval semantics still holds Retry: a plan review (`issue_document` target), a governed tool action, a secret proposal, a connection authorization, and other confirmations. A pending approval still holds every continuation. A new message supersedes a plain question, as on master, and waits for a decision with approval semantics. - `admitExplicitNativeContinuation`: the person's own resolution of an interaction (`mutation: "interaction"`, answered/accepted/rejected, resolved by the same user) is explicit user input. It continues the held task like a new message: a fresh session, the hold resolved as `explicitUserContinuation` with the `interactionId`, and `issue.execution_recovery_settled` activity. It never replays the failed run. Governed tool-action confirmations are excluded. The existing `readQueuedInteractionResponse` reads the answer. - `heartbeat/queue.ts`: both admission calls pass the wake payload to the admission. - `buildExecutionContinuation` (`server/src/services/execution-continuation.ts`): dispatch rechecks the user authorization. It accepted an interaction receipt only through a saved interrupt queue, so the answered successor failed with `continuation_user_authorization_missing`. It now also accepts the person's own answer: the interaction named by the durable authorization was resolved by that user and not by an agent, and the authorization names this successor run. The check does not depend on the run's current trigger, so a later answer that replaces the trigger cannot invalidate it. - New API-level test `server/src/services/governed-wait-runner-loss.test.ts` with the two staging shapes (a parked question from the failed run; an older connection request on a monitor wake). It checks that Retry starts exactly one run and leaves the decision pending, that the person's answer starts exactly one run and settles the hold, and that an answer resolved by a different user keeps the hold. - `doc/execution-semantics.md`: describes the rule. ## Verification - `cd server && pnpm exec vitest run src/services/governed-wait-runner-loss.test.ts`: 5 passed. The answer tests now also build the successor's continuation as dispatch does and check that the agent receives the answer. - Red for dispatch (first head, ac17aebe): both answer tests fail with `continuation_user_authorization_missing`. - Ponytail follow-ups: a two-answer test (the second answer waits behind the successor; dispatch also holds with a replaced trigger) and 'keeps Retry on hold while %s is pending' for four approval-bearing kinds. On head 60beb773, 5 of these fail. Two more tests show that a new message supersedes a plain question and waits for a plan review (both unchanged from master). All 12 tests in the file pass now. The PR adds no code comments; the rationale is in `doc/execution-semantics.md` and here. - Continuation suites (execution-continuation, execution-continuation-provenance, legacy-continuation-authority, explicit-native-continuation, task-retry, governed-wait-runner-loss): 634 passed. - Red on master (product hunks removed, test kept): 4 of 5 fail. Retry tests: `successors: 0`, deferred `decision_pending`. Answer tests: `successors: 0`, deferred `execution_recovery`. Every case asserts `runError: "runner_remote_process_identity_unavailable"`. - Related suites: explicit-native-continuation, task-retry, workspace-recovery-admission, retry-workspace-continuity, workspace-recovery-retry-copyback, workspace-recovery-fallback, native-workspace-reconstruction, native-recovery-preparation-failure, issue-thread-interactions-service. 957 of 959 passed. The 2 failures are 30 s `afterEach` hook timeouts in native-workspace-reconstruction; all 51 of its tests pass alone, and the same file on unchanged master shows 5 hook timeouts on this machine. - `cd server && pnpm exec tsc --noEmit -p tsconfig.json`: passes. ## Risks - Medium. A Retry can now start while a plain question or connection request is pending; decisions with approval semantics still hold it. The new run sees the open decision in its task context, and the decision's answer still wakes the agent later. Approvals keep holding Retry, so approval gates are unchanged. - An answer now continues a held task in a fresh session. All other admission gates still run: stop proof, saved-work recovery, workspace repair, active execution, pause, budget and ownership. Only the user who resolved the interaction can continue this way. - This does not make recovery automatic. A task whose Runner was lost still rests on the hold until a person clicks Retry or answers. Automatic settlement of a parked governed wait is a separate change. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), with tool use and code execution in an agent harness. ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/execution-semantics.md | 13 ++ server/src/services/execution-continuation.ts | 7 +- .../services/explicit-native-continuation.ts | 13 +- .../governed-wait-runner-loss.test.ts | 207 ++++++++++++++++++ server/src/services/heartbeat/queue.ts | 2 + 5 files changed, 239 insertions(+), 3 deletions(-) create mode 100644 server/src/services/governed-wait-runner-loss.test.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index f86cdf04cf0..0101065c965 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1739,6 +1739,19 @@ recovered workspace hold offers Retry (`canRetry`) and says "Retry or send a message to continue." An intentional Stop keeps "send a message" as its only continuation. +A held task can also have a pending question or connection request, for example +the question a parked turn asked before its Runner was lost. Such a decision does +not hold a person's Retry: the new run sees the open decision, and its later +answer still wakes the agent. Only a plain question (`ask_user_questions`) or +connection request (`connection_intent`) is released this way. A pending approval +and any decision with approval semantics (a plan review, a governed tool action, +a secret proposal, a connection authorization, or another confirmation) still +hold every continuation, Retry included. A new message supersedes a plain +question, as before, but waits for a decision with approval semantics. The person's own answer to the +decision is that explicit input: it continues the task through the hold in a +fresh session, like a new message, and never replays the failed run. An answer +that a different user resolved does not continue the task. + `native_remote_recovery_lease_mismatch` carries a closed detail: `native_remote_recovery_lease_mismatch: stage= field=`. The stage is `workspace_preparation`, `environment_acquire`, `environment_resume`, diff --git a/server/src/services/execution-continuation.ts b/server/src/services/execution-continuation.ts index b5882d0f19f..29a44f2e316 100644 --- a/server/src/services/execution-continuation.ts +++ b/server/src/services/execution-continuation.ts @@ -498,12 +498,17 @@ export async function buildExecutionContinuation(input: { ? value.failedRunId === failedRunId && retryWakes.some(wake => wake.runId === value.runId && wake.requestedByActorId === value.actorId && priorRuns.some(run => run.id === wake.runId && run.retryOfRunId === failedRunId)) - : value.interactionId + : value.interactionId && value.queuedCommentInterruptId ? interactions.some(interaction => interaction.id === value.interactionId && interaction.id === triggerInteraction?.id && ["accepted", "answered", "rejected"].includes(interaction.status) && interruptQueues.some(queue => queue.id === value.queuedCommentInterruptId && queue.runId === value.runId && object(queue.payload?.queuedCommentInterrupt).actorId === value.actorId && queuedInteractionId(queue.payload) === interaction.id && queue.payload?.interactionStatus === interaction.status)) + : value.interactionId + ? (!input.runId || value.runId === input.runId) && priorRuns.some(run => run.id === value.runId) && + interactions.some(interaction => interaction.id === value.interactionId && + ["accepted", "answered", "rejected"].includes(interaction.status) && + interaction.resolvedByUserId === value.actorId && !interaction.resolvedByAgentId) : rows.some(comment => comment.id === value.commentId && comment.authorType === "user" && (!("commentUpdatedAt" in value || "commentBodyHash" in value || value.automaticRetry) || ( diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index 836f636e109..460374426f1 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -142,6 +142,7 @@ export async function admitExplicitNativeContinuation(input: { /** Internal delivery of an unconsumed, user-authored saved queue entry. */ queuedCommentRequestId?: string; durableChatRequest?: DurableChatWakeupRequest; + interactionPayload?: unknown; dryRun?: boolean; onBlocked?: (reason: string, message: string) => void; }): Promise<{ previousRunId: string; commentId: string | null; failedRunId?: string; taskRetryRequestId?: string; queuedCommentInterruptId?: string; readiness?: ContinuationReadiness } | null> { @@ -150,7 +151,12 @@ export async function admitExplicitNativeContinuation(input: { if (input.actorType !== "user" || !actorId) return null; const retry = input.reason === "retry_failed_run" && z.string().guid().safeParse(input.failedRunId).success; - if (!retry && !input.taskRetryRequestId && !input.queuedCommentInterruptId && (!commentId || !z.string().guid().safeParse(commentId).success || + const decisionResponse = !retry && !input.taskRetryRequestId && !input.queuedCommentInterruptId && + !input.queuedCommentRequestId && !commentId && input.reason === "issue_commented" && input.interactionPayload + ? await readQueuedInteractionResponse(db, companyId, issueId, input.interactionPayload) : null; + const decided = decisionResponse && decisionResponse.comment.authorType === "user" && + decisionResponse.comment.authorUserId === actorId && !decisionResponse.toolActionOutcome ? decisionResponse : null; + if (!retry && !input.taskRetryRequestId && !input.queuedCommentInterruptId && !decided && (!commentId || !z.string().guid().safeParse(commentId).success || !isUserCommentContinuationReason(input.reason))) return null; const [task] = await db.select().from(issues).where(and( eq(issues.companyId, companyId), eq(issues.id, issueId), @@ -173,7 +179,7 @@ export async function admitExplicitNativeContinuation(input: { sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${actorId}`, )) : []; const response = interruptQueue - ? await readQueuedInteractionResponse(db, companyId, issueId, interruptQueue.payload) : null; + ? await readQueuedInteractionResponse(db, companyId, issueId, interruptQueue.payload) : decided; const queuedInterrupt = Boolean(interruptQueue && (response || (commentId && queuedCommentIdsFromWakePayload(interruptQueue.payload).includes(commentId)))); if (input.queuedCommentInterruptId && !queuedInterrupt) return null; @@ -307,6 +313,9 @@ export async function admitExplicitNativeContinuation(input: { eq(issueThreadInteractions.companyId, companyId), eq(issueThreadInteractions.issueId, issueId), eq(issueThreadInteractions.status, "pending"), activeIssueInteractionCondition({ responsibleUserId: actorId }), + taskRetry ? sql`not (${issueThreadInteractions.kind} in ('ask_user_questions', 'connection_intent') + and not (${issueThreadInteractions.payload} ?| array['toolAction', 'secretProposal', 'connectionAuthorization']) + and coalesce(${issueThreadInteractions.payload} #>> '{target,type}', '') <> 'issue_document')` : undefined, )).limit(1); const [pendingApproval] = await db.select({ id: approvals.id }).from(issueApprovals).innerJoin(approvals, and( eq(approvals.id, issueApprovals.approvalId), eq(approvals.companyId, companyId), diff --git a/server/src/services/governed-wait-runner-loss.test.ts b/server/src/services/governed-wait-runner-loss.test.ts new file mode 100644 index 00000000000..ba0277a7a3c --- /dev/null +++ b/server/src/services/governed-wait-runner-loss.test.ts @@ -0,0 +1,207 @@ +import { randomUUID } from "node:crypto"; +import { readFile } from "node:fs/promises"; +import path from "node:path"; +import { and, eq } from "drizzle-orm"; +import { describe, expect, it, vi } from "vitest"; +import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import { activityLog, agentWakeupRequests, heartbeatRuns, issueComments, issueRecoveryActions, issueThreadInteractions, issues } from "@paperclipai/db"; +import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; +import { heartbeatService } from "./heartbeat.js"; +import { requestInterruptedNativeTurnRetry } from "./heartbeat/restart-interrupted-turn.js"; +import { getExecutionBlocker } from "./execution-blocker.js"; +import { taskInterruptionPresentation } from "./task-interruption-presentation.js"; +import { readSystemRecovery } from "./system-recovery-state.js"; +import { reconcileRetainedWorkspaces } from "./workspace-recovery.js"; +import { requestTaskRetry } from "./task-retry.js"; +import { buildExecutionContinuation } from "./execution-continuation.js"; +import { useWorkspaceRecoveryFixture, action, db, run, seed } from "./__tests__/workspace-recovery-fixture.js"; + +const targets = vi.hoisted(() => new Map()); +vi.mock("./environment-execution-target.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, resolveEnvironmentExecutionTarget: async (input: Parameters[0]) => + (input.leaseId ? targets.get(input.leaseId) : undefined) ?? actual.resolveEnvironmentExecutionTarget(input) }; +}); + +describe("a held task with a pending decision after its remote runner was lost", () => { + useWorkspaceRecoveryFixture(targets); + + const shapes = { + parkedQuestion: { wakeReason: "issue_assigned", turnEvents: ["turn.started", "turn.accepted"], + kind: "ask_user_questions", fromFailedRun: true }, + olderConnectionRequest: { wakeReason: "issue_monitor_due", turnEvents: [], + kind: "connection_intent", fromFailedRun: false }, + } as const; + + async function lostRunnerWithPendingDecision(shape: keyof typeof shapes) { + const s = shapes[shape]; + const f = await seed("native"); + const original = await run(f.runId); + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId, + source: "automation", triggerDetail: "system", reason: s.wakeReason, status: "failed", runId: f.runId, + finishedAt: new Date(), payload: { issueId: f.issueId } }).returning(); + await db.update(heartbeatRuns).set({ errorCode: "native_continuation_requires_reconciliation", + error: "runner_remote_process_identity_unavailable", wakeupRequestId: wake.id, responsibleUserId: "local-board", + startedAt: new Date(Date.now() - 20 * 60_000), contextSnapshot: { issueId: f.issueId, wakeReason: s.wakeReason }, + }).where(eq(heartbeatRuns.id, f.runId)); + for (const eventType of s.turnEvents) + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, agentId: original.agentId, + eventType, stream: "system", level: "info", message: eventType }); + const [older] = s.fromFailedRun ? [] : await db.insert(heartbeatRuns).values({ companyId: f.companyId, + agentId: original.agentId, issueId: f.issueId, status: "succeeded", createdAt: new Date(Date.now() - 2 * 86_400_000), + finishedAt: new Date(Date.now() - 86_400_000), contextSnapshot: { issueId: f.issueId } }).returning(); + const [decision] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, + kind: s.kind, status: "pending", continuationPolicy: "wake_assignee", sourceRunId: older?.id ?? f.runId, + createdByAgentId: original.agentId, title: "Which default should new agents get?", + payload: { version: 1, questions: [{ id: "q1", prompt: "Which default?", selectionMode: "single", + options: [{ id: "a", label: "Read only" }, { id: "b", label: "Full" }] }] } as never }).returning(); + const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: original.agentId, status: "running" }); + expect(await requestInterruptedNativeTurnRetry(db, f.runId)).toMatchObject({ requested: false }); + await vi.waitFor(async () => { + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, current => service.resumeRemoteStopComments(current)) ?? []); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + }, { timeout: 10_000, interval: 10 }); + expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n"); + return { ...f, agentId: original.agentId, decision, service }; + } + const queuedSuccessors = async (companyId: string) => db.select().from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "queued"))); + const deferredWaits = async (companyId: string) => (await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.status, "deferred_issue_execution")))) + .map(wake => (wake.payload?.executionWait as { reason?: string } | undefined)?.reason); + type Decision = typeof issueThreadInteractions.$inferSelect; + const resolve = (decision: Decision, resolvedBy: string) => db.update(issueThreadInteractions).set({ status: "answered", + resolvedByUserId: resolvedBy, resolvedAt: new Date(), result: { version: 1, answers: [{ questionId: "q1", optionIds: ["a"] }] } as never }) + .where(eq(issueThreadInteractions.id, decision.id)); + const deliverAnswer = (f: Awaited>, decision: Decision, userId: string) => + f.service.wakeup(f.agentId, { + source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: f.issueId, interactionId: decision.id, interactionKind: decision.kind, + interactionStatus: "answered", sourceCommentId: null, sourceRunId: decision.sourceRunId, mutation: "interaction" }, + idempotencyKey: `interaction:${decision.id}:answered`, + requestedByActorType: "user", requestedByActorId: userId, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, interactionId: decision.id, + interactionKind: decision.kind, interactionStatus: "answered", sourceCommentId: null, + sourceRunId: decision.sourceRunId, wakeReason: "issue_commented", source: "issue.interaction.resolve" }, + }); + const answer = async (f: Awaited>, userId: string, resolvedBy = userId) => { + await resolve(f.decision, resolvedBy); + await deliverAnswer(f, f.decision, userId); + }; + const dispatchContinuation = (f: Awaited>, successor: typeof heartbeatRuns.$inferSelect) => + buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + runId: successor.id, context: successor.contextSnapshot ?? {}, summary: null, exposeLowTrustRaw: false }); + + it.each(Object.keys(shapes) as Array)("%s: Retry starts exactly one run after the sandbox work is saved", async shape => { + const f = await lostRunnerWithPendingDecision(shape); + const issue = (await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]; + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ runId: f.runId, canRetry: true, + runError: "runner_remote_process_identity_unavailable", workspaceRecovery: { status: "recovered" } }); + expect(await taskInterruptionPresentation(db, issue)).toMatchObject({ state: "failed", canRetry: true, observedRunId: f.runId }); + const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: "local-board", + actorSource: "local_implicit", idempotencyKey: randomUUID(), observedRunId: f.runId }); + expect(receipt).toMatchObject({ accepted: true }); + await f.service.resumeTaskRetry(f.companyId, receipt.requestId!); + expect({ successors: (await queuedSuccessors(f.companyId)).length, deferred: await deferredWaits(f.companyId) }) + .toEqual({ successors: 1, deferred: [] }); + const [successor] = await queuedSuccessors(f.companyId); + expect(successor.contextSnapshot).toMatchObject({ previousRunId: f.runId, taskRetryRequestId: receipt.requestId, forceFreshSession: true }); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, f.decision.id)))[0].status).toBe("pending"); + }, 60_000); + + it.each(Object.keys(shapes) as Array)("%s: the person's answer to the pending decision continues the task", async shape => { + const f = await lostRunnerWithPendingDecision(shape); + await answer(f, "local-board"); + expect({ successors: (await queuedSuccessors(f.companyId)).length, deferred: await deferredWaits(f.companyId) }) + .toEqual({ successors: 1, deferred: [] }); + const [successor] = await queuedSuccessors(f.companyId); + expect(successor.contextSnapshot).toMatchObject({ interactionId: f.decision.id, previousRunId: f.runId, forceFreshSession: true, + explicitUserContinuation: { previousRunId: f.runId, commentId: null } }); + expect(await action(f.actionId)).toMatchObject({ status: "resolved", + evidence: { explicitUserContinuation: { actorId: "local-board", interactionId: f.decision.id, previousRunId: f.runId } } }); + expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, f.companyId), + eq(activityLog.action, "issue.execution_recovery_settled"), eq(activityLog.actorId, "local-board")))).toHaveLength(1); + const envelope = await dispatchContinuation(f, successor); + expect(envelope).toMatchObject({ trigger: { interactionId: f.decision.id }, interruptedRunId: f.runId }); + expect(envelope.interactionOutcomes).toEqual(expect.arrayContaining([expect.objectContaining({ id: f.decision.id })])); + }, 60_000); + + it("dispatches the first answer's successor after a second answer coalesces into it", async () => { + const f = await lostRunnerWithPendingDecision("parkedQuestion"); + const [second] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, + kind: "ask_user_questions", status: "pending", continuationPolicy: "wake_assignee", sourceRunId: f.runId, + createdByAgentId: f.agentId, title: "Which audit level?", payload: f.decision.payload }).returning(); + await resolve(f.decision, "local-board"); + await resolve(second, "local-board"); + await deliverAnswer(f, f.decision, "local-board"); + await deliverAnswer(f, second, "local-board"); + const successors = await queuedSuccessors(f.companyId); + expect(successors).toHaveLength(1); + const [secondWake] = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + eq(agentWakeupRequests.idempotencyKey, `interaction:${second.id}:answered`))); + expect(secondWake).toMatchObject({ status: "deferred_issue_execution", runId: null }); + for (const context of [successors[0].contextSnapshot ?? {}, + { ...successors[0].contextSnapshot, interactionId: second.id, interactionStatus: "answered" }]) { + const envelope = await dispatchContinuation(f, { ...successors[0], contextSnapshot: context }); + expect(envelope.interactionOutcomes).toEqual(expect.arrayContaining([ + expect.objectContaining({ id: f.decision.id }), expect.objectContaining({ id: second.id })])); + } + }, 60_000); + + it.each([ + ["a plan review", "request_confirmation", { version: 1, prompt: "Approve the plan?", target: { type: "issue_document", key: "plan", revisionId: randomUUID() } }], + ["a governed tool action", "request_confirmation", { version: 1, prompt: "Run it?", toolAction: { tool: "deploy" } }], + ["a secret proposal", "request_confirmation", { version: 1, prompt: "Store it?", secretProposal: { name: "TOKEN" } }], + ["a connection authorization", "connection_intent", { version: 1, serviceName: "Sentry", connectionAuthorization: { scope: "read" } }], + ])("keeps Retry on hold while %s is pending", async (_label, kind, payload) => { + const f = await lostRunnerWithPendingDecision("parkedQuestion"); + await db.update(issueThreadInteractions).set({ kind, payload: payload as never }) + .where(eq(issueThreadInteractions.id, f.decision.id)); + const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: "local-board", + actorSource: "local_implicit", idempotencyKey: randomUUID(), observedRunId: f.runId }); + await f.service.resumeTaskRetry(f.companyId, receipt.requestId!); + expect({ successors: (await queuedSuccessors(f.companyId)).length, deferred: await deferredWaits(f.companyId) }) + .toEqual({ successors: 0, deferred: ["decision_pending"] }); + }, 60_000); + + it("lets a new message supersede a plain question and continue the task", async () => { + const f = await lostRunnerWithPendingDecision("parkedQuestion"); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "local-board", body: "Please continue." }).returning(); + await f.service.wakeup(f.agentId, { + source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" }, + requestedByActorType: "user", requestedByActorId: "local-board", + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, wakeReason: "issue_commented", wakeCommentId: comment.id }, + }); + expect({ successors: (await queuedSuccessors(f.companyId)).length, deferred: await deferredWaits(f.companyId) }) + .toEqual({ successors: 1, deferred: [] }); + }, 60_000); + + it("keeps a new message waiting while a plan review is pending", async () => { + const f = await lostRunnerWithPendingDecision("parkedQuestion"); + await db.update(issueThreadInteractions).set({ kind: "request_confirmation", + payload: { version: 1, prompt: "Approve the plan?", target: { type: "issue_document", key: "plan", revisionId: randomUUID() } } as never }) + .where(eq(issueThreadInteractions.id, f.decision.id)); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "local-board", body: "Please continue." }).returning(); + await f.service.wakeup(f.agentId, { + source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" }, + requestedByActorType: "user", requestedByActorId: "local-board", + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, wakeReason: "issue_commented", wakeCommentId: comment.id }, + }); + expect({ successors: (await queuedSuccessors(f.companyId)).length, deferred: await deferredWaits(f.companyId) }) + .toEqual({ successors: 0, deferred: ["decision_pending"] }); + }, 60_000); + + it("keeps the hold when the wake's user did not give the answer", async () => { + const f = await lostRunnerWithPendingDecision("parkedQuestion"); + await answer(f, "other-user", "local-board"); + expect(await queuedSuccessors(f.companyId)).toHaveLength(0); + expect(await deferredWaits(f.companyId)).toEqual(["execution_recovery"]); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)))[0].evidence) + .not.toHaveProperty("explicitUserContinuation"); + }, 60_000); +}); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index d504a101231..c949cfd5384 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -3692,6 +3692,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende queuedCommentRequestId: opts.queuedCommentRequestId, taskRetryRequestId: opts.taskRetryRequestId, ...(durableRequest?.resumeDeferred && !failedChatRetry ? { durableChatRequest: durableRequest } : {}), + interactionPayload: payload, dryRun: true, onBlocked: (reason, message) => { continuationBlocked = true; continuationWait = { reason, message }; }, })) && (executionBlocker || (continuationBlocked && !verifiedConversationReset))) return deferBlockedExecution(continuationHold()); @@ -4553,6 +4554,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende queuedCommentRequestId: opts.queuedCommentRequestId, taskRetryRequestId: opts.taskRetryRequestId, ...(durableRequest?.resumeDeferred && !failedChatRetry ? { durableChatRequest: durableRequest } : {}), + interactionPayload: payload, onBlocked: (reason, message) => { continuationBlocked = true; continuationWait = { reason, message }; }, }); if (!explicitContinuation && (executionBlocker || (continuationBlocked && !verifiedConversationReset)) && !ownsReconstructionContinuation)