diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 492b8f0cf76..ad6a8ae47cc 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1715,6 +1715,15 @@ 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. A pending approval still holds every continuation. +A new message still waits for the decision. 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/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index c198f1bbfb0..a50e990d979 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -141,6 +141,8 @@ export async function admitExplicitNativeContinuation(input: { /** Internal delivery of an unconsumed, user-authored saved queue entry. */ queuedCommentRequestId?: string; durableChatRequest?: DurableChatWakeupRequest; + /** The wake payload; a user's resolution of a pending decision carries its interaction here. */ + 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> { @@ -149,7 +151,15 @@ 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 || + // A held task with a pending decision waits for that decision ("decision_pending" + // below). The person's own answer to it is therefore the explicit user input + // that continues the task, like a new message; it never replays the failed run. + 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), @@ -172,7 +182,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; @@ -311,7 +321,10 @@ export async function admitExplicitNativeContinuation(input: { eq(approvals.id, issueApprovals.approvalId), eq(approvals.companyId, companyId), )).where(and(eq(issueApprovals.companyId, companyId), eq(issueApprovals.issueId, issueId), inArray(approvals.status, ["pending", "revision_requested"]))).limit(1); - if (pendingInteraction || pendingApproval) return blocked("decision_pending", "A pending approval or question must be resolved before this message can start."); + // A person's Retry is itself the decision to continue: an open question stays + // on the task for the new run to see and its answer still wakes the agent. + // Approvals keep gating every continuation. + if ((pendingInteraction && !taskRetry) || pendingApproval) return blocked("decision_pending", "A pending approval or question must be resolved before this message can start."); if (!sourceChecks.length) return null; const sources: Run[] = []; 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..4c4a4d8f992 --- /dev/null +++ b/server/src/services/governed-wait-runner-loss.test.ts @@ -0,0 +1,143 @@ +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, 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 { 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) }; +}); + +// Staging classes D′ and T (aeaa64a7, 3c4af814, a7a2f8dc, c14bab19): a remote +// native run's same-run retry found its Runner process gone (`process_not_running`, +// lifecycle `ready`) and ended with `runner_remote_process_identity_unavailable`. +// Saved-work recovery copied the sandbox files back and the task offered Retry. +// Each task also had a pending decision: D′ the question its own turn parked on, +// T and the Sentry tasks an older connection request. Retry then waited for the +// decision ("decision_pending") and the decision's answer waited for execution +// recovery ("execution_recovery"), so neither control could continue the task. +describe("a held task with a pending decision after its remote runner was lost", () => { + useWorkspaceRecoveryFixture(targets); + + const shapes = { + // D′: the turn was accepted, then parked on its own question; no provider terminal. + parkedQuestion: { wakeReason: "issue_assigned", turnEvents: ["turn.started", "turn.accepted"], + kind: "ask_user_questions", fromFailedRun: true }, + // T: a monitor wake whose session snapshot timed out before any turn; an + // older run's connection request is still pending on the task. + 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 }); + // Keep the agent's only slot busy so a successor stays observable as queued. + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: original.agentId, status: "running" }); + // A system wake gets no automatic Retry; the person's controls must work. + 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); + // The same wake the interaction route queues for `wake_assignee`. + const answer = async (f: Awaited>, userId: string, resolvedBy = userId) => { + await 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, f.decision.id)); + await f.service.wakeup(f.agentId, { + source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: f.issueId, interactionId: f.decision.id, interactionKind: f.decision.kind, + interactionStatus: "answered", sourceCommentId: null, sourceRunId: f.decision.sourceRunId, mutation: "interaction" }, + idempotencyKey: `interaction:${f.decision.id}:answered`, + requestedByActorType: "user", requestedByActorId: userId, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, interactionId: f.decision.id, + interactionKind: f.decision.kind, interactionStatus: "answered", sourceCommentId: null, + sourceRunId: f.decision.sourceRunId, wakeReason: "issue_commented", source: "issue.interaction.resolve" }, + }); + }; + + 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 }); + // The open decision stays on the task for the new run; Retry did not answer it. + 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 } }); + // The hold is settled as an explicit user continuation, never a replay. + 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); + }, 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)