diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 6f1334faebd..f86cdf04cf0 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1725,7 +1725,15 @@ assigned the task, or clicked Retry. It does this only when the run recorded no provider turn or its last provider turn is settled. Activity records the request as `restart-recovery` on behalf of that user. Admission keeps every gate: saved-work recovery, stop proof, approvals, pauses, budget and ownership. A Retry that the -server requested gets no second automatic attempt. System wakes and runs with an +server requested gets no second automatic attempt. A system wake (for example a +monitor or an automatic assignment) continues this way only with a positive +record that no turn was ever issued: the refused reattachment proved the Runner +process gone, and the run's controller journal, which records each turn command +before it is sent, holds no turn command and no turn event +(`providerTurnIssued: false` on `native.remote_runner_recovery_failed`). The run +must also have no turn or tool event, and the Retry runs under the run's +responsible user. Missing events alone never qualify, because a turn can reach +the provider before its events are stored. Other system wakes and runs with an in-flight provider turn do not continue automatically. For these failed runs, a 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 diff --git a/server/src/services/heartbeat/restart-interrupted-turn.ts b/server/src/services/heartbeat/restart-interrupted-turn.ts index a1458a27826..16564be7238 100644 --- a/server/src/services/heartbeat/restart-interrupted-turn.ts +++ b/server/src/services/heartbeat/restart-interrupted-turn.ts @@ -36,6 +36,29 @@ export function isProvenTurnSettlement(event: { eventType: string; payload: unkn asRecord(asRecord(payload.prpEvent).payload).providerTerminalObserved !== false; } +/** Any turn or tool event: the run may have done provider work that a new turn could repeat. */ +const PROVIDER_TURN_ACTIVITY_EVENTS = [...TURN_STARTED_EVENTS, ...TURN_SETTLED_EVENTS, + "tool.execution.started", "tool.execution.completed"]; + +/** + * The run's last refused reattachment proved its Runner process gone and its + * controller journal proved that no turn command was ever issued + * (`providerTurnIssued: false`, recorded by the executor). No turn or tool + * event may exist either. + */ +async function hasProvenNoProviderTurn(db: Db, run: { companyId: string; id: string }) { + const [activity] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + inArray(heartbeatRunEvents.eventType, PROVIDER_TURN_ACTIVITY_EVENTS))).limit(1); + if (activity) return false; + const [refusal] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + eq(heartbeatRunEvents.eventType, "native.remote_runner_recovery_failed"), + )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); + const proof = asRecord(asRecord(refusal?.payload).remoteRunnerRecovery); + return proof.stage === "process_verification" && proof.reason === "process_not_running" && proof.providerTurnIssued === false; +} + /** Automatic continuation is requested at most once per chain of runs. */ export const RESTART_RETRY_KEY_PREFIX = "restart-interrupted-turn:"; @@ -47,7 +70,9 @@ type Outcome = { requested: true; requestId: string } | { requested: false; reas * request the same Retry a user would click, once and idempotently per run, * under the requesting user's authority. Only when the run's user input was not * yet acted on, or its last provider turn is settled; an in-flight turn's - * outcome is unknown, so it keeps the visible Retry. Admission keeps every gate, + * outcome is unknown, so it keeps the visible Retry. A system wake continues + * this way only when the controller journal proved that no turn was ever + * issued, under the run's responsible user. Admission keeps every gate, * including stop proof and saved-work recovery before the next turn. */ export async function requestRestartInterruptedTurnRetry(db: Db, claim: NativeRestartRecoveryClaim): Promise { @@ -71,11 +96,19 @@ export async function requestInterruptedNativeTurnRetry(db: Db, runId: string): const [wake] = await db.select().from(agentWakeupRequests).where(and( eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.id, run.wakeupRequestId), eq(agentWakeupRequests.runId, run.id))); - const actorId = wake?.requestedByActorType === "user" ? wake.requestedByActorId : null; + const userActorId = wake?.requestedByActorType === "user" ? wake.requestedByActorId : null; const reason = typeof context.wakeReason === "string" ? context.wakeReason : wake?.reason; - // System wakes keep their policy: only a new user turn continues them. - if (!actorId) return { requested: false, reason: "not_user_request" }; - if (reason === "task_retry") { + let actorId = userActorId; + if (!userActorId) { + // System wakes keep their policy (only a new user turn continues them), + // except when the controller journal positively proves that no turn was + // ever issued: nothing the run did can be repeated, so the Retry runs once + // under the run's responsible user. Missing events alone prove nothing, + // because a turn can reach the provider before its events are stored. + if (!wake || reason === "task_retry" || !run.responsibleUserId || !await hasProvenNoProviderTurn(db, run)) + return { requested: false, reason: "not_user_request" }; + actorId = run.responsibleUserId; + } else if (reason === "task_retry") { // A Retry this hook already requested gets no second automatic attempt. const retryId = typeof context.taskRetryRequestId === "string" ? context.taskRetryRequestId : null; const [receipt] = retryId ? await db.select({ key: agentWakeupRequests.idempotencyKey }).from(agentWakeupRequests).where(and( @@ -88,10 +121,11 @@ export async function requestInterruptedNativeTurnRetry(db: Db, runId: string): const [comment] = commentId ? await db.select({ id: issueComments.id }).from(issueComments).where(and( eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId), sql`${issueComments.id}::text = ${commentId}`, eq(issueComments.authorType, "user"), - eq(issueComments.authorUserId, actorId), isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId), + eq(issueComments.authorUserId, userActorId), isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId), )) : []; if (!comment && reason !== "issue_assigned") return { requested: false, reason: "message_unavailable" }; } else return { requested: false, reason: "not_user_request" }; + if (!actorId) return { requested: false, reason: "not_user_request" }; const [issue] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId }).from(issues) .where(and(eq(issues.companyId, run.companyId), eq(issues.id, issueId))); if (!issue || issue.assigneeAgentId !== run.agentId || ["done", "cancelled"].includes(issue.status)) diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index add78046e35..c46e7976c96 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -6252,11 +6252,15 @@ async function migrateRunnerdStateRootForExecution(input: { await verifyPluginRunnerTaskReattachment(target.runnerTask, identity); return; } - await verifyRemoteRunnerOrSettledExit({ - claim: input.restartRecovery, target: input.runnerExecutionTarget, identity, - runId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution), - providerAttempt: input.restartRecovery.providerAttempt, - }); + try { + await verifyRemoteRunnerOrSettledExit({ + claim: input.restartRecovery, target: input.runnerExecutionTarget, identity, + runId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution), + providerAttempt: input.restartRecovery.providerAttempt, + }); + } catch (error) { + throw withRemoteRunnerTurnProof(error, scoped); + } return; } if (durableIdentityMatchesExecution(identity, input.execution)) { @@ -6858,6 +6862,20 @@ export async function stopStaleRemoteRunnerBeforeBootstrap(input: { } } +/** + * A refused reattachment whose Runner process is proven gone records whether + * this run's controller journal proves that no turn command was ever issued. + * The controller writes each turn command there before sending it, so only + * that positive record (never an absence of run events) allows an automatic + * relaunch that cannot repeat provider work. + */ +export function withRemoteRunnerTurnProof(error: unknown, root: string): unknown { + const diagnostic = readRemoteRunnerRecoveryError(error); + if (diagnostic?.stage !== "process_verification" || diagnostic.reason !== "process_not_running" || + !runnerdStateProvesNoRemoteProviderTurn(root)) return error; + return attachRemoteRunnerRecoveryDiagnostic(error, { ...diagnostic, providerTurnIssued: false }); +} + /** The remote controller record proves its runner was never sent a turn. */ export function runnerdStateProvesNoRemoteProviderTurn(root: string): boolean { try { diff --git a/server/src/services/native-runtime/remote-runner-recovery-diagnostics.ts b/server/src/services/native-runtime/remote-runner-recovery-diagnostics.ts index 78d660b5bfc..cd6b7710842 100644 --- a/server/src/services/native-runtime/remote-runner-recovery-diagnostics.ts +++ b/server/src/services/native-runtime/remote-runner-recovery-diagnostics.ts @@ -21,6 +21,8 @@ export interface RemoteRunnerRecoveryDiagnostic { reason: typeof reasons[number]; runnerLifecycle?: typeof runnerLifecycles[number]; runnerSettled?: boolean; + /** Present only as `false`: the controller journal proves no turn command was ever issued. */ + providerTurnIssued?: false; } export function describeExitedRemoteRunner(state: Record): Pick; + const { stage, reason, runnerLifecycle, runnerSettled, providerTurnIssued } = value as Record; if (!stages.some(candidate => candidate === stage) || !reasons.some(candidate => candidate === reason)) return null; return { stage, reason, ...(runnerLifecycles.some(candidate => candidate === runnerLifecycle) ? { runnerLifecycle } : {}), ...(typeof runnerSettled === "boolean" ? { runnerSettled } : {}), + ...(providerTurnIssued === false ? { providerTurnIssued } : {}), } as RemoteRunnerRecoveryDiagnostic; } catch { return null; diff --git a/server/src/services/native-runtime/remote-runner-turn-proof.test.ts b/server/src/services/native-runtime/remote-runner-turn-proof.test.ts new file mode 100644 index 00000000000..ff585e5058d --- /dev/null +++ b/server/src/services/native-runtime/remote-runner-turn-proof.test.ts @@ -0,0 +1,52 @@ +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { withRemoteRunnerTurnProof } from "./native-session-executor.js"; +import { RemoteRunnerRecoveryError, readRemoteRunnerRecoveryError } from "./remote-runner-recovery-diagnostics.js"; + +const roots: string[] = []; +afterEach(async () => { await Promise.all(roots.splice(0).map(root => rm(root, { recursive: true, force: true }))); }); + +async function journal(commands: Array<{ type: string; status: string }>, eventTypes = ["session.started"]) { + const root = await mkdtemp(path.join(os.tmpdir(), "turn-proof-")); + roots.push(root); + await mkdir(path.join(root, "control-plane"), { recursive: true }); + await writeFile(path.join(root, "control-plane", "control-plane-state.json"), JSON.stringify({ + schema: "paperclip.runner.durable.control-plane-state.v2", connectionCount: 1, + commands, committedEvents: eventTypes.map((eventType, index) => ({ sourceSeq: index + 1, eventType })), + ackedSourceSeq: eventTypes.length, + })); + return root; +} +const exited = () => new RemoteRunnerRecoveryError({ stage: "process_verification", reason: "process_not_running", + runnerLifecycle: "ready", runnerSettled: false }); + +// Staging class T: the Runner process was proven gone. Only the controller +// journal, which records each turn command before sending it, can prove no +// turn was issued; an absence of run events cannot. +describe("withRemoteRunnerTurnProof", () => { + it("records the proof when the journal holds only session commands", async () => { + const root = await journal([{ type: "run.prepare", status: "completed" }, { type: "session.open", status: "completed" }, + { type: "session.snapshot", status: "pending" }]); + expect(readRemoteRunnerRecoveryError(withRemoteRunnerTurnProof(exited(), root))).toMatchObject({ + reason: "process_not_running", providerTurnIssued: false }); + }); + + it.each([ + ["a turn command was issued", [{ type: "session.open", status: "completed" }, { type: "turn.start", status: "pending" }], ["session.started"]], + ["a turn event was committed", [{ type: "session.open", status: "completed" }], ["session.started", "turn.started"]], + ])("records nothing when %s", async (_label, commands, events) => { + const root = await journal(commands, events); + expect(readRemoteRunnerRecoveryError(withRemoteRunnerTurnProof(exited(), root))).not.toHaveProperty("providerTurnIssued"); + }); + + it("records nothing without a readable journal or without a proven exit", async () => { + const empty = await mkdtemp(path.join(os.tmpdir(), "turn-proof-")); + roots.push(empty); + expect(readRemoteRunnerRecoveryError(withRemoteRunnerTurnProof(exited(), empty))).not.toHaveProperty("providerTurnIssued"); + const root = await journal([{ type: "session.open", status: "completed" }]); + const unreachable = new RemoteRunnerRecoveryError({ stage: "process_verification", reason: "probe_transport_failed" }); + expect(readRemoteRunnerRecoveryError(withRemoteRunnerTurnProof(unreachable, root))).not.toHaveProperty("providerTurnIssued"); + }); +}); diff --git a/server/src/services/workspace-recovery-retry-copyback.test.ts b/server/src/services/workspace-recovery-retry-copyback.test.ts index 39bbad5f77b..def29ff74d5 100644 --- a/server/src/services/workspace-recovery-retry-copyback.test.ts +++ b/server/src/services/workspace-recovery-retry-copyback.test.ts @@ -114,6 +114,17 @@ describe("automatic retained workspace recovery", () => { } const queuedSuccessors = async (companyId: string) => db.select().from(heartbeatRuns) .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "queued"))); + async function systemWake(f: Awaited>, proof?: { providerTurnIssued?: false }) { + await db.update(agentWakeupRequests).set({ requestedByActorType: "system", requestedByActorId: null, + reason: "issue_monitor_due" }).where(eq(agentWakeupRequests.id, f.wake.id)); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, wakeReason: "issue_monitor_due" } }) + .where(eq(heartbeatRuns.id, f.runId)); + // The executor's refused reattachment (see withRemoteRunnerTurnProof). + if (proof) await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, agentId: f.agentId, + eventType: "native.remote_runner_recovery_failed", stream: "system", level: "error", message: "Could not verify the existing remote runner.", + payload: { remoteRunnerRecovery: { stage: "process_verification", reason: "process_not_running", + runnerLifecycle: "ready", runnerSettled: false, ...proof } } }); + } it.each([ { title: "the first attachment found the previous runner gone", turn: "none" as const, error: "runner_remote_process_identity_unavailable", initial: true }, @@ -122,8 +133,14 @@ describe("automatic retained workspace recovery", () => { { title: "a same-run retry found the settled turn's remote runner gone", turn: "settled" as const, error: "runner_remote_process_identity_unavailable", initial: false }, { title: "a same-run retry found the settled turn's sandbox lease gone", turn: "settled" as const, error: "native_remote_recovery_lease_mismatch", initial: false }, { title: "a same-run retry named the lease check that failed", turn: "settled" as const, error: "native_remote_recovery_lease_mismatch: stage=environment_acquire field=lease_status", initial: false }, - ])("continues a user turn once, without a second message, after $title and recovery saved the work", async ({ turn, error, initial }) => { + // Staging class T (c14bab19): a monitor wake's session snapshot timed out + // before any turn, then the same-run retry found the Runner gone. Its + // controller journal proved no turn command was ever issued, so nothing can + // be repeated and the run's responsible user's Retry continues it once. + { title: "a system wake's run found its runner gone and its journal proved no turn", turn: "none" as const, error: "runner_remote_process_identity_unavailable", initial: false, system: true }, + ])("continues a user turn once, without a second message, after $title and recovery saved the work", async ({ turn, error, initial, system }) => { const f = await restartAllocationLostSource(turn, error); + if (system) await systemWake(f, { providerTurnIssued: false }); const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); // Keep the agent's only slot busy so the successor stays observable as queued. await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); @@ -164,18 +181,24 @@ describe("automatic retained workspace recovery", () => { it.each([ { title: "its provider turn was still in flight", change: "turn" }, { title: "its provider turn ended without an observed provider terminal", change: "unproven" }, - { title: "the source was a system wake", change: "system" }, + { title: "the source was a system wake whose provider turn settled", change: "system" }, + // No run events is not proof: a turn can reach the provider before its + // events are stored. Only the controller journal can prove it. + { title: "the source was a system wake with no run events but no journal proof", change: "system_unproven" }, + { title: "the source was a system wake whose run used a tool before its turn event", change: "system_tool" }, + { title: "the source was a system wake whose run had no responsible user", change: "system_unowned" }, { title: "the failure was not a lost runner or allocation", change: "error" }, // Staging class P: the new sandbox had no Runner state to resume the saved // provider session. The Runner now names it, so the run ends with the // permanent code and Retry continues in a fresh provider session. { title: "the Runner had no saved state to resume the provider session", change: "unrestorable" }, ])("does not continue automatically when $title, and offers a Retry that works on first use", async ({ change }) => { - const f = await restartAllocationLostSource(change === "turn" ? "in_flight" : change === "unproven" ? "unproven" : "none"); - if (change === "system") await db.update(agentWakeupRequests).set({ requestedByActorType: "system", requestedByActorId: null, - reason: "issue_children_completed" }).where(eq(agentWakeupRequests.id, f.wake.id)); - if (change === "system") await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, wakeReason: "issue_children_completed" } }) - .where(eq(heartbeatRuns.id, f.runId)); + const f = await restartAllocationLostSource(change === "turn" ? "in_flight" : change === "unproven" ? "unproven" + : change === "system" ? "settled" : "none"); + if (change.startsWith("system")) await systemWake(f, change === "system_unproven" ? undefined : { providerTurnIssued: false }); + if (change === "system_tool") await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, + agentId: f.agentId, eventType: "tool.execution.started", stream: "system", level: "info", message: "tool" }); + if (change === "system_unowned") await db.update(heartbeatRuns).set({ responsibleUserId: null }).where(eq(heartbeatRuns.id, f.runId)); if (change === "error") await db.update(heartbeatRuns).set({ error: "native_provider_terminal_failed" }).where(eq(heartbeatRuns.id, f.runId)); if (change === "unrestorable") await db.update(heartbeatRuns).set({ errorCode: "native_provider_state_incompatible", error: "native_session_recovery_failed: NativeRunnerResumeStateUnavailableError: PRP provider resume state is unavailable" })