From 028cca540693be3c9835e04c64c82e95941184a2 Mon Sep 17 00:00:00 2001 From: Mahmoud Mabrouk Date: Wed, 8 Jul 2026 12:35:26 +0200 Subject: [PATCH 1/5] feat(runner): park approval pauses on the live session (Claude ACP gates only) Slice 2 of the session-keepalive plan (#5153), stacked on the pool slice. A Claude ACP permission gate now parks the live session (awaiting_approval, 10-minute TTL) holding the pending permission id and the suspended prompt. The human's approve or deny answers the parked request directly via respondPermission, so the original tool call executes with its original byte-exact arguments; no model re-issues anything. This removes the argument-drift and task-restart failure classes from the live path. - Pi relay gates, Pi builtin gates, and client-tool MCP pauses never park (cold as today, asserted by tests); the cold decision-map fallback is untouched and covers TTL expiry, restarts, and every mismatch - the F-024 latch-loser sibling settle runs unconditionally on every pause - checkoutApproval removes the session from the map so a racing request can never destroy an in-flight resume; respondPermission is at-most-once - a parked prompt rejection evicts the session immediately Claude-Session: https://claude.ai/code/session_01AumZJ9xRd4XYNHThqy4rTv --- services/runner/src/engines/sandbox_agent.ts | 192 ++- .../engines/sandbox_agent/acp-interactions.ts | 25 + .../src/engines/sandbox_agent/session-pool.ts | 121 +- services/runner/src/server.ts | 246 +++- .../unit/session-keepalive-approval.test.ts | 1254 +++++++++++++++++ .../runner/tests/unit/session-pool.test.ts | 185 +++ 6 files changed, 1940 insertions(+), 83 deletions(-) create mode 100644 services/runner/tests/unit/session-keepalive-approval.test.ts diff --git a/services/runner/src/engines/sandbox_agent.ts b/services/runner/src/engines/sandbox_agent.ts index 4b0005a066..32c91b1667 100644 --- a/services/runner/src/engines/sandbox_agent.ts +++ b/services/runner/src/engines/sandbox_agent.ts @@ -358,6 +358,56 @@ interface CurrentTurn { onPermissionRequest?: (req: unknown) => void; } +/** + * A Claude ACP permission gate that paused the turn and can be answered later on the SAME live + * session (slice 2 keep-alive). Recorded ONLY for a harness ACP permission gate (never a Pi relay + * gate, a Pi builtin gate, or a client-tool MCP pause — those cannot be answered across a turn + * boundary and stay on the cold path). Existence of this record is what makes the dispatch park a + * paused session in `awaiting_approval` instead of tearing it down. + */ +export interface ParkedApproval { + /** Marks the pending-gate shape; the dispatch treats any other shape as a cold-fallback. */ + gateType: "claude-acp-permission"; + /** The ACP permission-request id, answered later via `session.respondPermission`. */ + permissionId: string; + /** The gated tool call's id — matched against the incoming approval envelope's toolCallId. */ + toolCallId: string; + /** The gated tool name (logging + the durable interaction row); never its args, in logs. */ + toolName: string | undefined; + /** The gated call's original args, used to seed the resume turn's trace/egress tool span. */ + args: unknown; + /** The durable interaction row token, resolved on the answer via the onResolveInteraction hook. */ + interactionToken: string; + /** The held original `prompt()` promise; the resume awaits it after `respondPermission`. */ + promptPromise?: Promise; +} + +/** Answer a parked Claude ACP permission gate on the live session (the keep-alive resume input). */ +export interface ResumeApprovalInput { + permissionId: string; + reply: "once" | "reject"; + toolCallId: string; + toolName: string | undefined; + args: unknown; + interactionToken: string; + promptPromise?: Promise; +} + +/** Per-turn options for `runTurn`. Absent (flag off / cold) means today's byte-identical path. */ +export interface RunTurnOptions { + /** A live continuation: send only the new user text instead of the full cold transcript. */ + continuation?: boolean; + /** + * Keep-alive approval park mode: on a Claude ACP permission gate the pause keeps the session + * alive (no settle/abort/destroy) so a later resume can answer it. A non-parkable pause (Pi + * relay, client tool) still tears down exactly as today, so this is safe to set on any eligible + * keep-alive turn. + */ + approvalParkMode?: boolean; + /** A live approval resume: answer the parked gate and stream the continued prompt's events. */ + resume?: ResumeApprovalInput; +} + /** * A session-scoped environment that can serve many turns. Everything expensive to build lives * here (sandbox, session, internal tool-MCP server, mounted cwd, relay/temp dirs); `destroy()` @@ -397,6 +447,18 @@ export interface SessionEnvironment { * so a tool-using turn still matches its own continuation (the FE keeps assistant tool parts). */ lastTurnToolCallIds: string[]; + /** + * The Claude ACP permission gate the LAST turn paused on (slice 2), or undefined. Set only for a + * harness ACP permission gate, reset at each turn start; the dispatch reads it after a paused + * turn to decide whether to park in `awaiting_approval` and, on the next request, how to resume. + */ + parkedApproval?: ParkedApproval; + /** + * How many Claude ACP permission gates resolved to pendingApproval THIS turn (reset at turn + * start). More than one means parallel gates the single-gate resume cannot answer, so the + * dispatch does not park (tears down cold as today). + */ + approvalGateCount: number; destroyed: boolean; /** Complete, idempotent teardown (all the finalizers the old per-run `finally` ran). */ destroy: () => Promise; @@ -603,6 +665,8 @@ export async function acquireEnvironment( closeToolMcp: undefined, currentTurn: undefined, lastTurnToolCallIds: [], + parkedApproval: undefined, + approvalGateCount: 0, destroyed: false, destroy: async () => {}, clearTurn: () => {}, @@ -900,8 +964,11 @@ export async function acquireEnvironment( turn.onPermissionRequest(req); return; } - // Between turns: slice 1 never parks an approval, so no turn owns this. Cancel by policy - // so a stray gate cannot hang; slice 2 will extend this handler to park it. + // Between turns: no turn owns this gate. A slice-2 approval park is recorded DURING the + // active turn (the gate fires while a prompt runs, routing through currentTurn), and a + // parked-on-approval session leaves its harness suspended on that gate — so nothing new + // fires here while parked. A gate that reaches this handler is therefore a genuine stray + // (e.g. a late teardown artifact): cancel by policy so it cannot hang. logger( `[keepalive] between-turns permission request, cancelling by policy id=${req?.id}`, ); @@ -937,13 +1004,18 @@ export async function runTurn( request: AgentRunRequest, emit?: EmitEvent, signal?: AbortSignal, - opts: { continuation?: boolean } = {}, + opts: RunTurnOptions = {}, ): Promise { const { plan, logger, deps } = env; const sessionId = env.sessionId; // Reset the per-turn tool-call id record (the park folds the completed turn's ids into the // expected next-history fingerprint). env.lastTurnToolCallIds = []; + // Reset the per-turn approval-park bookkeeping. A fresh turn starts with no parked gate; this + // turn re-records it only if it pauses on a Claude ACP permission gate. (The dispatch has + // already captured any prior park into `opts.resume` before calling us.) + env.parkedApproval = undefined; + env.approvalGateCount = 0; // Hoisted so the catch can flush a partial trace (mirroring the pre-split `otel?` handling — // a createOtel throw must still return `{ ok: false }`, not propagate raw) and the finally can // stop this turn's relay on EVERY exit path (a cleared sink must never orphan it). @@ -1002,10 +1074,23 @@ export async function runTurn( }); const pause = new PendingApprovalPauseController(() => { + // The F-024 sibling settle runs UNCONDITIONALLY, park mode or not: latch-loser tool calls + // announced before the winning gate can never execute this turn, and skipping the settle + // here would leave them as orphaned open parts whenever the dispatch later refuses the park + // (multi-gate, pool full) — `env.destroy()` does not re-run it. The exclusion keeps the + // gated (paused) call itself open, so the live resume is untouched. run.settleOpenToolCalls( (id) => pause.isPausedToolCall(id), TOOL_NOT_EXECUTED_PAUSED, ); + // Slice 2 park mode: a parkable Claude ACP permission gate recorded `env.parkedApproval` + // BEFORE firing this pause (the onUserApprovalGate hook runs before the single-pause latch). + // Keep the live session — the gated tool runs on the resume — so skip ONLY the mcpAbort and + // the destroySession. The teardown is not lost: the dispatch either parks the session or, + // if it decides not to (multi-gate, pool full), calls `env.destroy()` which runs them. A + // non-parkable pause (flag off, Pi relay/builtin gate, client tool) never records + // `parkedApproval`, so it still tears down here exactly as today. + if (opts.approvalParkMode && env.parkedApproval) return; // Abort any in-flight loopback `tools/call` (a paused Claude client tool) BEFORE the // session teardown, so its handler cannot write a result after the turn ends. env.mcpAbort.abort(); @@ -1106,6 +1191,21 @@ export async function runTurn( () => cred, ); }; + // Transition the durable interaction row to resolved once its gate is answered. Used both by + // the cold decision-map path (via attachPermissionResponder) and the live approval resume, + // which answers the parked gate directly. It mirrors the cold path's ordering against + // `cancelStaleInteractions` (server.ts): that sweep cancels only PENDING gates of OTHER turns, + // and by resume time the human already marked this gate responded, so it is spared here too. + const resolveInteractionToken = (token: string): void => { + const cred = runCredential(request); + if (!cred) return; + if ( + !buildWorkflowReferences(request.runContext?.workflow) + ?.workflow_revision + ) + return; + void resolveInteraction(sessionId, token, () => cred); + }; // Exactly one gate per call: the harness gate on Claude, the relay on Pi. const relayPermissions: RelayPermissions = { enforce: plan.isPi, @@ -1155,16 +1255,29 @@ export async function runTurn( onPause: () => pause.pause(), onPausedToolCall: (id) => pause.markPausedToolCall(id), onCreateInteraction: recordPendingInteraction, - onResolveInteraction: (token) => { - const cred = runCredential(request); - if (!cred) return; - if ( - !buildWorkflowReferences(request.runContext?.workflow) - ?.workflow_revision - ) - return; - void resolveInteraction(sessionId, token, () => cred); - }, + onResolveInteraction: resolveInteractionToken, + // Slice 2: record the parkable Claude ACP permission gate (only in keep-alive park mode) so + // the dispatch can resume it live. Fires per pending gate (before the latch) so a parallel + // gate is counted; the single-gate resume records only the FIRST gate's answer target. + onUserApprovalGate: opts.approvalParkMode + ? (info) => { + env.approvalGateCount += 1; + if ( + env.approvalGateCount === 1 && + info.permissionId && + info.toolCallId + ) { + env.parkedApproval = { + gateType: "claude-acp-permission", + permissionId: info.permissionId, + toolCallId: info.toolCallId, + toolName: info.toolName, + args: info.args, + interactionToken: info.interactionToken, + }; + } + } + : undefined, }); // Resolve the ONE client-tool seam both delivery paths share. The correlation index is wired @@ -1193,12 +1306,45 @@ export async function runTurn( ); } - // Race the prompt against the pause signal: on a HITL pause the prompt either resolves with a - // cancelled stop reason or never resolves at all; either way the pause signal ends the turn. - const promptPromise = Promise.resolve( - env.session.prompt([{ type: "text", text: turnText }]), - ); - promptPromise.catch(() => {}); + // The prompt promise this turn races against the pause signal. A normal/continuation turn + // sends a fresh prompt; a live approval resume answers the parked gate on the SAME session and + // continues the ORIGINAL, still-pending prompt promise (the tool then runs with its original + // byte-exact args). Either way, on a HITL pause the prompt resolves cancelled or never + // resolves, and the pause signal ends the turn. + let promptPromise: Promise; + if (opts.resume) { + // The new (resume) turn owns streaming + tracing; the environment is already wired to route + // continued events into this turn's sink (env.currentTurn was set above). Seed this run's + // trace with the parked tool call so the completing `tool_call_update` closes it and the FE + // approval part flips to output-available even if the adapter re-announces nothing. Then + // answer the gate on the live session — the original prompt continues from here. + run.handleUpdate({ + sessionUpdate: "tool_call", + toolCallId: opts.resume.toolCallId, + title: opts.resume.toolName, + kind: opts.resume.toolName, + rawInput: opts.resume.args, + }); + promptPromise = Promise.resolve(opts.resume.promptPromise); + promptPromise.catch(() => {}); + await env.session.respondPermission( + opts.resume.permissionId, + opts.resume.reply, + ); + // The gate is answered: resolve the durable interaction row (the parked pending row the cold + // path would otherwise resolve via its decision map). The fresh per-turn pause controller + // starts with an EMPTY pausedToolCallIds set, so the resumed call's `tool_call_update` frames + // are no longer suppressed and stream through — the "clear pausedToolCallIds on resume" step. + resolveInteractionToken(opts.resume.interactionToken); + logger( + `[keepalive] resume answered gate reply=${opts.resume.reply} tool=${opts.resume.toolName ?? "?"}`, + ); + } else { + promptPromise = Promise.resolve( + env.session.prompt([{ type: "text", text: turnText }]), + ); + promptPromise.catch(() => {}); + } const raced = await Promise.race([ promptPromise, pause.signal.then(() => PAUSED), @@ -1212,6 +1358,14 @@ export async function runTurn( const stopReason = raced === PAUSED || pause.active ? "paused" : (raced as any)?.stopReason; const result = raced === PAUSED ? undefined : raced; + // A parkable pause this turn: hand the still-pending prompt promise to the parked record so a + // later resume can await the same continuation. (Set after the race so `promptPromise` exists. + // The read is asserted because the onUserApprovalGate callback set the field via an async + // mutation TS's flow analysis cannot see, so it would otherwise narrow the reset to `never`.) + const parkedThisTurn = env.parkedApproval as ParkedApproval | undefined; + if (opts.approvalParkMode && pause.active && parkedThisTurn) { + parkedThisTurn.promptPromise = promptPromise; + } await turn.toolRelay?.stop(); logger(`prompt stopReason=${stopReason}`); diff --git a/services/runner/src/engines/sandbox_agent/acp-interactions.ts b/services/runner/src/engines/sandbox_agent/acp-interactions.ts index c3b0bff327..4409da5caf 100644 --- a/services/runner/src/engines/sandbox_agent/acp-interactions.ts +++ b/services/runner/src/engines/sandbox_agent/acp-interactions.ts @@ -33,6 +33,20 @@ export interface AttachPermissionResponderInput { ) => void; /** Called after a stored decision was successfully forwarded to the harness. */ onResolveInteraction?: (token: string) => void; + /** + * Fires for EVERY Claude ACP permission gate (harness executor) that resolves to + * pendingApproval, BEFORE the single-pause latch. Slice 2 keep-alive uses it to record the + * parked permission id / tool-call id (for a live resume via `respondPermission`) and to count + * how many gates are pending this turn (a multi-gate pause does not park). It never fires for a + * client-tool gate (`pauseClientTool`), so only Claude ACP permission gates can park. + */ + onUserApprovalGate?: (info: { + permissionId: string; + toolCallId: string; + toolName: string | undefined; + args: unknown; + interactionToken: string; + }) => void; } /** Wire ACP permission reverse-RPC into the runner's event stream and responder policy. */ @@ -47,6 +61,7 @@ export function attachPermissionResponder({ onPausedToolCall, onCreateInteraction, onResolveInteraction, + onUserApprovalGate, }: AttachPermissionResponderInput): void { session.onPermissionRequest((req: any) => { void handleRequest(req).catch((err) => { @@ -75,6 +90,16 @@ export function attachPermissionResponder({ id: string, gate: GateDescriptor, ): void => { + // Signal the parkable gate BEFORE the latch so a keep-alive resume can record the pending + // permission id and the multi-gate detector counts every pending gate (not just the first). + const gateToolCallId = stringValue(req?.toolCall?.toolCallId); + onUserApprovalGate?.({ + permissionId: id, + toolCallId: gateToolCallId ?? "", + toolName: gate.toolName, + args: gate.args, + interactionToken: interactionEventId(id, gateToolCallId), + }); if (!latch.tryAcquire()) return; const toolCallId = stringValue(req?.toolCall?.toolCallId); const eventId = interactionEventId(id, toolCallId); diff --git a/services/runner/src/engines/sandbox_agent/session-pool.ts b/services/runner/src/engines/sandbox_agent/session-pool.ts index 7d949cdb0f..d647c797b0 100644 --- a/services/runner/src/engines/sandbox_agent/session-pool.ts +++ b/services/runner/src/engines/sandbox_agent/session-pool.ts @@ -22,6 +22,7 @@ import { type ContentBlock, messageText, } from "../../protocol.ts"; +import { approvalDecisionOf } from "../../responder.ts"; function log(message: string): void { process.stderr.write(`[keepalive] ${message}\n`); @@ -219,6 +220,33 @@ export function priorConversation(request: AgentRunRequest): ChatMessage[] { return messages.slice(); } +/** + * The approval decision (allow/deny) the incoming request carries for a specific parked gate's + * tool-call id, or undefined when the request has no approval envelope for that id. Reuses the + * cold path's `approvalDecisionOf` (responder.ts) to parse the `{approved}` envelope, and matches + * strictly by toolCallId (the parked gate's id) — never by name+args — so a live resume answers + * exactly the gate that parked. An incoming reply for a different id, or a plain user message, + * yields undefined and the dispatch degrades to cold. + */ +export function approvalDecisionForToolCall( + request: AgentRunRequest, + toolCallId: string, +): "allow" | "deny" | undefined { + if (!toolCallId) return undefined; + for (const message of request.messages ?? []) { + const content = message?.content; + if (!Array.isArray(content)) continue; + for (const block of content) { + if (block?.type !== "tool_result" || block.toolCallId !== toolCallId) { + continue; + } + const decision = approvalDecisionOf(block); + if (decision !== undefined) return decision; + } + } + return undefined; +} + /** * True when the request's tail is a fresh user message with text and NOT an approval envelope. * A continuation only takes the live path for a plain new user turn; an approval reply (a @@ -380,10 +408,35 @@ export class SessionPool { } /** - * Return a checked-out (busy) session to idle after a successful continuation: refresh its + * Check out an approval-parked session for a live resume: clear its (longer) approval TTL timer, + * REMOVE it from the map, mark it busy, and return it. Returns undefined when the key is absent + * or not awaiting_approval. Removing it is what makes a racing request safe: the resume turn + * owns the environment exclusively, a duplicate approval or fresh message simply misses the pool + * and runs cold (today's concurrent-request semantics), and no supersede path can destroy the + * environment while it is executing the just-approved tool. The gate is therefore answered at + * most once: only the checkout winner ever holds the parked permission id. After the resume + * turn, `repark` re-inserts only if the slot is still empty (see there). + */ + checkoutApproval(key: string): LiveSession | undefined { + const session = this.sessions.get(key); + if (!session || session.state !== "awaiting_approval") return undefined; + this.clearTimer(session); + this.sessions.delete(key); + session.state = "busy"; + session.lastUsed = Date.now(); + return session; + } + + /** + * Return a checked-out (busy) session to the pool after a completed turn: refresh its * fingerprints + credential epoch and re-arm the TTL timer, keeping the SAME live environment. - * Verifies identity so a session superseded mid-turn (its map slot taken by a racing turn) is - * NOT resurrected — returns false, and the caller tears its now-orphaned environment down. + * Two checkout shapes are handled: + * - `checkoutIdle` left the busy session IN the map: the slot must still hold this exact + * session (a racing turn may have superseded it — never clobber the newer one). + * - `checkoutApproval` REMOVED it from the map: re-insert only if the slot is still EMPTY; + * an occupant is a newer session parked by a racing request and must not be clobbered. + * A destroyed session (e.g. drained by `destroyAll` mid-turn) is never resurrected. + * Returns false when the session cannot return; the caller destroys its orphaned environment. */ repark( session: LiveSession, @@ -393,21 +446,30 @@ export class SessionPool { credentialEpoch: CredentialEpoch; }, ttlMs: number, + state: "idle" | "awaiting_approval" = "idle", ): boolean { - if (this.sessions.get(session.key) !== session) return false; + if (session.state === "destroyed") return false; + const current = this.sessions.get(session.key); + if (current !== undefined && current !== session) return false; + if (current === undefined) { + // Re-inserting a checked-out-and-removed session: respect the cap like `park` does. + if (this.sessions.size >= this.config.poolMax && !this.evictLruIdle()) { + this.logger( + `re-park skipped (pool full, nothing idle to evict) key=${session.key}`, + ); + return false; + } + this.sessions.set(session.key, session); + } this.clearTimer(session); session.configFingerprint = update.configFingerprint; session.historyFingerprint = update.historyFingerprint; session.credentialEpoch = update.credentialEpoch; - session.state = "idle"; + session.state = state; session.lastUsed = Date.now(); - session.ttlTimer = setTimeout(() => { - this.logger(`expire key=${session.key} (idle TTL ${ttlMs}ms)`); - void this.evict(session.key, "expire"); - }, ttlMs); - session.ttlTimer.unref?.(); + this.armTtl(session, ttlMs, state); this.logger( - `park key=${session.key} ttl=${ttlMs}ms (re-park) poolSize=${this.sessions.size}`, + `park key=${session.key} ttl=${ttlMs}ms state=${state} (re-park) poolSize=${this.sessions.size}`, ); return true; } @@ -417,7 +479,11 @@ export class SessionPool { * awaiting_approval session. If nothing evictable frees a slot, the session is NOT parked and * the caller tears it down as today (parking is best-effort). Returns whether it parked. */ - park(input: ParkInput, ttlMs: number): boolean { + park( + input: ParkInput, + ttlMs: number, + state: "idle" | "awaiting_approval" = "idle", + ): boolean { // A supersede/re-park on the same key replaces any prior entry (destroy the old one first). const existing = this.sessions.get(input.key); if (existing) { @@ -439,23 +505,38 @@ export class SessionPool { configFingerprint: input.configFingerprint, historyFingerprint: input.historyFingerprint, credentialEpoch: input.credentialEpoch, - state: "idle", + state, lastUsed: Date.now(), destroy: input.destroy, }; - session.ttlTimer = setTimeout(() => { - this.logger(`expire key=${input.key} (idle TTL ${ttlMs}ms)`); - void this.evict(input.key, "expire"); - }, ttlMs); - // Do not let an idle TTL timer keep the process alive on its own. - session.ttlTimer.unref?.(); + this.armTtl(session, ttlMs, state); this.sessions.set(input.key, session); this.logger( - `park key=${input.key} ttl=${ttlMs}ms poolSize=${this.sessions.size}`, + `park key=${input.key} ttl=${ttlMs}ms state=${state} poolSize=${this.sessions.size}`, ); return true; } + /** + * Arm the TTL reaper on a parked session. An idle park uses the short idle TTL; an approval park + * uses the longer approval TTL and logs `approval-ttl-expire` when it fires so an expired + * approval (which degrades to the cold decision-map path) is greppable. Never lets the timer + * keep the process alive on its own. + */ + private armTtl( + session: LiveSession, + ttlMs: number, + state: "idle" | "awaiting_approval", + ): void { + const label = + state === "awaiting_approval" ? "approval-ttl-expire" : "expire"; + session.ttlTimer = setTimeout(() => { + this.logger(`${label} key=${session.key} (TTL ${ttlMs}ms)`); + void this.evict(session.key, label); + }, ttlMs); + session.ttlTimer.unref?.(); + } + /** * Remove a key and destroy it. Idempotent: a missing key is a no-op (resolves false). * The returned promise resolves once the destroy completed, so a caller that reacquires the diff --git a/services/runner/src/server.ts b/services/runner/src/server.ts index cd36f4b9dd..439f0770fc 100644 --- a/services/runner/src/server.ts +++ b/services/runner/src/server.ts @@ -35,10 +35,12 @@ import { resolveKeepaliveMount, runSandboxAgent, runTurn, + type RunTurnOptions, type SessionEnvironment, } from "./engines/sandbox_agent.ts"; import type { MountCredentials } from "./engines/sandbox_agent/mount.ts"; import { + approvalDecisionForToolCall, computeCredentialEpoch, configFingerprint, credentialEpochValid, @@ -50,6 +52,7 @@ import { SessionPool, tailIsFreshUserMessage, type KeepaliveConfig, + type LiveSession, } from "./engines/sandbox_agent/session-pool.ts"; import { runnerInfo } from "./version.ts"; import { isEntrypoint } from "./entry.ts"; @@ -236,7 +239,7 @@ export interface KeepaliveEngine { request: AgentRunRequest, emit: EmitEvent | undefined, signal: AbortSignal | undefined, - opts: { continuation?: boolean }, + opts: RunTurnOptions, ): Promise; /** * Today's cold path (acquire -> runTurn -> teardown). Used when a request must not park. @@ -351,13 +354,133 @@ export async function runWithKeepalive( const cfgFp = configFingerprint(request); const incomingEpoch = computeCredentialEpoch(request, mountCreds.expiresAt); + // The fingerprint the NEXT request's prior conversation is expected to hash to (slice 1's + // prediction; the same one works for an approval park, whose gated tool_call id the FE folds + // back into the resume request's assistant turn). + const nextHistoryFp = (env: SessionEnvironment): string => + expectedNextHistoryFingerprint( + request.messages ?? [], + env.lastTurnToolCallIds ?? [], + ); + + // Whether a paused turn holds a single, parkable Claude ACP permission gate (slice 2). Only such + // a gate carries a `respondPermission`-answerable id; a Pi relay/builtin gate or a client-tool + // MCP pause never records `parkedApproval`, and more than one pending gate cannot be answered by + // the single-gate resume — both stay on the cold path, logged. + const approvalToPark = ( + env: SessionEnvironment, + result: AgentRunResult, + ): boolean => { + if (result.stopReason !== "paused") return false; + if (!env.parkedApproval) { + klog(`non-claude-gate-no-park key=${key}`); + return false; + } + if ((env.approvalGateCount ?? 0) > 1) { + klog(`multi-gate-no-park key=${key} gates=${env.approvalGateCount}`); + return false; + } + // An approval park waits for the HUMAN, who is still on the page even if the streaming client + // dropped right after the pause frame (plan Q4). So, unlike a normal park, do NOT consult + // clientGone or the abort signal here; the approval TTL bounds the wait and an expiry degrades + // to the cold decision-map path. + return true; + }; + + // A parked prompt that REJECTS while the session sits in awaiting_approval means the harness + // or sandbox died mid-park; the dead session must not occupy a pool slot until the 10-minute + // approval TTL (plan Q4 lists a rejected parked prompt as a teardown trigger). Identity-checked: + // the handler evicts only while THIS exact entry is still parked at the key. A rejection that + // lands after a successful checkout (the resume is in flight and owns the environment; its own + // try/catch handles the failure) or after a supersede is not ours and does nothing. `evict` is + // idempotent through the session's one destroy, so no double-destroy is possible. The promise + // already carries runTurn's swallowing catch, so no unhandled rejection is introduced. + const watchParkedPrompt = (env: SessionEnvironment): void => { + const promptPromise = env.parkedApproval?.promptPromise; + const entry = pool.get(key); + if (!promptPromise || !entry || entry.environment !== env) return; + promptPromise.catch(() => { + const current = pool.get(key); + if (current !== entry || current.state !== "awaiting_approval") return; + klog(`parked-prompt-rejected key=${key}; evict`); + void pool.evict(key, "parked-prompt-rejected"); + }); + }; + + // Park a freshly cold-acquired environment (new pool slot) as approval / idle, or tear it down. + const parkFreshOrDestroy = async ( + env: SessionEnvironment, + result: AgentRunResult, + ): Promise => { + env.clearTurn(); + const input = { + key, + environment: env, + configFingerprint: cfgFp, + historyFingerprint: nextHistoryFp(env), + credentialEpoch: incomingEpoch, + destroy: env.destroy, + }; + if (approvalToPark(env, result)) { + klog( + `park-approval key=${key} tool=${env.parkedApproval?.toolName ?? "?"}`, + ); + if (!pool.park(input, config.approvalTtlMs, "awaiting_approval")) { + await env.destroy(); + } else { + watchParkedPrompt(env); + } + } else if (shouldPark(result, signal, clientGone)) { + if (!pool.park(input, config.ttlMs)) await env.destroy(); + } else { + await env.destroy(); + } + }; + + // Re-park a checked-out pool session (same slot) as approval / idle, or evict it. + const reparkOrEvict = async ( + live: LiveSession, + result: AgentRunResult, + ): Promise => { + const env = live.environment; + env.clearTurn(); + const update = { + configFingerprint: cfgFp, + historyFingerprint: nextHistoryFp(env), + credentialEpoch: incomingEpoch, + }; + if (approvalToPark(env, result)) { + klog( + `park-approval key=${key} tool=${env.parkedApproval?.toolName ?? "?"}`, + ); + if ( + !pool.repark(live, update, config.approvalTtlMs, "awaiting_approval") + ) { + await live.destroy(); + } else { + watchParkedPrompt(env); + } + } else if (shouldPark(result, signal, clientGone)) { + if (!pool.repark(live, update, config.ttlMs)) await live.destroy(); + } else { + await pool.evictIfCurrent( + live, + `no-park:${result.stopReason ?? "failed"}`, + ); + } + }; + const coldAndPark = async (): Promise => { const acq = await engine.acquireEnvironment(request, signal, mountCreds); if (!acq.ok) return { ok: false, error: acq.error }; const env = acq.env; let result: AgentRunResult; try { - result = await engine.runTurn(env, request, emit, signal, {}); + // Park mode on: a Claude ACP permission gate this turn keeps the session alive instead of + // tearing down. A non-parkable pause (Pi relay/builtin, client tool) still destroys as today. + result = await engine.runTurn(env, request, emit, signal, { + approvalParkMode: true, + }); } catch (err) { await env.destroy(); return { @@ -365,27 +488,7 @@ export async function runWithKeepalive( error: String(err instanceof Error ? err.message : err), }; } - // Record/settle pending state (none in slice 1) then clear the sink before parking. - env.clearTurn(); - if (shouldPark(result, signal, clientGone)) { - const parked = pool.park( - { - key, - environment: env, - configFingerprint: cfgFp, - historyFingerprint: expectedNextHistoryFingerprint( - request.messages ?? [], - env.lastTurnToolCallIds ?? [], - ), - credentialEpoch: incomingEpoch, - destroy: env.destroy, - }, - config.ttlMs, - ); - if (!parked) await env.destroy(); - } else { - await env.destroy(); - } + await parkFreshOrDestroy(env, result); return result; }; @@ -413,8 +516,10 @@ export async function runWithKeepalive( klog(`hit-continue key=${key}`); let result: AgentRunResult; try { + // A continuation can itself raise an approval gate, so it runs in park mode too. result = await engine.runTurn(live.environment, request, emit, signal, { continuation: true, + approvalParkMode: true, }); } catch (err) { // A continuation that throws destroys the session and retries once cold. Identity-checked @@ -434,33 +539,86 @@ export async function runWithKeepalive( await pool.evictIfCurrent(live, "continuation-failed"); return coldAndPark(); } - live.environment.clearTurn(); - if (shouldPark(result, signal, clientGone)) { - const reparked = pool.repark( - live, - { - configFingerprint: cfgFp, - historyFingerprint: expectedNextHistoryFingerprint( - request.messages ?? [], - live.environment.lastTurnToolCallIds ?? [], - ), - credentialEpoch: incomingEpoch, + await reparkOrEvict(live, result); + return result; + } + // checkout lost a race; fall through to cold. + } else if (existing && existing.state === "awaiting_approval") { + // Slice 2: an approval-parked session. Only a validated approval decision that matches the + // parked Claude ACP gate resumes it live; anything else evicts and degrades to cold. + const parked = existing.environment.parkedApproval; + const decision = parked + ? approvalDecisionForToolCall(request, parked.toolCallId) + : undefined; + const priorFp = historyFingerprint(priorConversation(request)); + let mismatch: string | undefined; + if (!parked || parked.gateType !== "claude-acp-permission") { + mismatch = "not-claude-gate"; // defensive: only a Claude ACP gate ever parks here + } else if (!decision) { + mismatch = "no-matching-approval"; // fresh user text, or an approval for another id + } else if (cfgFp !== existing.configFingerprint) { + mismatch = "config"; + } else if (priorFp !== existing.historyFingerprint) { + mismatch = "history"; + } else if (!credentialEpochValid(existing.credentialEpoch, incomingEpoch)) { + mismatch = "credentials"; + } + + if (mismatch || !parked || !decision) { + klog( + `approval-mismatch (${mismatch ?? "unknown"}) key=${key}; evict + cold`, + ); + await pool.evict(key, `approval-mismatch:${mismatch ?? "unknown"}`); + return coldAndPark(); + } + + const live = pool.checkoutApproval(key); + if (live) { + const reply = decision === "allow" ? "once" : "reject"; + klog( + `${reply === "once" ? "resume-approve" : "resume-reject"} key=${key} ` + + `tool=${parked.toolName ?? "?"}`, + ); + let result: AgentRunResult; + try { + // Answer the parked gate on the SAME live session; the original prompt continues and this + // (new) turn owns streaming + tracing. The gated tool runs with its original byte-exact + // args — no model re-issues anything, so argument drift/task restart cannot happen. + result = await engine.runTurn(live.environment, request, emit, signal, { + approvalParkMode: true, + resume: { + permissionId: parked.permissionId, + reply, + toolCallId: parked.toolCallId, + toolName: parked.toolName, + args: parked.args, + interactionToken: parked.interactionToken, + promptPromise: parked.promptPromise, }, - config.ttlMs, - ); - if (!reparked) await live.destroy(); - } else { - await pool.evictIfCurrent( - live, - `no-park:${result.stopReason ?? "failed"}`, - ); + }); + } catch (err) { + klog(`evict (resume-threw) key=${key}; retry cold`); + live.environment.clearTurn(); + await pool.evictIfCurrent(live, "resume-threw"); + void err; + return coldAndPark(); + } + if (!result.ok) { + klog(`evict (resume-failed) key=${key}; retry cold`); + live.environment.clearTurn(); + await pool.evictIfCurrent(live, "resume-failed"); + return coldAndPark(); } + await reparkOrEvict(live, result); return result; } // checkout lost a race; fall through to cold. } else if (existing) { - // Busy / awaiting_approval / destroyed: two turns racing one session. Supersede (destroy the - // parked one, cold-start the new turn) — awaited so its teardown cannot overlap our acquire. + // Busy / destroyed: two turns racing one session. Only a checkoutIdle continuation leaves a + // busy entry in the map (checkoutApproval REMOVES its session, so an in-flight approval + // resume can never be found — a duplicate approval misses the pool and runs cold, and its + // environment can never be destroyed by this branch). Supersede — destroy the parked one and + // cold-start — awaited so its teardown cannot overlap our acquire. klog(`evict (supersede-${existing.state}) key=${key}; cold`); await pool.evict(key, `supersede-${existing.state}`); } else { diff --git a/services/runner/tests/unit/session-keepalive-approval.test.ts b/services/runner/tests/unit/session-keepalive-approval.test.ts new file mode 100644 index 0000000000..8ff2ec0dfc --- /dev/null +++ b/services/runner/tests/unit/session-keepalive-approval.test.ts @@ -0,0 +1,1254 @@ +/** + * Slice 2 tests: keep-alive across approval pauses (Claude ACP permission gates only). + * + * Two seams: + * - Dispatch (`runWithKeepalive`) with a fake `KeepaliveEngine` that models a paused turn setting + * `env.parkedApproval`, so the pool/park/resume POLICY is exercised without a live harness. + * - Engine (`acquireEnvironment` / `runTurn`) with a pausable fake harness, so the real park + + * `respondPermission` resume MECHANICS are exercised (the gate is answered on the same live + * session, the held prompt continues, and the post-resume update streams to the new turn). + * + * Run: pnpm test (or: pnpm exec vitest run tests/unit/session-keepalive-approval.test.ts) + */ +import { describe, it, vi } from "vitest"; +import assert from "node:assert/strict"; + +import type { + AgentEvent, + AgentRunRequest, + AgentRunResult, +} from "../../src/protocol.ts"; +import { + runWithKeepalive, + type KeepaliveContext, + type KeepaliveEngine, +} from "../../src/server.ts"; +import { + SessionPool, + type KeepaliveConfig, +} from "../../src/engines/sandbox_agent/session-pool.ts"; +import type { MountCredentials } from "../../src/engines/sandbox_agent/mount.ts"; +import { + acquireEnvironment, + runTurn, + type ParkedApproval, + type SandboxAgentDeps, + type SessionEnvironment, +} from "../../src/engines/sandbox_agent.ts"; +import { TOOL_NOT_EXECUTED_PAUSED } from "../../src/tracing/otel.ts"; + +function flush(): Promise { + return new Promise((resolve) => setImmediate(resolve)); +} + +const auth = { + telemetry: { + exporters: { otlp: { headers: { authorization: "ApiKey run" } } }, + }, +}; + +// ---------------------------------------------------------------------------- // +// Dispatch seam: a fake engine whose runTurn scripts each turn (pause / resume). // +// ---------------------------------------------------------------------------- // + +interface TurnScript { + /** The turn pauses on a single Claude ACP permission gate (records parkedApproval). */ + approvalPause?: { + permissionId: string; + toolCallId: string; + toolName?: string; + /** How many gates pended (>1 = multi-gate; still records the first). Default 1. */ + gates?: number; + }; + /** The turn pauses on a non-Claude gate (Pi relay / client tool): paused, no parkedApproval. */ + nonClaudePause?: boolean; + /** Override the completed result. */ + result?: AgentRunResult; + /** Tool-call ids the turn "emitted" (folded into the park fingerprint). */ + toolCallIds?: string[]; + /** Hold the turn pending until released (models an in-flight resume for the double-answer case). */ + hold?: boolean; +} + +interface DispatchFakeEnv { + id: number; + destroyed: number; + turnsCleared: number; + lastTurnToolCallIds: string[]; + parkedApproval?: ParkedApproval; + approvalGateCount: number; + clearTurn: () => void; + destroyImpl: () => Promise; + destroy: () => Promise; +} + +function makeApprovalEngine( + scripts: TurnScript[] = [], + mountOpts: { expiresAt?: string } = {}, +) { + const calls = { + resolveMount: 0, + acquire: 0, + cold: 0, + turns: [] as Array<{ env: DispatchFakeEnv; opts: any; idx: number }>, + resumes: [] as Array<{ + permissionId: string; + reply: string; + toolCallId: string; + }>, + acquiredEnvs: [] as DispatchFakeEnv[], + /** One control per approvalPause turn: settle the parked prompt promise from the test. */ + promptControls: [] as Array<{ + resolve: (value: unknown) => void; + reject: (err: unknown) => void; + }>, + }; + const holds = new Map void>(); + + let nextEnvId = 1; + const makeEnv = (): DispatchFakeEnv => { + const env: DispatchFakeEnv = { + id: nextEnvId++, + destroyed: 0, + turnsCleared: 0, + lastTurnToolCallIds: [], + parkedApproval: undefined, + approvalGateCount: 0, + clearTurn: () => { + env.turnsCleared += 1; + }, + destroyImpl: async () => { + env.destroyed += 1; + }, + destroy: () => env.destroyImpl(), + }; + return env; + }; + + const signedMount = (): MountCredentials => ({ + region: "us-east-1", + bucket: "b", + prefix: "mounts/proj/mount", + accessKey: "AK", + secretKey: "SK", + expiresAt: mountOpts.expiresAt, + projectId: "proj-1", + }); + + const applyScript = async ( + env: DispatchFakeEnv, + opts: any, + ): Promise => { + const idx = calls.turns.length; + const script = scripts[idx] ?? {}; + // Mirror the real runTurn per-turn reset. + env.parkedApproval = undefined; + env.approvalGateCount = 0; + env.lastTurnToolCallIds = script.toolCallIds ?? []; + calls.turns.push({ env, opts, idx }); + if (opts?.resume) { + calls.resumes.push({ + permissionId: opts.resume.permissionId, + reply: opts.resume.reply, + toolCallId: opts.resume.toolCallId, + }); + } + if (script.hold) { + await new Promise((resolve) => holds.set(idx, resolve)); + } + if (script.approvalPause) { + env.approvalGateCount = script.approvalPause.gates ?? 1; + // The held original prompt: pending until the test settles it (mirrors the real Claude + // prompt that never resolves on an unanswered gate). Carries the same swallowing catch + // runTurn attaches, so a test-driven rejection is never unhandled. + const promptPromise = new Promise((resolve, reject) => { + calls.promptControls.push({ resolve, reject }); + }); + promptPromise.catch(() => {}); + env.parkedApproval = { + gateType: "claude-acp-permission", + permissionId: script.approvalPause.permissionId, + toolCallId: script.approvalPause.toolCallId, + toolName: script.approvalPause.toolName, + args: {}, + interactionToken: script.approvalPause.toolCallId, + promptPromise, + }; + return { ok: true, stopReason: "paused" }; + } + if (script.nonClaudePause) return { ok: true, stopReason: "paused" }; + return script.result ?? { ok: true, output: "ok", stopReason: "complete" }; + }; + + const engine: KeepaliveEngine = { + async resolveKeepaliveMount(_request) { + calls.resolveMount += 1; + return signedMount(); + }, + async acquireEnvironment(_request, _signal, _presigned) { + calls.acquire += 1; + const env = makeEnv(); + calls.acquiredEnvs.push(env); + return { ok: true, env: env as unknown as SessionEnvironment }; + }, + async runTurn(env, _request, _emit, _signal, opts) { + return applyScript(env as unknown as DispatchFakeEnv, opts); + }, + async runCold(_request, _emit, _signal, _presigned) { + calls.cold += 1; + return { ok: true, output: "cold", stopReason: "complete" }; + }, + }; + + return { + engine, + calls, + releaseHold: (idx: number) => holds.get(idx)?.(), + }; +} + +function makeCtx( + engine: KeepaliveEngine, + overrides: Partial = {}, + clientGone?: () => boolean, +) { + const config: KeepaliveConfig = { + enabled: true, + ttlMs: 60_000, + approvalTtlMs: 600_000, + poolMax: 8, + ...overrides, + }; + const pool = new SessionPool( + { poolMax: config.poolMax }, + () => {}, + ); + return { engine, pool, config, clientGone } satisfies KeepaliveContext; +} + +const POOL_KEY = "proj-1:s1"; + +/** A turn that pauses on a Claude ACP permission gate for tool-call `tc-gate`. */ +function pauseTurn(sessionId = "s1"): AgentRunRequest { + return { + harness: "claude", + model: "m1", + sessionId, + ...auth, + messages: [{ role: "user", content: "do X" }], + }; +} + +/** The FE's approval resume: the gated assistant tool_call plus the {approved} envelope. */ +function approveResume( + approved = true, + overrides: Partial = {}, +): AgentRunRequest { + return { + harness: "claude", + model: "m1", + sessionId: "s1", + ...auth, + messages: [ + { role: "user", content: "do X" }, + { + role: "assistant", + content: [ + { type: "tool_call", toolCallId: "tc-gate", toolName: "commit" }, + ], + }, + { + role: "user", + content: [ + { type: "tool_result", toolCallId: "tc-gate", output: { approved } }, + ], + }, + ], + ...overrides, + }; +} + +function captureStderr() { + const lines: string[] = []; + const orig = process.stderr.write.bind(process.stderr); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (process.stderr as any).write = (chunk: any) => { + lines.push(String(chunk)); + return true; + }; + return { + lines, + restore: () => { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (process.stderr as any).write = orig; + }, + }; +} + +describe("runWithKeepalive: approval park + resume", () => { + it("parks a paused Claude gate in awaiting_approval, then answers it live on the resume (approve)", async () => { + const { engine, calls } = makeApprovalEngine([ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + }, + toolCallIds: ["tc-gate"], + }, + ]); + const ctx = makeCtx(engine); + + const r1 = await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + assert.equal(r1.stopReason, "paused"); + assert.equal(ctx.pool.size(), 1, "the paused Claude gate parked"); + assert.equal(ctx.pool.get(POOL_KEY)!.state, "awaiting_approval"); + assert.equal( + calls.acquiredEnvs[0].destroyed, + 0, + "the parked session is kept alive, not destroyed", + ); + + const r2 = await runWithKeepalive( + approveResume(true), + undefined, + undefined, + ctx, + ); + assert.equal(r2.ok, true); + assert.equal(calls.acquire, 1, "the resume did NOT re-acquire"); + assert.equal(calls.resumes.length, 1, "the gate is answered exactly once"); + assert.equal(calls.resumes[0].permissionId, "perm-1"); + assert.equal( + calls.resumes[0].reply, + "once", + "approve -> respondPermission once", + ); + assert.equal(calls.resumes[0].toolCallId, "tc-gate"); + assert.equal( + calls.turns[1].env, + calls.turns[0].env, + "the resume ran on the SAME live environment", + ); + assert.equal( + ctx.pool.get(POOL_KEY)!.state, + "idle", + "a completing resume re-parks idle", + ); + }); + + it("answers a denied gate live with reject on the resume", async () => { + const { engine, calls } = makeApprovalEngine([ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + }, + toolCallIds: ["tc-gate"], + }, + ]); + const ctx = makeCtx(engine); + await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + const r2 = await runWithKeepalive( + approveResume(false), + undefined, + undefined, + ctx, + ); + assert.equal(r2.ok, true); + assert.equal(calls.resumes.length, 1); + assert.equal( + calls.resumes[0].reply, + "reject", + "deny -> respondPermission reject", + ); + }); + + it("logs park-approval and resume-approve/reject", async () => { + const cap = captureStderr(); + try { + const { engine } = makeApprovalEngine([ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + }, + toolCallIds: ["tc-gate"], + }, + ]); + const ctx = makeCtx(engine); + await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + await runWithKeepalive(approveResume(true), undefined, undefined, ctx); + assert.ok(cap.lines.some((l) => l.includes("park-approval"))); + assert.ok(cap.lines.some((l) => l.includes("resume-approve"))); + } finally { + cap.restore(); + } + }); +}); + +describe("runWithKeepalive: never-park gate types stay cold", () => { + it("a non-Claude gate pause (Pi relay / client-tool MCP) never parks, tears down cold", async () => { + const cap = captureStderr(); + try { + const { engine, calls } = makeApprovalEngine([{ nonClaudePause: true }]); + const ctx = makeCtx(engine); + const r = await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + assert.equal(r.stopReason, "paused"); + assert.equal( + calls.acquiredEnvs[0].destroyed, + 1, + "no parked approval -> torn down as today", + ); + assert.equal(ctx.pool.size(), 0, "nothing parked"); + assert.ok(cap.lines.some((l) => l.includes("non-claude-gate-no-park"))); + } finally { + cap.restore(); + } + }); + + it("a multi-gate pause (parallel gates) never parks, tears down cold", async () => { + const cap = captureStderr(); + try { + const { engine, calls } = makeApprovalEngine([ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + gates: 2, + }, + toolCallIds: ["tc-gate"], + }, + ]); + const ctx = makeCtx(engine); + const r = await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + assert.equal(r.stopReason, "paused"); + assert.equal( + calls.acquiredEnvs[0].destroyed, + 1, + "the single-gate resume cannot answer >1 gate -> cold", + ); + assert.equal(ctx.pool.size(), 0); + assert.ok(cap.lines.some((l) => l.includes("multi-gate-no-park"))); + } finally { + cap.restore(); + } + }); +}); + +describe("runWithKeepalive: approval resume validation degrades to cold", () => { + async function parkThenResume(resume: AgentRunRequest) { + const { engine, calls } = makeApprovalEngine([ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + }, + toolCallIds: ["tc-gate"], + }, + { result: { ok: true, output: "cold", stopReason: "complete" } }, + ]); + const ctx = makeCtx(engine); + await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + const env1 = calls.acquiredEnvs[0]; + await runWithKeepalive(resume, undefined, undefined, ctx); + return { calls, env1, ctx }; + } + + it("an edited history evicts the parked approval and cold-starts", async () => { + const edited = approveResume(true, { + messages: [ + { role: "user", content: "do X EDITED" }, + { + role: "assistant", + content: [ + { type: "tool_call", toolCallId: "tc-gate", toolName: "commit" }, + ], + }, + { + role: "user", + content: [ + { + type: "tool_result", + toolCallId: "tc-gate", + output: { approved: true }, + }, + ], + }, + ], + }); + const { calls, env1 } = await parkThenResume(edited); + assert.equal(env1.destroyed, 1, "the parked approval session is destroyed"); + assert.equal(calls.acquire, 2, "cold-started a fresh env"); + assert.equal( + calls.resumes.length, + 0, + "no live respondPermission on a mismatch", + ); + }); + + it("a changed config evicts the parked approval and cold-starts", async () => { + const { calls, env1 } = await parkThenResume( + approveResume(true, { model: "m2" }), + ); + assert.equal(env1.destroyed, 1); + assert.equal(calls.acquire, 2); + assert.equal(calls.resumes.length, 0); + }); + + it("an approval for a different toolCallId (no match) evicts to cold", async () => { + const wrongId = approveResume(true, { + messages: [ + { role: "user", content: "do X" }, + { + role: "assistant", + content: [ + { type: "tool_call", toolCallId: "tc-gate", toolName: "commit" }, + ], + }, + { + role: "user", + content: [ + { + type: "tool_result", + toolCallId: "tc-OTHER", + output: { approved: true }, + }, + ], + }, + ], + }); + const { calls, env1 } = await parkThenResume(wrongId); + assert.equal(env1.destroyed, 1); + assert.equal(calls.acquire, 2); + assert.equal(calls.resumes.length, 0); + }); +}); + +describe("runWithKeepalive: approval lifecycle edges", () => { + it("approval TTL expiry destroys the parked session; the next request runs cold", async () => { + vi.useFakeTimers(); + try { + const { engine, calls } = makeApprovalEngine([ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + }, + toolCallIds: ["tc-gate"], + }, + { result: { ok: true, output: "recovered", stopReason: "complete" } }, + ]); + const ctx = makeCtx(engine, { approvalTtlMs: 5000 }); + await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + assert.equal(ctx.pool.get(POOL_KEY)!.state, "awaiting_approval"); + + await vi.advanceTimersByTimeAsync(5001); + assert.equal( + calls.acquiredEnvs[0].destroyed, + 1, + "the expired approval park is destroyed", + ); + assert.equal(ctx.pool.size(), 0); + + const r2 = await runWithKeepalive( + approveResume(true), + undefined, + undefined, + ctx, + ); + assert.equal(r2.ok, true); + assert.equal( + calls.acquire, + 2, + "the next request cold-starts (pool missed)", + ); + assert.equal( + calls.resumes.length, + 0, + "the cold decision-map path answers it, not respondPermission", + ); + } finally { + vi.useRealTimers(); + } + }); + + it("a client disconnect after the pause frame STILL parks the approval (the human is waiting)", async () => { + // Slice 1 destroys a disconnected client's session; an approval park is the exception, because + // the pause happened before the disconnect and the HUMAN who must click is still on the page. + const { engine, calls } = makeApprovalEngine([ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + }, + toolCallIds: ["tc-gate"], + }, + ]); + const ctx = makeCtx(engine, {}, () => true /* clientGone */); + const r = await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + assert.equal(r.stopReason, "paused"); + assert.equal( + calls.acquiredEnvs[0].destroyed, + 0, + "the approval park ignores clientGone", + ); + assert.equal(ctx.pool.get(POOL_KEY)!.state, "awaiting_approval"); + }); + + it("a second identical approval while the first resume is in flight does not double-respond and does not destroy the in-flight env", async () => { + const { engine, calls, releaseHold } = makeApprovalEngine([ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + }, + toolCallIds: ["tc-gate"], + }, + { + hold: true, + result: { ok: true, output: "resumed", stopReason: "complete" }, + }, + { result: { ok: true, output: "cold", stopReason: "complete" } }, + ]); + const ctx = makeCtx(engine); + await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + const env1 = calls.acquiredEnvs[0]; + + // First resume: checks the session OUT of the map (the resume turn owns it) and holds. + const p1 = runWithKeepalive(approveResume(true), undefined, undefined, ctx); + await flush(); + assert.equal( + ctx.pool.get(POOL_KEY), + undefined, + "the checked-out session left the map; a racing request misses", + ); + + // A duplicate approval arrives mid-resume: it must NOT answer the parked gate a second + // time, and it must NOT destroy the environment the resume is executing the tool on. + await runWithKeepalive(approveResume(true), undefined, undefined, ctx); + assert.equal( + env1.destroyed, + 0, + "the in-flight resume environment stays alive through the racing request", + ); + releaseHold(1); + await p1; + + assert.equal( + calls.resumes.length, + 1, + "respondPermission-equivalent resume happened exactly once", + ); + assert.equal( + calls.turns[2]?.opts?.resume, + undefined, + "the duplicate ran as a cold turn, not a second live resume", + ); + // The duplicate's cold turn completed and parked a NEWER session into the slot; the resumed + // env must not clobber it — it is destroyed only AFTER its turn finished. + assert.equal( + env1.destroyed, + 1, + "the resumed env is destroyed post-turn (occupied slot), never mid-flight", + ); + assert.equal( + ctx.pool.get(POOL_KEY)!.environment, + calls.acquiredEnvs[1] as unknown as SessionEnvironment, + "the newer session parked by the duplicate keeps the slot", + ); + }); + + it("a parked prompt rejection while parked evicts the dead session; the next request runs cold", async () => { + const cap = captureStderr(); + try { + const { engine, calls } = makeApprovalEngine([ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + }, + toolCallIds: ["tc-gate"], + }, + { result: { ok: true, output: "cold", stopReason: "complete" } }, + ]); + const ctx = makeCtx(engine); + await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + assert.equal(ctx.pool.get(POOL_KEY)!.state, "awaiting_approval"); + + // The sandbox dies mid-park: the held prompt rejects while the session is parked. + calls.promptControls[0].reject(new Error("sandbox died mid-park")); + await flush(); + assert.equal( + calls.acquiredEnvs[0].destroyed, + 1, + "the dead parked session is destroyed promptly, not at the approval TTL", + ); + assert.equal(ctx.pool.size(), 0, "the pool slot is freed"); + assert.ok( + cap.lines.some((l) => l.includes("parked-prompt-rejected")), + "the rejection eviction is greppable", + ); + + const r2 = await runWithKeepalive( + approveResume(true), + undefined, + undefined, + ctx, + ); + assert.equal(r2.ok, true); + assert.equal(calls.acquire, 2, "the next request cold-starts"); + assert.equal(calls.resumes.length, 0, "no live resume on a dead park"); + } finally { + cap.restore(); + } + }); + + it("a prompt rejection AFTER checkout (resume in flight) does nothing (no double-destroy)", async () => { + const { engine, calls, releaseHold } = makeApprovalEngine([ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + }, + toolCallIds: ["tc-gate"], + }, + { + hold: true, + result: { ok: true, output: "resumed", stopReason: "complete" }, + }, + ]); + const ctx = makeCtx(engine); + await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + const env1 = calls.acquiredEnvs[0]; + + // The resume checks the session out, then the (already-answered) prompt rejects. + const p1 = runWithKeepalive(approveResume(true), undefined, undefined, ctx); + await flush(); + calls.promptControls[0].reject(new Error("late failure")); + await flush(); + assert.equal( + env1.destroyed, + 0, + "the rejection watcher does not touch a checked-out session (the resume owns it)", + ); + + releaseHold(1); + await p1; + // The slot stayed empty, so the completed resume reparked; nothing was double-destroyed. + assert.equal(env1.destroyed, 0); + assert.equal(ctx.pool.get(POOL_KEY)!.state, "idle"); + assert.equal(calls.resumes.length, 1); + }); +}); + +// ---------------------------------------------------------------------------- // +// Engine seam: the real park + respondPermission resume mechanics. // +// ---------------------------------------------------------------------------- // + +interface FakeRun { + id: number; + handled: any[]; + emitted: AgentEvent[]; + /** Tool-call ids announced but not yet completed (models the real otel toolSpans map). */ + open: Set; + /** What settleOpenToolCalls settled: the orphaned-sibling record (F-024). */ + settled: Array<{ id: string; message: string }>; +} + +function pausableHarness(opts: { clientTool?: boolean } = {}) { + const calls = { + permissionReplies: [] as Array<{ id: string; reply: string }>, + runs: [] as FakeRun[], + sandboxDestroyed: 0, + sandboxDisposed: 0, + sessionDestroyed: 0, + logs: [] as string[], + resolvePrompt: undefined as ((value: unknown) => void) | undefined, + promptCount: 0, + }; + const captured = { + onEvent: undefined as ((event: any) => void) | undefined, + onPermissionRequest: undefined as ((req: any) => void) | undefined, + }; + + const session = { + id: "session-1", + onEvent(handler: (event: any) => void) { + captured.onEvent = handler; + }, + onPermissionRequest(handler: (req: any) => void) { + captured.onPermissionRequest = handler; + }, + async respondPermission(id: string, reply: string) { + calls.permissionReplies.push({ id, reply }); + }, + prompt(_blocks: any) { + calls.promptCount += 1; + // Stays pending (Claude never resolves prompt on an unanswered gate) until the test resolves + // it — modelling the ORIGINAL prompt continuing after the parked gate is answered. + return new Promise((resolve) => { + calls.resolvePrompt = resolve; + }); + }, + }; + + const sandbox = { + async createSession(_opts: any) { + return session; + }, + async destroySession(_id: string) { + calls.sessionDestroyed += 1; + }, + async destroySandbox() { + calls.sandboxDestroyed += 1; + }, + async dispose() { + calls.sandboxDisposed += 1; + }, + }; + + const makeRun = (): FakeRun & Record => { + const run: FakeRun & Record = { + id: calls.runs.length + 1, + handled: [], + emitted: [], + open: new Set(), + settled: [], + start() {}, + // Track open tool calls the way the real otel run does, so settleOpenToolCalls is + // meaningful: a tool_call opens an id, a completed/failed tool_call_update closes it. + handleUpdate(update: any) { + run.handled.push(update); + const kind = update?.sessionUpdate; + if (kind === "tool_call" && typeof update.toolCallId === "string") { + run.open.add(update.toolCallId); + } + if ( + kind === "tool_call_update" && + (update.status === "completed" || update.status === "failed") + ) { + run.open.delete(update.toolCallId); + } + }, + emitEvent(event: AgentEvent) { + run.emitted.push(event); + }, + usage() { + return { input: 0, output: 0, total: 0, cost: 0 }; + }, + setUsage() {}, + finish() { + return "assistant output"; + }, + recordError() {}, + output() { + return "assistant output"; + }, + async flush() {}, + events() { + return run.emitted; + }, + settleOpenToolCalls( + isExcluded: (id: string) => boolean, + message: string, + ) { + for (const id of [...run.open]) { + if (isExcluded(id)) continue; + run.open.delete(id); + run.settled.push({ id, message }); + } + }, + traceId() { + return "trace-1"; + }, + }; + calls.runs.push(run); + return run; + }; + + const deps: SandboxAgentDeps = { + log: (message) => calls.logs.push(message), + createLocalCwd: (durable?: string) => durable ?? "/tmp/agenta-fake-cwd", + createDaytonaCwd: (durable?: string) => + durable ?? "/home/sandbox/agenta-fake-cwd", + resolveSkillDirs: () => ({ skills: [], cleanup: () => {} }), + buildDaemonEnv: () => ({}), + resolveDaemonBinary: () => "/bin/sandbox-agent", + buildSandboxProvider: () => ({ provider: true }) as any, + createPersist: () => ({}) as any, + startSandboxAgent: (async () => sandbox) as any, + prepareWorkspace: (async () => ({ cleanup: async () => {} })) as any, + probeCapabilities: async () => + ({ + source: "probed", + capabilities: { + mcpTools: true, + toolCalls: true, + usage: true, + streamingDeltas: true, + }, + }) as any, + applyModel: async (_session, model) => model ?? "resolved-model", + createOtel: (() => makeRun()) as any, + startToolRelay: (() => ({ stop: async () => {} })) as any, + localRelayHost: (() => "local-relay-host") as any, + sandboxRelayHost: (() => "sandbox-relay-host") as any, + // A permission gate pends approval (needs a human); a client-tool gate pends via onClientTool. + responderFactory: () => ({ + async onPermission() { + return { kind: "pendingApproval" } as const; + }, + async onClientTool() { + return opts.clientTool + ? ({ kind: "pendingApproval" } as const) + : ({ kind: "deny" } as const); + }, + }), + }; + + return { calls, deps, captured }; +} + +const engineReq: AgentRunRequest = { + harness: "claude", + messages: [{ role: "user", content: "do X" }], +}; + +function updateEvent(update: Record) { + return { payload: { update } }; +} + +describe("runTurn: real approval park + respondPermission resume", () => { + it("parks a Claude ACP gate (session alive), then answers it live and streams the continuation", async () => { + const { calls, deps, captured } = pausableHarness(); + const acquired = await acquireEnvironment(engineReq, deps); + assert.equal(acquired.ok, true); + if (!acquired.ok) return; + const env = acquired.env; + + // Turn 1: the prompt runs, a Claude ACP permission gate fires, the turn pauses. + const p1 = runTurn(env, engineReq, undefined, undefined, { + approvalParkMode: true, + }); + await flush(); + captured.onEvent!( + updateEvent({ + sessionUpdate: "tool_call", + toolCallId: "tc-gate", + title: "commit", + rawInput: { message: "hi" }, + }), + ); + captured.onPermissionRequest!({ + id: "perm-1", + availableReplies: ["once", "reject"], + toolCall: { + toolCallId: "tc-gate", + name: "commit", + rawInput: { message: "hi" }, + }, + }); + await flush(); + const r1 = await p1; + + assert.equal(r1.stopReason, "paused"); + assert.ok(env.parkedApproval, "the parked approval was recorded"); + assert.equal(env.parkedApproval!.gateType, "claude-acp-permission"); + assert.equal(env.parkedApproval!.permissionId, "perm-1"); + assert.equal(env.parkedApproval!.toolCallId, "tc-gate"); + assert.ok( + env.parkedApproval!.promptPromise, + "the held prompt promise is captured", + ); + assert.deepEqual(env.lastTurnToolCallIds, ["tc-gate"]); + assert.equal( + calls.sessionDestroyed, + 0, + "the parked session is NOT destroyed", + ); + assert.equal(calls.sandboxDestroyed, 0); + + // Turn 2 (resume): the dispatch cleared the sink; answer the parked gate live. + env.clearTurn(); + const held = env.parkedApproval!.promptPromise!; + const p2 = runTurn(env, approveResume(true), undefined, undefined, { + approvalParkMode: true, + resume: { + permissionId: "perm-1", + reply: "once", + toolCallId: "tc-gate", + toolName: "commit", + args: { message: "hi" }, + interactionToken: "tc-gate", + promptPromise: held, + }, + }); + await flush(); + assert.deepEqual( + calls.permissionReplies, + [{ id: "perm-1", reply: "once" }], + "the gate was answered on the live session exactly once", + ); + + // The resumed tool completes: its update streams to the NEW run (pausedToolCallIds cleared). + captured.onEvent!( + updateEvent({ + sessionUpdate: "tool_call_update", + toolCallId: "tc-gate", + status: "completed", + content: "committed", + }), + ); + // The held ORIGINAL prompt now resolves (the tool ran with its original args). + calls.resolvePrompt!({ + stopReason: "complete", + usage: { inputTokens: 1, outputTokens: 1 }, + }); + const r2 = await p2; + + assert.equal(r2.ok, true); + assert.equal(r2.stopReason, "complete"); + assert.equal( + calls.promptCount, + 1, + "no new prompt was sent; the original continued", + ); + assert.equal( + env.parkedApproval, + undefined, + "the consumed approval was reset", + ); + assert.equal( + calls.sessionDestroyed, + 0, + "the live session was reused, never destroyed", + ); + + const run2 = calls.runs[1]; + assert.ok( + run2.handled.some( + (u: any) => + u.sessionUpdate === "tool_call" && u.toolCallId === "tc-gate", + ), + "the resume seeded the parked tool call into the new run's trace", + ); + assert.ok( + run2.handled.some( + (u: any) => + u.sessionUpdate === "tool_call_update" && u.toolCallId === "tc-gate", + ), + "the post-resume tool_call_update streamed (not suppressed) into the new run", + ); + + await env.destroy(); + }); + + it("forwards a reject on the resume when the decision is deny", async () => { + const { calls, deps, captured } = pausableHarness(); + const acquired = await acquireEnvironment(engineReq, deps); + assert.equal(acquired.ok, true); + if (!acquired.ok) return; + const env = acquired.env; + + const p1 = runTurn(env, engineReq, undefined, undefined, { + approvalParkMode: true, + }); + await flush(); + captured.onPermissionRequest!({ + id: "perm-1", + availableReplies: ["once", "reject"], + toolCall: { toolCallId: "tc-gate", name: "commit", rawInput: {} }, + }); + await flush(); + await p1; + + env.clearTurn(); + const held = env.parkedApproval!.promptPromise!; + const p2 = runTurn(env, approveResume(false), undefined, undefined, { + approvalParkMode: true, + resume: { + permissionId: "perm-1", + reply: "reject", + toolCallId: "tc-gate", + toolName: "commit", + args: {}, + interactionToken: "tc-gate", + promptPromise: held, + }, + }); + await flush(); + calls.resolvePrompt!({ stopReason: "complete" }); + await p2; + + assert.deepEqual(calls.permissionReplies, [ + { id: "perm-1", reply: "reject" }, + ]); + await env.destroy(); + }); + + it("a client-tool MCP pause is NOT parkable and tears down cold, even in park mode", async () => { + const { calls, deps, captured } = pausableHarness({ clientTool: true }); + const acquired = await acquireEnvironment(engineReq, deps); + assert.equal(acquired.ok, true); + if (!acquired.ok) return; + const env = acquired.env; + + const p1 = runTurn(env, engineReq, undefined, undefined, { + approvalParkMode: true, + }); + await flush(); + // A client-executor gate (spec.kind === "client") routes through pauseClientTool, which never + // fires onUserApprovalGate, so no parkedApproval is recorded and the pause tears down as today. + captured.onPermissionRequest!({ + id: "perm-c", + availableReplies: ["once", "reject"], + toolCall: { + toolCallId: "tc-client", + name: "browser", + spec: { kind: "client", name: "browser" }, + }, + }); + await flush(); + const r1 = await p1; + + assert.equal(r1.stopReason, "paused"); + assert.equal( + env.parkedApproval, + undefined, + "a client-tool pause is not parkable", + ); + assert.equal( + calls.sessionDestroyed, + 1, + "the non-parkable pause destroyed the session, exactly as today", + ); + + await env.destroy(); + }); + + it("the F-024 sibling settle runs on a PARKABLE pause: the sibling settles, the gated call stays open", async () => { + const { calls, deps, captured } = pausableHarness(); + const acquired = await acquireEnvironment(engineReq, deps); + assert.equal(acquired.ok, true); + if (!acquired.ok) return; + const env = acquired.env; + + const p1 = runTurn(env, engineReq, undefined, undefined, { + approvalParkMode: true, + }); + await flush(); + // A latch-loser sibling tool call announced BEFORE the winning gate: it can never execute + // this turn and must be settled with the deterministic paused result, park or no park. + captured.onEvent!( + updateEvent({ + sessionUpdate: "tool_call", + toolCallId: "tc-sib", + title: "read", + rawInput: { path: "a" }, + }), + ); + captured.onEvent!( + updateEvent({ + sessionUpdate: "tool_call", + toolCallId: "tc-gate", + title: "commit", + rawInput: { message: "hi" }, + }), + ); + captured.onPermissionRequest!({ + id: "perm-1", + availableReplies: ["once", "reject"], + toolCall: { + toolCallId: "tc-gate", + name: "commit", + rawInput: { message: "hi" }, + }, + }); + await flush(); + const r1 = await p1; + + assert.equal(r1.stopReason, "paused"); + assert.ok(env.parkedApproval, "the gate parked (park path)"); + const run1 = calls.runs[0]; + assert.deepEqual( + run1.settled, + [{ id: "tc-sib", message: TOOL_NOT_EXECUTED_PAUSED }], + "the orphaned sibling was settled with TOOL_NOT_EXECUTED_PAUSED despite the park", + ); + assert.ok( + run1.open.has("tc-gate"), + "the gated (paused) call itself stays OPEN for the live resume", + ); + assert.equal(calls.sessionDestroyed, 0, "the park kept the session alive"); + + await env.destroy(); + }); + + it("the sibling settle also covers the multi-gate case the dispatch then destroys", async () => { + const { calls, deps, captured } = pausableHarness(); + const acquired = await acquireEnvironment(engineReq, deps); + assert.equal(acquired.ok, true); + if (!acquired.ok) return; + const env = acquired.env; + + const p1 = runTurn(env, engineReq, undefined, undefined, { + approvalParkMode: true, + }); + await flush(); + // Two parallel gates: gate 1 wins the latch and pauses; at pause time gate 2's call is an + // open non-paused sibling, so the UNCONDITIONAL settle covers it even though parkedApproval + // is already set (the exact case an early-return-before-settle would orphan). + captured.onEvent!( + updateEvent({ + sessionUpdate: "tool_call", + toolCallId: "tc-g1", + title: "commit", + }), + ); + captured.onEvent!( + updateEvent({ + sessionUpdate: "tool_call", + toolCallId: "tc-g2", + title: "deploy", + }), + ); + captured.onPermissionRequest!({ + id: "perm-1", + availableReplies: ["once", "reject"], + toolCall: { toolCallId: "tc-g1", name: "commit", rawInput: {} }, + }); + captured.onPermissionRequest!({ + id: "perm-2", + availableReplies: ["once", "reject"], + toolCall: { toolCallId: "tc-g2", name: "deploy", rawInput: {} }, + }); + await flush(); + const r1 = await p1; + + assert.equal(r1.stopReason, "paused"); + assert.equal(env.approvalGateCount, 2, "both pending gates were counted"); + const run1 = calls.runs[0]; + assert.deepEqual( + run1.settled, + [{ id: "tc-g2", message: TOOL_NOT_EXECUTED_PAUSED }], + "the second gate's call settled as a sibling at pause time", + ); + assert.ok(run1.open.has("tc-g1"), "the winning gate's call stays open"); + + // The dispatch refuses to park a multi-gate pause and destroys: the teardown the park + // skipped (destroySession, sandbox teardown) runs through env.destroy(). + await env.destroy(); + assert.equal(calls.sessionDestroyed, 1, "destroy tore the session down"); + assert.equal(calls.sandboxDestroyed, 1); + }); +}); diff --git a/services/runner/tests/unit/session-pool.test.ts b/services/runner/tests/unit/session-pool.test.ts index d4cd182ce2..6030f33310 100644 --- a/services/runner/tests/unit/session-pool.test.ts +++ b/services/runner/tests/unit/session-pool.test.ts @@ -8,6 +8,7 @@ import assert from "node:assert/strict"; import type { AgentRunRequest } from "../../src/protocol.ts"; import { + approvalDecisionForToolCall, computeCredentialEpoch, configFingerprint, credentialEpochValid, @@ -52,6 +53,57 @@ function parkInput(key: string, env = fakeEnv()) { }; } +describe("approvalDecisionForToolCall", () => { + const req = (content: unknown[]): AgentRunRequest => ({ + messages: [ + { role: "user", content: "do X" }, + { + role: "assistant", + content: [ + { type: "tool_call", toolCallId: "tc-1", toolName: "commit" }, + ], + }, + { role: "user", content: content as never }, + ], + }); + + it("returns allow for an {approved:true} envelope matching the gate's toolCallId", () => { + const request = req([ + { type: "text", text: "ok" }, + { type: "tool_result", toolCallId: "tc-1", output: { approved: true } }, + ]); + assert.equal(approvalDecisionForToolCall(request, "tc-1"), "allow"); + }); + + it("returns deny for an {approved:false} envelope", () => { + const request = req([ + { type: "tool_result", toolCallId: "tc-1", output: { approved: false } }, + ]); + assert.equal(approvalDecisionForToolCall(request, "tc-1"), "deny"); + }); + + it("returns undefined for a different toolCallId or a non-approval tool_result", () => { + const other = req([ + { type: "tool_result", toolCallId: "tc-1", output: { approved: true } }, + ]); + assert.equal(approvalDecisionForToolCall(other, "tc-OTHER"), undefined); + const plain = req([ + { type: "tool_result", toolCallId: "tc-1", output: "browser result" }, + ]); + assert.equal(approvalDecisionForToolCall(plain, "tc-1"), undefined); + }); + + it("returns undefined when the tail is a fresh user message (no approval)", () => { + const request: AgentRunRequest = { + messages: [ + { role: "user", content: "do X" }, + { role: "user", content: "changed my mind" }, + ], + }; + assert.equal(approvalDecisionForToolCall(request, "tc-1"), undefined); + }); +}); + describe("readKeepaliveConfig", () => { const KEYS = [ "AGENTA_RUNNER_SESSION_KEEPALIVE", @@ -385,6 +437,139 @@ describe("SessionPool", () => { assert.deepEqual(pool.keys().sort(), ["a", "c"]); }); + it("checkoutApproval REMOVES the session from the map (a racing request misses)", () => { + const pool = new SessionPool(cfg, () => {}); + const { input } = parkInput("k1"); + pool.park(input, 10_000, "awaiting_approval"); + assert.equal(pool.get("k1")!.state, "awaiting_approval"); + // The idle checkout ignores an approval-parked session; the approval checkout takes it out. + assert.equal(pool.checkoutIdle("k1"), undefined); + const live = pool.checkoutApproval("k1"); + assert.ok(live, "the approval-parked session is checked out"); + assert.equal(live!.state, "busy"); + assert.equal( + pool.get("k1"), + undefined, + "the resume turn owns it exclusively; a racing request misses the pool", + ); + // A duplicate approval cannot check the gate out a second time. + assert.equal(pool.checkoutApproval("k1"), undefined); + }); + + it("repark re-inserts a checked-out approval session into an EMPTY slot", () => { + const pool = new SessionPool(cfg, () => {}); + const { input, env } = parkInput("k1"); + pool.park(input, 10_000, "awaiting_approval"); + const live = pool.checkoutApproval("k1")!; + const ok = pool.repark( + live, + { + configFingerprint: "cfg2", + historyFingerprint: "hist2", + credentialEpoch: epoch, + }, + 10_000, + ); + assert.equal(ok, true, "an empty slot accepts the returning session"); + assert.equal( + pool.get("k1"), + live, + "the SAME session object is back in the map", + ); + assert.equal(pool.get("k1")!.state, "idle"); + assert.equal(env.state.destroyed, 0); + }); + + it("repark refuses when a newer session occupies the slot (never clobbers it)", async () => { + const pool = new SessionPool(cfg, () => {}); + const a = parkInput("k1"); + pool.park(a.input, 10_000, "awaiting_approval"); + const live = pool.checkoutApproval("k1")!; + // A racing request parked a NEWER session under the same key while the resume ran. + const b = parkInput("k1"); + pool.park(b.input, 10_000); + const ok = pool.repark( + live, + { + configFingerprint: "cfg2", + historyFingerprint: "hist2", + credentialEpoch: epoch, + }, + 10_000, + ); + assert.equal(ok, false, "the caller must destroy the orphaned resumed env"); + assert.equal( + pool.get("k1")!.environment, + b.env, + "the newer session is untouched", + ); + await Promise.resolve(); + assert.equal(b.env.state.destroyed, 0); + }); + + it("repark never resurrects a destroyed session into an empty slot", async () => { + const pool = new SessionPool(cfg, () => {}); + const { input, env } = parkInput("k1"); + pool.park(input, 10_000); + const live = pool.checkoutIdle("k1")!; + // A /kill drain destroys everything, including the checked-out-but-mapped busy session. + await pool.destroyAll(); + assert.equal(env.state.destroyed, 1); + const ok = pool.repark( + live, + { + configFingerprint: "cfg2", + historyFingerprint: "hist2", + credentialEpoch: epoch, + }, + 10_000, + ); + assert.equal(ok, false, "a destroyed session never returns to the pool"); + assert.equal(pool.size(), 0); + }); + + it("an awaiting_approval session is NEVER the LRU victim at the cap", async () => { + const pool = new SessionPool({ poolMax: 2 }, () => {}); + const a = parkInput("a"); + const b = parkInput("b"); + const c = parkInput("c"); + // a is approval-parked (longer TTL), b is idle. Parking c at the cap must evict b, not a. + pool.park(a.input, 600_000, "awaiting_approval"); + pool.park(b.input, 10_000); + assert.equal(pool.park(c.input, 10_000), true); + await Promise.resolve(); + assert.equal(b.env.state.destroyed, 1, "the idle entry was evicted"); + assert.equal( + a.env.state.destroyed, + 0, + "the awaiting_approval entry is never LRU-evicted", + ); + assert.deepEqual(pool.keys().sort(), ["a", "c"]); + }); + + it("approval TTL expiry destroys the session and logs approval-ttl-expire", async () => { + vi.useFakeTimers(); + try { + const logs: string[] = []; + const pool = new SessionPool(cfg, (m) => logs.push(m)); + const { input, env } = parkInput("k1"); + pool.park(input, 5000, "awaiting_approval"); + await vi.advanceTimersByTimeAsync(5001); + assert.equal( + env.state.destroyed, + 1, + "the expired approval session is destroyed", + ); + assert.equal(pool.size(), 0); + assert.ok( + logs.some((l) => l.includes("approval-ttl-expire")), + "the approval TTL expiry is greppable", + ); + } finally { + vi.useRealTimers(); + } + }); + it("does not park when the pool is full and nothing is idle to evict", async () => { const pool = new SessionPool({ poolMax: 1 }, () => {}); const a = parkInput("a"); From 6610c7ec513398c00312db7f996584747ac57658 Mon Sep 17 00:00:00 2001 From: Mahmoud Mabrouk Date: Wed, 8 Jul 2026 12:40:07 +0200 Subject: [PATCH 2/5] docs(agent-workflows): document the session keep-alive env vars Adds AGENTA_RUNNER_SESSION_KEEPALIVE, _SESSION_TTL_MS, _SESSION_APPROVAL_TTL_MS, and _SESSION_POOL_MAX to the canonical agent env-var list. The flags land across the two keep-alive slices (#5156, #5158); this stacked lane carries the doc so both are in by the time it applies. Claude-Session: https://claude.ai/code/session_01AumZJ9xRd4XYNHThqy4rTv --- .../documentation/running-the-agent.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/docs/design/agent-workflows/documentation/running-the-agent.md b/docs/design/agent-workflows/documentation/running-the-agent.md index 7ed31c2c9a..39bb5c4fcf 100644 --- a/docs/design/agent-workflows/documentation/running-the-agent.md +++ b/docs/design/agent-workflows/documentation/running-the-agent.md @@ -176,6 +176,18 @@ sections). this non-zero auto-stop so a sandbox the runner leaks (a process KILL skips the per-run teardown) self-reaps instead of burning credit. Values below `1` fall back to the default (a `0` would re-disable auto-stop and reintroduce the leak). +- `AGENTA_RUNNER_SESSION_KEEPALIVE`. Gates session keep-alive: after a turn ends, the runner + parks the live harness session and continues it on the next matching message in the same + conversation, instead of cold-replaying the transcript. Default off. Local sandbox only; + requires mount signing (no mount scope means the session never parks). Design: + `docs/design/agent-workflows/projects/session-keepalive/plan.md`. +- `AGENTA_RUNNER_SESSION_TTL_MS`. How long an idle parked session lives before it is + destroyed. Default `60000`. +- `AGENTA_RUNNER_SESSION_APPROVAL_TTL_MS`. How long a session parked on a Claude ACP approval + gate holds its pending permission request open. Default `600000`. Expiry degrades to the + cold decision-map path. +- `AGENTA_RUNNER_SESSION_POOL_MAX`. Maximum parked sessions per runner replica (LRU evicts + idle sessions; busy and awaiting-approval sessions are never evicted). Default `8`. The `runner` container deliberately has no `env_file`. The harness sandbox must not inherit the stack's secrets. The compose block comments explain this From 3a5d80fa43146054ce66cfc506e34fe591cc15b5 Mon Sep 17 00:00:00 2001 From: Mahmoud Mabrouk Date: Wed, 8 Jul 2026 13:56:24 +0200 Subject: [PATCH 3/5] fix(runner): approval resume must not require re-minted credentials to match Every tool approval on the dev box needed approving twice. Root cause: the awaiting_approval dispatch branch required the resume request's config fingerprint and credential epoch to equal the parked session's. But each approval reply is a fresh /run the backend mints with freshly minted short-lived material (gateway/Composio secret values, a per-turn tool-callback bearer), so the credential epoch -- and often the config fingerprint that can embed those tokens -- practically never matched. The parked live session was evicted on every approval, the turn ran cold, and cold-replay argument drift re-prompted the user. Fix: the awaiting_approval branch keeps the approval-decision match, the history fingerprint, and a hard mount-expiry bound (an expired parked mount can no longer write its cwd, so it still evicts to cold), but drops the config and credential-epoch equality checks. The parked process already holds its own credentials baked at acquire time; the resume only delivers the human yes/no. The idle-continuation branch keeps its full validation, with the ambiguous credentials mismatch reason split into credentials-expired vs credentials-rotated for log diagnosis. Claude-Session: https://claude.ai/code/session_01AumZJ9xRd4XYNHThqy4rTv --- .../src/engines/sandbox_agent/session-pool.ts | 41 ++++++- services/runner/src/server.ts | 37 ++++-- .../unit/session-keepalive-approval.test.ts | 110 +++++++++++++++--- .../runner/tests/unit/session-pool.test.ts | 41 +++++++ 4 files changed, 198 insertions(+), 31 deletions(-) diff --git a/services/runner/src/engines/sandbox_agent/session-pool.ts b/services/runner/src/engines/sandbox_agent/session-pool.ts index d647c797b0..257f892129 100644 --- a/services/runner/src/engines/sandbox_agent/session-pool.ts +++ b/services/runner/src/engines/sandbox_agent/session-pool.ts @@ -296,19 +296,50 @@ export function computeCredentialEpoch( }; } +/** + * Whether a parked session's MOUNT credentials have already expired, ignoring the secret material + * hash entirely. This answers only "can the parked environment still write its durable cwd?". + * + * The approval-resume path uses this instead of `credentialEpochValid`: a resume must NOT require + * the resume request's re-minted credentials to MATCH the parked ones (a fresh /run mints fresh + * short-lived material every time, so they practically never match), but an expired mount means + * the parked cwd can no longer be written, so it must still evict to cold. + */ +export function mountCredentialsExpired( + epoch: CredentialEpoch, + now = Date.now(), +): boolean { + return epoch.mountExpiresAtMs !== undefined && now >= epoch.mountExpiresAtMs; +} + +/** + * Why a parked epoch is no longer usable for an incoming request's epoch, or undefined when it + * still is. The two failure modes are distinguished so diagnosis works from logs: + * - `credentials-expired` — the mount credential's lifetime elapsed (time bound). + * - `credentials-rotated` — the resolved secret/tool-auth material changed (a rotated same-slug + * secret, a different tool-callback bearer). + */ +export function credentialEpochMismatch( + parked: CredentialEpoch, + incoming: CredentialEpoch, + now = Date.now(), +): "credentials-expired" | "credentials-rotated" | undefined { + if (mountCredentialsExpired(parked, now)) return "credentials-expired"; + if (parked.secretsHash !== incoming.secretsHash) return "credentials-rotated"; + return undefined; +} + /** * Whether a parked epoch is still valid for an incoming request's epoch. Invalid (evict, cold) - * when the mount credential expired, or the resolved secret/tool-auth material changed. + * when the mount credential expired, or the resolved secret/tool-auth material changed. Thin + * wrapper over `credentialEpochMismatch` for callers that only need the boolean. */ export function credentialEpochValid( parked: CredentialEpoch, incoming: CredentialEpoch, now = Date.now(), ): boolean { - if (parked.mountExpiresAtMs !== undefined && now >= parked.mountExpiresAtMs) { - return false; - } - return parked.secretsHash === incoming.secretsHash; + return credentialEpochMismatch(parked, incoming, now) === undefined; } /** diff --git a/services/runner/src/server.ts b/services/runner/src/server.ts index 439f0770fc..152c938048 100644 --- a/services/runner/src/server.ts +++ b/services/runner/src/server.ts @@ -43,7 +43,8 @@ import { approvalDecisionForToolCall, computeCredentialEpoch, configFingerprint, - credentialEpochValid, + credentialEpochMismatch, + mountCredentialsExpired, expectedNextHistoryFingerprint, historyFingerprint, poolKeyFor, @@ -496,11 +497,16 @@ export async function runWithKeepalive( if (existing && existing.state === "idle") { // Validate the continuation. Any failure evicts and degrades to cold; never fails the turn. const priorFp = historyFingerprint(priorConversation(request)); + // Splits the old ambiguous "credentials" reason into credentials-expired (mount lifetime + // elapsed) vs credentials-rotated (secret/tool-auth material changed) so log diagnosis works. + const credMismatch = credentialEpochMismatch( + existing.credentialEpoch, + incomingEpoch, + ); let mismatch: string | undefined; if (cfgFp !== existing.configFingerprint) mismatch = "config"; else if (priorFp !== existing.historyFingerprint) mismatch = "history"; - else if (!credentialEpochValid(existing.credentialEpoch, incomingEpoch)) - mismatch = "credentials"; + else if (credMismatch) mismatch = credMismatch; else if (!tailIsFreshUserMessage(request)) mismatch = "tail"; if (mismatch) { @@ -544,8 +550,23 @@ export async function runWithKeepalive( } // checkout lost a race; fall through to cold. } else if (existing && existing.state === "awaiting_approval") { - // Slice 2: an approval-parked session. Only a validated approval decision that matches the - // parked Claude ACP gate resumes it live; anything else evicts and degrades to cold. + // Slice 2: an approval-parked session. A validated approval decision that matches the parked + // Claude ACP gate resumes it live; anything else evicts and degrades to cold. + // + // Unlike the idle-continuation branch above, this branch does NOT require the resume request's + // configFingerprint or credential epoch to EQUAL the parked session's. Every approval reply is + // a fresh /run the backend mints carrying freshly minted short-lived material (gateway/Composio + // secret VALUES, a per-turn tool-callback bearer), so the incoming credential epoch — and often + // the config fingerprint, which can embed those per-turn tokens — practically never match the + // parked ones. But the parked live process already holds its OWN resolved credentials baked at + // acquire time; the resume request only delivers the human's yes/no. Re-minted per-turn material + // on the resume says nothing about the parked environment's validity, so matching it against the + // park would evict a perfectly good live session on every approval (the "approve twice" bug). + // + // We keep the checks that DO bound the parked environment: the approval-decision match, the + // history fingerprint (an edited transcript must not continue wrongly), and a hard mount-expiry + // bound — if the parked session's mount credentials are past expiry, its durable cwd can no + // longer be written, so evict to cold. const parked = existing.environment.parkedApproval; const decision = parked ? approvalDecisionForToolCall(request, parked.toolCallId) @@ -556,12 +577,10 @@ export async function runWithKeepalive( mismatch = "not-claude-gate"; // defensive: only a Claude ACP gate ever parks here } else if (!decision) { mismatch = "no-matching-approval"; // fresh user text, or an approval for another id - } else if (cfgFp !== existing.configFingerprint) { - mismatch = "config"; } else if (priorFp !== existing.historyFingerprint) { mismatch = "history"; - } else if (!credentialEpochValid(existing.credentialEpoch, incomingEpoch)) { - mismatch = "credentials"; + } else if (mountCredentialsExpired(existing.credentialEpoch)) { + mismatch = "credentials-expired"; } if (mismatch || !parked || !decision) { diff --git a/services/runner/tests/unit/session-keepalive-approval.test.ts b/services/runner/tests/unit/session-keepalive-approval.test.ts index 8ff2ec0dfc..fc5b1ded24 100644 --- a/services/runner/tests/unit/session-keepalive-approval.test.ts +++ b/services/runner/tests/unit/session-keepalive-approval.test.ts @@ -440,18 +440,24 @@ describe("runWithKeepalive: never-park gate types stay cold", () => { }); describe("runWithKeepalive: approval resume validation degrades to cold", () => { - async function parkThenResume(resume: AgentRunRequest) { - const { engine, calls } = makeApprovalEngine([ - { - approvalPause: { - permissionId: "perm-1", - toolCallId: "tc-gate", - toolName: "commit", + async function parkThenResume( + resume: AgentRunRequest, + mountOpts: { expiresAt?: string } = {}, + ) { + const { engine, calls } = makeApprovalEngine( + [ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + }, + toolCallIds: ["tc-gate"], }, - toolCallIds: ["tc-gate"], - }, - { result: { ok: true, output: "cold", stopReason: "complete" } }, - ]); + { result: { ok: true, output: "cold", stopReason: "complete" } }, + ], + mountOpts, + ); const ctx = makeCtx(engine); await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); const env1 = calls.acquiredEnvs[0]; @@ -491,13 +497,19 @@ describe("runWithKeepalive: approval resume validation degrades to cold", () => ); }); - it("a changed config evicts the parked approval and cold-starts", async () => { - const { calls, env1 } = await parkThenResume( - approveResume(true, { model: "m2" }), + it("an expired parked mount evicts the approval and cold-starts (hard mount-expiry bound)", async () => { + // The parked session's mount credentials are already past expiry: its durable cwd can no longer + // be written, so even a matching decision + history must degrade to cold. + const { calls, env1 } = await parkThenResume(approveResume(true), { + expiresAt: "2000-01-01T00:00:00.000Z", + }); + assert.equal(env1.destroyed, 1, "the expired parked session is destroyed"); + assert.equal(calls.acquire, 2, "cold-started a fresh env"); + assert.equal( + calls.resumes.length, + 0, + "no live respondPermission on an expired mount", ); - assert.equal(env1.destroyed, 1); - assert.equal(calls.acquire, 2); - assert.equal(calls.resumes.length, 0); }); it("an approval for a different toolCallId (no match) evicts to cold", async () => { @@ -529,6 +541,70 @@ describe("runWithKeepalive: approval resume validation degrades to cold", () => }); }); +describe("runWithKeepalive: approval resume ignores re-minted credentials/config", () => { + // The "approve twice" bug: every approval reply is a fresh /run carrying freshly minted + // short-lived material (gateway/Composio secret VALUES, a per-turn tool-callback bearer), so its + // credential epoch — and often its config fingerprint (per-turn tokens embed in it) — never match + // the parked session's. The parked live process already holds its own baked credentials; the + // resume only delivers the human's yes/no, so a mismatch there must NOT evict the live session. + async function parkThenResume(resume: AgentRunRequest) { + const { engine, calls } = makeApprovalEngine([ + { + approvalPause: { + permissionId: "perm-1", + toolCallId: "tc-gate", + toolName: "commit", + }, + toolCallIds: ["tc-gate"], + }, + { result: { ok: true, output: "resumed", stopReason: "complete" } }, + ]); + const ctx = makeCtx(engine); + await runWithKeepalive(pauseTurn(), undefined, undefined, ctx); + const env1 = calls.acquiredEnvs[0]; + const r2 = await runWithKeepalive(resume, undefined, undefined, ctx); + return { calls, env1, ctx, r2 }; + } + + it("resumes LIVE when the resume carries a DIFFERENT credential epoch AND config fingerprint but a matching decision + history", async () => { + // The resume request re-mints a fresh tool-callback bearer (changes both the config fingerprint + // via toolCallback.endpoint and the credential epoch via secrets + toolCallback.authorization). + const { calls, env1, r2 } = await parkThenResume( + approveResume(true, { + toolCallback: { + endpoint: "https://gateway/tools/call", + authorization: "fresh-per-turn-bearer", + }, + secrets: { OPENAI_API_KEY: "sk-freshly-minted" }, + }), + ); + assert.equal(r2.ok, true); + assert.equal( + calls.acquire, + 1, + "no cold re-acquire; the live parked session was reused", + ); + assert.equal(env1.destroyed, 0, "the parked session was NOT evicted"); + assert.equal( + calls.resumes.length, + 1, + "the gate was answered live exactly once (respondPermission)", + ); + assert.equal(calls.resumes[0].reply, "once"); + assert.equal(calls.resumes[0].permissionId, "perm-1"); + }); + + it("a changed model on the resume still resumes live (config fingerprint no longer gates the approval branch)", async () => { + const { calls, env1, r2 } = await parkThenResume( + approveResume(true, { model: "m2" }), + ); + assert.equal(r2.ok, true); + assert.equal(calls.acquire, 1, "no cold re-acquire"); + assert.equal(env1.destroyed, 0, "the parked session was reused"); + assert.equal(calls.resumes.length, 1, "answered live exactly once"); + }); +}); + describe("runWithKeepalive: approval lifecycle edges", () => { it("approval TTL expiry destroys the parked session; the next request runs cold", async () => { vi.useFakeTimers(); diff --git a/services/runner/tests/unit/session-pool.test.ts b/services/runner/tests/unit/session-pool.test.ts index 6030f33310..25a8ebe988 100644 --- a/services/runner/tests/unit/session-pool.test.ts +++ b/services/runner/tests/unit/session-pool.test.ts @@ -11,7 +11,9 @@ import { approvalDecisionForToolCall, computeCredentialEpoch, configFingerprint, + credentialEpochMismatch, credentialEpochValid, + mountCredentialsExpired, expectedNextHistoryFingerprint, historyFingerprint, poolKeyFor, @@ -378,6 +380,45 @@ describe("credential epoch", () => { const incoming = computeCredentialEpoch({ secrets: { A: "2" } }); assert.equal(credentialEpochValid(parked, incoming, Date.now()), false); }); + + it("credentialEpochMismatch splits the reason: expired vs rotated vs none", () => { + const parked = computeCredentialEpoch( + { secrets: { A: "1" } }, + "2026-01-01T00:00:10.000Z", + ); + const same = computeCredentialEpoch({ secrets: { A: "1" } }); + const rotated = computeCredentialEpoch({ secrets: { A: "2" } }); + const before = Date.parse("2026-01-01T00:00:05.000Z"); + const after = Date.parse("2026-01-01T00:00:15.000Z"); + assert.equal(credentialEpochMismatch(parked, same, before), undefined); + assert.equal( + credentialEpochMismatch(parked, same, after), + "credentials-expired", + ); + assert.equal( + credentialEpochMismatch(parked, rotated, before), + "credentials-rotated", + ); + // Expiry takes precedence over a rotation when both hold. + assert.equal( + credentialEpochMismatch(parked, rotated, after), + "credentials-expired", + ); + }); + + it("mountCredentialsExpired checks only the mount lifetime, ignoring the secret hash", () => { + const parked = computeCredentialEpoch( + { secrets: { A: "1" } }, + "2026-01-01T00:00:10.000Z", + ); + const before = Date.parse("2026-01-01T00:00:05.000Z"); + const after = Date.parse("2026-01-01T00:00:15.000Z"); + assert.equal(mountCredentialsExpired(parked, before), false); + assert.equal(mountCredentialsExpired(parked, after), true); + // No expiry recorded => never expired, regardless of the secret material. + const noExpiry = computeCredentialEpoch({ secrets: { A: "1" } }); + assert.equal(mountCredentialsExpired(noExpiry, after), false); + }); }); describe("poolKeyFor", () => { From c948434c6176453745df819d2c514a2f2fc322a7 Mon Sep 17 00:00:00 2001 From: Mahmoud Mabrouk Date: Wed, 8 Jul 2026 14:45:35 +0200 Subject: [PATCH 4/5] fix(runner): address codex review on keep-alive pool Restore time-based run limits in the split turn path (P1), skip the cold retry after a failed live turn already streamed to the client (P2), and await a replaced same-key session's teardown before parking (P2). Rides the stacked approvals lane because the streaming-retry fix edits the slice-2 approval-resume branch; lands with the stack bottom-up into big-agents. Claude-Session: https://claude.ai/code/session_01AumZJ9xRd4XYNHThqy4rTv --- .../src/engines/sandbox_agent/session-pool.ts | 9 +- services/runner/src/server.ts | 103 ++++++++++++++---- .../unit/sandbox-agent-orchestration.test.ts | 70 ++++++------ .../unit/session-keepalive-dispatch.test.ts | 61 ++++++++++- .../runner/tests/unit/session-pool.test.ts | 65 ++++++++++- 5 files changed, 240 insertions(+), 68 deletions(-) diff --git a/services/runner/src/engines/sandbox_agent/session-pool.ts b/services/runner/src/engines/sandbox_agent/session-pool.ts index 257f892129..a4556622fc 100644 --- a/services/runner/src/engines/sandbox_agent/session-pool.ts +++ b/services/runner/src/engines/sandbox_agent/session-pool.ts @@ -510,17 +510,20 @@ export class SessionPool { * awaiting_approval session. If nothing evictable frees a slot, the session is NOT parked and * the caller tears it down as today (parking is best-effort). Returns whether it parked. */ - park( + async park( input: ParkInput, ttlMs: number, state: "idle" | "awaiting_approval" = "idle", - ): boolean { + ): Promise { // A supersede/re-park on the same key replaces any prior entry (destroy the old one first). + // AWAIT the teardown before taking the slot, exactly like `evict`: the replaced session shares + // the SAME durable cwd/mount as the successor, so its unmount/delete must complete BEFORE the + // new session is parked, or the old destroy could unmount the cwd out from under the successor. const existing = this.sessions.get(input.key); if (existing) { this.clearTimer(existing); this.sessions.delete(input.key); - void this.safeDestroy(existing); + await this.safeDestroy(existing); } if (this.sessions.size >= this.config.poolMax && !this.evictLruIdle()) { diff --git a/services/runner/src/server.ts b/services/runner/src/server.ts index 152c938048..be5216311f 100644 --- a/services/runner/src/server.ts +++ b/services/runner/src/server.ts @@ -330,6 +330,20 @@ export async function runWithKeepalive( const { engine, pool, config, clientGone } = ctx; const sessionId = request.sessionId?.trim(); + // Track whether anything reached the client on this streaming edge. A live continuation/resume + // that fails AFTER emitting (a partial answer or an error event) must NOT retry cold: the client + // and persistence already saw the failed live stream, and a following cold answer would duplicate + // it. Only a live turn that emitted NOTHING yet may fall back to a fresh cold turn (today's + // resilience). In buffered mode (`emit` undefined) nothing is ever streamed, so a cold retry is + // always safe. `emit` stays undefined when undefined so `runTurn` keeps buffering. + let emitted = false; + const trackedEmit: EmitEvent | undefined = emit + ? (event) => { + emitted = true; + emit(event); + } + : undefined; + // Eligibility: session-owned + local sandbox. Otherwise never park; run cold as today // (no up-front sign happened, so the cold path signs itself: still exactly once). if (!sessionId || !isLocalSandbox(request)) { @@ -426,13 +440,15 @@ export async function runWithKeepalive( klog( `park-approval key=${key} tool=${env.parkedApproval?.toolName ?? "?"}`, ); - if (!pool.park(input, config.approvalTtlMs, "awaiting_approval")) { + if ( + !(await pool.park(input, config.approvalTtlMs, "awaiting_approval")) + ) { await env.destroy(); } else { watchParkedPrompt(env); } } else if (shouldPark(result, signal, clientGone)) { - if (!pool.park(input, config.ttlMs)) await env.destroy(); + if (!(await pool.park(input, config.ttlMs))) await env.destroy(); } else { await env.destroy(); } @@ -479,7 +495,7 @@ export async function runWithKeepalive( try { // Park mode on: a Claude ACP permission gate this turn keeps the session alive instead of // tearing down. A non-parkable pause (Pi relay/builtin, client tool) still destroys as today. - result = await engine.runTurn(env, request, emit, signal, { + result = await engine.runTurn(env, request, trackedEmit, signal, { approvalParkMode: true, }); } catch (err) { @@ -523,26 +539,49 @@ export async function runWithKeepalive( let result: AgentRunResult; try { // A continuation can itself raise an approval gate, so it runs in park mode too. - result = await engine.runTurn(live.environment, request, emit, signal, { - continuation: true, - approvalParkMode: true, - }); + result = await engine.runTurn( + live.environment, + request, + trackedEmit, + signal, + { + continuation: true, + approvalParkMode: true, + }, + ); } catch (err) { // A continuation that throws destroys the session and retries once cold. Identity-checked // (a racing turn may have superseded this slot and parked its own session — never clobber // it) and awaited (the teardown's unmount must finish before the cold acquire remounts). - klog(`evict (continuation-threw) key=${key}; retry cold`); + // But NOT if the failed turn already streamed to the client: a cold retry would duplicate. live.environment.clearTurn(); await pool.evictIfCurrent(live, "continuation-threw"); + if (emitted) { + klog( + `evict (continuation-threw) key=${key}; already streamed, no retry`, + ); + return { + ok: false, + error: String(err instanceof Error ? err.message : err), + }; + } + klog(`evict (continuation-threw) key=${key}; retry cold`); void err; return coldAndPark(); } if (!result.ok) { // A failed continuation may mean a broken live session: destroy and retry once cold - // (identity-checked + awaited, same as the throw path above). - klog(`evict (continuation-failed) key=${key}; retry cold`); + // (identity-checked + awaited, same as the throw path above). But NOT if the failed turn + // already streamed to the client: return the failure, a cold retry would duplicate. live.environment.clearTurn(); await pool.evictIfCurrent(live, "continuation-failed"); + if (emitted) { + klog( + `evict (continuation-failed) key=${key}; already streamed, no retry`, + ); + return result; + } + klog(`evict (continuation-failed) key=${key}; retry cold`); return coldAndPark(); } await reparkOrEvict(live, result); @@ -603,29 +642,47 @@ export async function runWithKeepalive( // Answer the parked gate on the SAME live session; the original prompt continues and this // (new) turn owns streaming + tracing. The gated tool runs with its original byte-exact // args — no model re-issues anything, so argument drift/task restart cannot happen. - result = await engine.runTurn(live.environment, request, emit, signal, { - approvalParkMode: true, - resume: { - permissionId: parked.permissionId, - reply, - toolCallId: parked.toolCallId, - toolName: parked.toolName, - args: parked.args, - interactionToken: parked.interactionToken, - promptPromise: parked.promptPromise, + result = await engine.runTurn( + live.environment, + request, + trackedEmit, + signal, + { + approvalParkMode: true, + resume: { + permissionId: parked.permissionId, + reply, + toolCallId: parked.toolCallId, + toolName: parked.toolName, + args: parked.args, + interactionToken: parked.interactionToken, + promptPromise: parked.promptPromise, + }, }, - }); + ); } catch (err) { - klog(`evict (resume-threw) key=${key}; retry cold`); + // As in the continuation branch: retry cold only if nothing streamed to the client yet. live.environment.clearTurn(); await pool.evictIfCurrent(live, "resume-threw"); + if (emitted) { + klog(`evict (resume-threw) key=${key}; already streamed, no retry`); + return { + ok: false, + error: String(err instanceof Error ? err.message : err), + }; + } + klog(`evict (resume-threw) key=${key}; retry cold`); void err; return coldAndPark(); } if (!result.ok) { - klog(`evict (resume-failed) key=${key}; retry cold`); live.environment.clearTurn(); await pool.evictIfCurrent(live, "resume-failed"); + if (emitted) { + klog(`evict (resume-failed) key=${key}; already streamed, no retry`); + return result; + } + klog(`evict (resume-failed) key=${key}; retry cold`); return coldAndPark(); } await reparkOrEvict(live, result); diff --git a/services/runner/tests/unit/sandbox-agent-orchestration.test.ts b/services/runner/tests/unit/sandbox-agent-orchestration.test.ts index 4b61c2b429..8192f369af 100644 --- a/services/runner/tests/unit/sandbox-agent-orchestration.test.ts +++ b/services/runner/tests/unit/sandbox-agent-orchestration.test.ts @@ -413,7 +413,10 @@ describe("runSandboxAgent orchestration", () => { "local durable cwd is mounted before first workspace write", ); assert.equal(workspaceCalls, 2, "workspace prep is retried once"); - assert.equal(calls.createSessionOptions.cwd, "/tmp/agenta/mounts/proj-1/mount-1"); + assert.equal( + calls.createSessionOptions.cwd, + "/tmp/agenta/mounts/proj-1/mount-1", + ); assert.equal(cleanupCalls, 1); }); @@ -496,7 +499,11 @@ describe("runSandboxAgent orchestration", () => { "initial mount + one capped runtime remount after ENOTCONN", ); assert.deepEqual(seenMountAccessKeys, ["AK-1", "AK-2"]); - assert.equal(unmountCalls, 1, "cleanup waits for runtime remount before unmount"); + assert.equal( + unmountCalls, + 1, + "cleanup waits for runtime remount before unmount", + ); }); it("skips the durable cwd delete when unmount is not confirmed", async () => { @@ -1018,10 +1025,7 @@ describe("runSandboxAgent orchestration", () => { // The harness-readable env carries a file path, never the bearer itself. assert.equal(env.OTEL_EXPORTER_OTLP_HEADERS, undefined); assert.equal(typeof env.AGENTA_AGENT_OTLP_AUTH_FILE, "string"); - assert.equal( - JSON.stringify(env).includes("reusable-caller-token"), - false, - ); + assert.equal(JSON.stringify(env).includes("reusable-caller-token"), false); }); it("sets Claude Bedrock env and strict selected model pass-through", async () => { @@ -1235,7 +1239,10 @@ describe("runSandboxAgent default ApprovalResponder wiring", () => { isError: true, }, ]); - assert.equal(toolResults?.some((event) => event.id === "tool-a"), false); + assert.equal( + toolResults?.some((event) => event.id === "tool-a"), + false, + ); }); it("settles a latch-loser sibling when read-only permission requests race", async () => { @@ -1685,27 +1692,23 @@ describe("runSandboxAgent default ApprovalResponder wiring", () => { ]); }); }); - -describe("runSandboxAgent run-limits deadline", () => { - // Tiny real-ms limits: the run-limits DI seam overrides the whole resolve step, and the - // integration path needs the real createRunLimits (timers + abort wiring) rather than a fake - // clock, so the windows are set to a few ms and driven by real timers here. +describe("runTurn run-limits deadline (split path)", () => { + // Tiny real-ms limits injected through the run-limits DI seam: the integration path exercises + // the real createRunLimits (timers + trip wiring), so the windows are a few ms and driven by + // real timers. TTFB (5ms) is the shortest, so a silent harness trips it first. const fastLimits = { resolveRunLimits: () => ({ - totalMs: 5, - idleMs: 5, + totalMs: 50, + idleMs: 50, ttfbMs: 5, - toolCallMs: 5, + toolCallMs: 50, }), }; - it("aborts a wedged never-responding run so the existing finally reclaims the sandbox", async () => { - const { calls, deps } = fakeHarness({ - // The harness comes up but the prompt never resolves on its own — a wedged run. Only an - // abort of the run signal (which a tripped deadline fires) can unstick it. - hangPrompt: true, - abortSignalCancelsHungPrompt: true, - }); + it("ends a wedged never-responding turn as an error so the finally reclaims the sandbox", async () => { + // The harness comes up but the prompt never resolves on its own — a wedged run. With no + // deadline it would hold the sandbox forever; the tripped TTFB limit must end the turn. + const { calls, deps } = fakeHarness({ hangPrompt: true }); const result = await runSandboxAgent( { harness: "claude", messages: [{ role: "user", content: "hello" }] }, @@ -1714,25 +1717,25 @@ describe("runSandboxAgent run-limits deadline", () => { { ...deps, ...fastLimits }, ); - // The run RETURNED (did not hang) and the teardown finally ran: sandbox destroyed + disposed. + // The run RETURNED (did not hang) as a failure, and the teardown finally reclaimed the sandbox. + assert.equal(result.ok, false); + if (result.ok) return; + assert.match(result.error ?? "", /first response|deadline|idle|run limit/i); assert.equal(calls.sandboxDestroyed, 1); assert.equal(calls.sandboxDisposed, 1); - assert.equal(calls.runFinished, 1); - // The prompt resolved via the deadline-driven abort, so the turn ends cancelled, not a hang. - if (result.ok) assert.equal(result.stopReason, "cancelled"); }); - it("does NOT abort a turn that paused for human input, even past every deadline window", async () => { - // A gated tool call parks the turn (pause path), which retires the deadlines. With tiny - // limits, a wedged turn WOULD trip almost immediately — so a clean `paused` finish (never a - // deadline `cancelled`) proves the pause exemption held. The prompt hangs; only the local - // park signal ends the turn. Setup mirrors the F-040 park test above (real ApprovalResponder). + it("does NOT trip a turn that paused for human input, even past every deadline window", async () => { + // A gated tool call parks the turn (pause path), which retires the deadlines via notePaused. + // With tiny limits a wedged turn WOULD trip almost immediately — so a clean `paused` finish + // (never a deadline error) proves the pause exemption held. The prompt hangs; only the local + // park signal ends the turn. Setup mirrors the F-040 park test (real ApprovalResponder). const { calls, deps } = (() => { const { calls, deps } = fakeHarness({ emitPermission: true, hangPrompt: true, }); - delete deps.responderFactory; // engine ApprovalResponder -> pauses (effective ask, no decision) + delete deps.responderFactory; // engine ApprovalResponder -> pauses (ask, no stored decision) return { calls, deps }; })(); @@ -1751,9 +1754,8 @@ describe("runSandboxAgent run-limits deadline", () => { assert.equal(result.ok, true); if (!result.ok) return; - // Paused, not cancelled: the pause path — not a deadline abort — ended the turn. + // Paused, not a deadline error: the pause path — not a tripped limit — ended the turn. assert.equal(result.stopReason, "paused"); - // The finally still reclaimed the sandbox on the pause path. assert.equal(calls.sandboxDestroyed, 1); }); }); diff --git a/services/runner/tests/unit/session-keepalive-dispatch.test.ts b/services/runner/tests/unit/session-keepalive-dispatch.test.ts index 12f0d06fde..8790822d4f 100644 --- a/services/runner/tests/unit/session-keepalive-dispatch.test.ts +++ b/services/runner/tests/unit/session-keepalive-dispatch.test.ts @@ -10,7 +10,12 @@ import { describe, it } from "vitest"; import assert from "node:assert/strict"; -import type { AgentRunRequest, AgentRunResult } from "../../src/protocol.ts"; +import type { + AgentEvent, + AgentRunRequest, + AgentRunResult, + EmitEvent, +} from "../../src/protocol.ts"; import { runWithKeepalive, type KeepaliveContext, @@ -35,6 +40,9 @@ interface EngineOptions { turnResults?: AgentRunResult[]; /** Per-call emitted tool-call ids the fake runTurn records on the env (by call index). */ turnToolCallIds?: string[][]; + /** Per-call: when true, the fake runTurn streams one event through `emit` before returning + * its result (models a live turn that reached the client before failing). */ + turnEmits?: boolean[]; } interface FakeEnv { @@ -97,10 +105,13 @@ function makeEngine(options: EngineOptions = {}) { const runOneTurn = async ( env: FakeEnv, continuation: boolean, + emit: EmitEvent | undefined, ): Promise => { const idx = calls.turns.length; calls.turns.push({ env, continuation }); env.lastTurnToolCallIds = options.turnToolCallIds?.[idx] ?? []; + if (options.turnEmits?.[idx]) + emit?.({ type: "message_delta", id: "d1", delta: "partial" }); return ( options.turnResults?.[idx] ?? { ok: true, @@ -122,8 +133,8 @@ function makeEngine(options: EngineOptions = {}) { calls.acquiredEnvs.push(env); return { ok: true, env: env as unknown as SessionEnvironment }; }, - async runTurn(env, _request, _emit, _signal, opts) { - return runOneTurn(env as unknown as FakeEnv, !!opts.continuation); + async runTurn(env, _request, emit, _signal, opts) { + return runOneTurn(env as unknown as FakeEnv, !!opts.continuation, emit); }, async runCold(_request, _emit, _signal, presignedMount) { calls.cold += 1; @@ -589,6 +600,50 @@ describe("runWithKeepalive: races and failures", () => { assert.equal(ctx.pool.size(), 1, "the cold retry re-parked"); }); + it("does NOT retry cold when a failed continuation already streamed to the client", async () => { + // Streaming edge (emit defined): the continuation emits a partial answer, then fails. A cold + // retry would push a second, successful answer after the client already saw the failed live + // stream — a duplicate. The broken session is still evicted, but the failure is returned as-is. + const { engine, calls } = makeEngine({ + turnResults: [ + { ok: true, stopReason: "complete" }, + { ok: false, error: "session died mid-stream" }, + { ok: true, output: "recovered", stopReason: "complete" }, // must NOT run + ], + turnEmits: [false, true], // the continuation streams before failing + }); + const ctx = makeCtx(engine); + const seen: AgentEvent[] = []; + const emit: EmitEvent = (event) => { + seen.push(event); + }; + await runWithKeepalive(turn1(), emit, undefined, ctx); + const env1 = calls.acquiredEnvs[0]; + const r2 = await runWithKeepalive(turn2(), emit, undefined, ctx); + + assert.equal( + env1.destroyed, + 1, + "the broken live session is still destroyed", + ); + assert.equal( + calls.acquire, + 1, + "no cold reacquire after the client already streamed", + ); + assert.equal(calls.cold, 0, "no cold retry"); + assert.equal( + r2.ok, + false, + "the failure is returned, not masked by a cold answer", + ); + assert.equal( + seen.filter((e) => e.type === "message_delta").length, + 1, + "the client saw exactly the one live delta — no duplicated cold answer", + ); + }); + it("eviction before a cold reacquire is AWAITED (teardown cannot overlap the new acquire)", async () => { // The mismatch path evicts the old env (unmounting the shared durable cwd) and then // cold-starts the same key. The acquire must observe the destroy as already complete. diff --git a/services/runner/tests/unit/session-pool.test.ts b/services/runner/tests/unit/session-pool.test.ts index 25a8ebe988..8f8514bdd9 100644 --- a/services/runner/tests/unit/session-pool.test.ts +++ b/services/runner/tests/unit/session-pool.test.ts @@ -434,10 +434,10 @@ describe("poolKeyFor", () => { describe("SessionPool", () => { const cfg = { poolMax: 2 }; - it("park then checkoutIdle returns the same session (busy) and clears the timer", () => { + it("park then checkoutIdle returns the same session (busy) and clears the timer", async () => { const pool = new SessionPool(cfg, () => {}); const { input, env } = parkInput("k1"); - assert.equal(pool.park(input, 10_000), true); + assert.equal(await pool.park(input, 10_000), true); assert.equal(pool.size(), 1); const live = pool.checkoutIdle("k1"); assert.ok(live); @@ -471,7 +471,7 @@ describe("SessionPool", () => { pool.checkoutIdle("a"); pool.park(b.input, 10_000); // Pool now holds a(busy) + b(idle); parking c must evict b (the only idle), not a. - assert.equal(pool.park(c.input, 10_000), true); + assert.equal(await pool.park(c.input, 10_000), true); await Promise.resolve(); assert.equal(b.env.state.destroyed, 1, "the idle entry was evicted"); assert.equal(a.env.state.destroyed, 0, "the busy entry is never evicted"); @@ -577,7 +577,7 @@ describe("SessionPool", () => { // a is approval-parked (longer TTL), b is idle. Parking c at the cap must evict b, not a. pool.park(a.input, 600_000, "awaiting_approval"); pool.park(b.input, 10_000); - assert.equal(pool.park(c.input, 10_000), true); + assert.equal(await pool.park(c.input, 10_000), true); await Promise.resolve(); assert.equal(b.env.state.destroyed, 1, "the idle entry was evicted"); assert.equal( @@ -618,7 +618,7 @@ describe("SessionPool", () => { pool.checkoutIdle("a"); // busy, not evictable const b = parkInput("b"); assert.equal( - pool.park(b.input, 10_000), + await pool.park(b.input, 10_000), false, "park is best-effort; refused when full", ); @@ -717,6 +717,61 @@ describe("SessionPool", () => { ); }); + it("park AWAITS the replaced same-key session's teardown before taking the slot", async () => { + // Two cold turns for the same key finish near each other: the second park replaces the first. + // Both share the same durable cwd/mount, so the first's destroy (unmount/delete) must complete + // BEFORE the successor is parked, or it could unmount the cwd out from under the new session. + const pool = new SessionPool({ poolMax: 4 }, () => {}); + // A's destroy is gated: it does not resolve until we release it, standing in for a slow unmount. + let releaseADestroy: (() => void) | undefined; + const aState = { destroyed: 0 }; + const aEnv = { + state: aState, + destroy: async () => { + await new Promise((resolve) => { + releaseADestroy = resolve; + }); + aState.destroyed += 1; + }, + }; + const a = parkInput("k1", aEnv); + await pool.park(a.input, 10_000); + + const b = parkInput("k1"); + // The replacing park cannot resolve while A's teardown is still in flight. + const parked = pool.park(b.input, 10_000); + let settled = false; + void parked.then(() => { + settled = true; + }); + await Promise.resolve(); + assert.equal( + settled, + false, + "park is pending until the old destroy finishes", + ); + assert.equal(aState.destroyed, 0, "A's destroy has not completed yet"); + assert.equal( + pool.get("k1"), + undefined, + "the successor is NOT parked while the shared cwd is still being unmounted", + ); + + // Release A's teardown: only now does the successor take the slot. + releaseADestroy?.(); + assert.equal(await parked, true); + assert.equal( + aState.destroyed, + 1, + "the replaced session was destroyed first", + ); + assert.equal( + pool.get("k1")!.environment, + b.env, + "the successor holds the slot only after the old teardown completed", + ); + }); + it("destroyAll drains every parked session", async () => { const pool = new SessionPool({ poolMax: 8 }, () => {}); const envs = ["a", "b", "c"].map((k) => { From 7ef57be2490ead5da683db9babeffeb1186bd891 Mon Sep 17 00:00:00 2001 From: Mahmoud Mabrouk Date: Wed, 8 Jul 2026 15:36:57 +0200 Subject: [PATCH 5/5] refactor(runner): clarify the session event-demux (rename env2, extract helpers) Readability-only, no behavior change. Renames the SessionEnvironment `env2` to `environment` and extracts the two session-lifetime listeners into documented `routeSessionEventToActiveTurn` / `routePermissionRequestToActiveTurn` helpers, so `acquireEnvironment` reads as clear steps and the demux data flow is obvious. Addresses Mahmoud's #5156 review note that the event handling was hard to read. 723 tests unchanged, typecheck clean, reviewer confirmed no semantic drift. Lands on the stacked approvals lane (top of the keep-alive stack) because the working tree is the stack's applied state and GitButler attributes the edit to the top lane; it rides into big-agents with the stack. Claude-Session: https://claude.ai/code/session_01AumZJ9xRd4XYNHThqy4rTv --- services/runner/src/engines/sandbox_agent.ts | 274 +++++++++++-------- 1 file changed, 165 insertions(+), 109 deletions(-) diff --git a/services/runner/src/engines/sandbox_agent.ts b/services/runner/src/engines/sandbox_agent.ts index 32c91b1667..1d54ad08ef 100644 --- a/services/runner/src/engines/sandbox_agent.ts +++ b/services/runner/src/engines/sandbox_agent.ts @@ -634,7 +634,7 @@ export async function acquireEnvironment( // so its handler is torn down deterministically and cannot write a result after the turn ends. const mcpAbort = new AbortController(); - const env2: SessionEnvironment = { + const environment: SessionEnvironment = { plan, logger, deps, @@ -672,67 +672,74 @@ export async function acquireEnvironment( clearTurn: () => {}, }; - env2.clearTurn = () => { - env2.currentTurn = undefined; + environment.clearTurn = () => { + environment.currentTurn = undefined; }; // The one complete, idempotent teardown — the same steps the old per-run `finally` ran, in the // same order. Every resource is null-checked, so it is safe after a partial acquire and safe to // call twice (the guard returns on a second call). It must never throw. - env2.destroy = async () => { - if (env2.destroyed) return; - env2.destroyed = true; - await env2.runtimeRemount?.catch(() => {}); - if (env2.sandbox) inFlightSandboxes.delete(env2.sandbox); - await env2.currentTurn?.toolRelay?.stop().catch(() => {}); + environment.destroy = async () => { + if (environment.destroyed) return; + environment.destroyed = true; + await environment.runtimeRemount?.catch(() => {}); + if (environment.sandbox) inFlightSandboxes.delete(environment.sandbox); + await environment.currentTurn?.toolRelay?.stop().catch(() => {}); // Teardown backstop: destroy any in-flight loopback `tools/call` before closing the server. - env2.mcpAbort.abort(); - await env2.closeToolMcp?.().catch(() => {}); + environment.mcpAbort.abort(); + await environment.closeToolMcp?.().catch(() => {}); // Send a graceful `session/cancel` BEFORE tearing down the daemon (the ACP child process // leak, dev-box incident 2026-07-06): destroySandbox hard-kills the sandbox-agent server but // does not cascade to the ACP adapter subprocess it spawned, which then reparents to PID 1 // and never exits. Skip if the pause path already sent it (`sessionDestroyRequested`). - if (env2.session && !env2.sessionDestroyRequested) - await env2.sandbox?.destroySession?.(env2.session.id).catch(() => {}); - await env2.sandbox?.destroySandbox().catch(() => {}); - await env2.sandbox?.dispose().catch(() => {}); + if (environment.session && !environment.sessionDestroyRequested) + await environment.sandbox + ?.destroySession?.(environment.session.id) + .catch(() => {}); + await environment.sandbox?.destroySandbox().catch(() => {}); + await environment.sandbox?.dispose().catch(() => {}); // Unmount the durable cwd BEFORE removing the dir: data lives in the store, only the host // mountpoint is torn down. If unmount is not CONFIRMED gone, skip the delete: rmSync must // never run against a possibly-live FUSE mount into the durable store. - if (env2.mountedCwd) { - env2.durableCwdSafeToDelete = await ( - env2.deps.unmountStorage ?? unmountStorage - )(env2.mountedCwd, { log }).catch(() => false); + if (environment.mountedCwd) { + environment.durableCwdSafeToDelete = await ( + environment.deps.unmountStorage ?? unmountStorage + )(environment.mountedCwd, { log }).catch(() => false); } - if (!env2.durableCwdSafeToDelete) { + if (!environment.durableCwdSafeToDelete) { logger( `durable cwd unmount not confirmed, skipping workspace cleanup cwd=${plan.cwd}`, ); } else { - await env2.workspace?.cleanup().catch(() => {}); + await environment.workspace?.cleanup().catch(() => {}); } // The per-run Agenta agent dir (skills isolation) is throwaway; remove it too. - if (env2.runAgentDir) - rmSync(env2.runAgentDir, { recursive: true, force: true }); + if (environment.runAgentDir) + rmSync(environment.runAgentDir, { recursive: true, force: true }); // Backstop: the extension deletes this on read; remove it here too in case the harness never // started (or crashed before reading it), so the bearer never lingers. - if (env2.otlpAuthFilePath) rmSync(env2.otlpAuthFilePath, { force: true }); + if (environment.otlpAuthFilePath) + rmSync(environment.otlpAuthFilePath, { force: true }); // Remove the per-run skills temp root the materializer created (success or error). plan.skillsCleanup(); }; - // --- local durable cwd mount helpers (session-scoped, close over env2) ------ // + // --- local durable cwd mount helpers (session-scoped, close over environment) ------ // const mountLocalDurableCwd = async (reason: string): Promise => { - if (!env2.mountCreds || plan.isDaytona) return false; + if (!environment.mountCreds || plan.isDaytona) return false; logger( `local durable cwd mount (${reason}) session=${sessionForMount} cwd=${plan.cwd}`, ); if ( - await (deps.mountStorage ?? mountStorage)(plan.cwd, env2.mountCreds, { - log: logger, - }) + await (deps.mountStorage ?? mountStorage)( + plan.cwd, + environment.mountCreds, + { + log: logger, + }, + ) ) { - env2.mountedCwd = plan.cwd; + environment.mountedCwd = plan.cwd; return true; } return false; @@ -764,17 +771,21 @@ export async function acquireEnvironment( ); return false; } - env2.mountCreds = fresh; + environment.mountCreds = fresh; return mountLocalDurableCwd("enotconn-retry"); }; const remountLocalCwdAfterRuntimeEnotconn = (event: unknown): void => { - if (plan.isDaytona || !env2.mountCreds || !env2.mountedCwd) return; - if (env2.runtimeRemount || !containsTransportEndpointDisconnected(event)) + if (plan.isDaytona || !environment.mountCreds || !environment.mountedCwd) + return; + if ( + environment.runtimeRemount || + !containsTransportEndpointDisconnected(event) + ) return; logger( `local durable cwd ENOTCONN observed in ACP event session=${sessionForMount} cwd=${plan.cwd}; re-signing and remounting`, ); - env2.runtimeRemount = reSignAndRemountLocalCwd().catch((err) => { + environment.runtimeRemount = reSignAndRemountLocalCwd().catch((err) => { logger( `local durable cwd runtime remount failed session=${sessionForMount}: ${conciseError(err, plan.harness)}`, ); @@ -790,7 +801,7 @@ export async function acquireEnvironment( deps.startSandboxAgent ?? ((options: Parameters[0]) => SandboxAgent.start(options)); - env2.sandbox = await startSandboxAgent({ + environment.sandbox = await startSandboxAgent({ sandbox: (deps.buildSandboxProvider ?? buildSandboxProvider)( plan.sandboxId, env, @@ -811,12 +822,12 @@ export async function acquireEnvironment( }); // Track the live handle so a shutdown signal handler can delete it if `destroy` is skipped by // a process KILL; removed in `destroy` on every normal exit so it is never double-deleted. - if (env2.sandbox) inFlightSandboxes.add(env2.sandbox); + if (environment.sandbox) inFlightSandboxes.add(environment.sandbox); // On Daytona, push the harness login, the extension, and AGENTS.md into the remote sandbox. if (plan.isDaytona) { await prepareDaytonaPiAssets({ - sandbox: env2.sandbox, + sandbox: environment.sandbox, plan, log: logger, }); @@ -825,10 +836,10 @@ export async function acquireEnvironment( // Durable cwd: mount BEFORE createSession (so the session opens inside it) and BEFORE // workspace materialization (so AGENTS.md, harness files, and skills land in the durable // prefix instead of being hidden under the FUSE mount). - if (env2.mountCreds && !plan.isDaytona) { + if (environment.mountCreds && !plan.isDaytona) { await mountLocalDurableCwd("initial"); } - if (env2.mountCreds && plan.isDaytona) { + if (environment.mountCreds && plan.isDaytona) { const endpoint = await ( deps.discoverTunnelEndpoint ?? discoverTunnelEndpoint )({ @@ -837,9 +848,9 @@ export async function acquireEnvironment( if ( endpoint && (await (deps.mountStorageRemote ?? mountStorageRemote)( - env2.sandbox, + environment.sandbox, plan.cwd, - env2.mountCreds, + environment.mountCreds, { endpoint, log: logger, @@ -851,23 +862,27 @@ export async function acquireEnvironment( } try { - env2.workspace = await (deps.prepareWorkspace ?? prepareWorkspace)({ - sandbox: env2.sandbox, - plan, - log: logger, - }); + environment.workspace = await (deps.prepareWorkspace ?? prepareWorkspace)( + { + sandbox: environment.sandbox, + plan, + log: logger, + }, + ); } catch (err) { if ( !plan.isDaytona && - env2.mountCreds && + environment.mountCreds && isTransportEndpointDisconnected(err) && (await reSignAndRemountLocalCwd()) ) { logger( `retrying workspace preparation after local durable cwd remount`, ); - env2.workspace = await (deps.prepareWorkspace ?? prepareWorkspace)({ - sandbox: env2.sandbox, + environment.workspace = await ( + deps.prepareWorkspace ?? prepareWorkspace + )({ + sandbox: environment.sandbox, plan, log: logger, }); @@ -878,17 +893,18 @@ export async function acquireEnvironment( // Sandbox-start invariant: `startSandboxAgent` must hand back a usable handle. assert( - env2.sandbox && typeof env2.sandbox.createSession === "function", + environment.sandbox && + typeof environment.sandbox.createSession === "function", `sandbox provider '${plan.sandboxId}' returned no usable sandbox handle`, ); // Probe what this harness supports and branch on capabilities, not on the harness name. const probed = await (deps.probeCapabilities ?? probeCapabilities)( - env2.sandbox, + environment.sandbox, plan.acpAgent, ); const capabilities = probed.capabilities; - env2.capabilities = capabilities; + environment.capabilities = capabilities; // Fail loud (A7): a run that REQUIRES a capability the harness lacks errors specifically // rather than silently dropping the behavior. @@ -913,85 +929,125 @@ export async function acquireEnvironment( log: logger, }); // Close the internal gateway-tool MCP server (if one started) when the session is destroyed. - env2.closeToolMcp = sessionMcp.close; + environment.closeToolMcp = sessionMcp.close; - env2.session = await env2.sandbox.createSession({ + environment.session = await environment.sandbox.createSession({ agent: plan.acpAgent, cwd: plan.cwd, sessionInit: { cwd: plan.cwd, mcpServers: sessionMcp.servers }, }); - env2.sessionId = resolveRunSessionId(request, env2.session.id); + environment.sessionId = resolveRunSessionId( + request, + environment.session.id, + ); // Resolve the model first: when the harness rejects the requested id and keeps its own // default, `model` is undefined and the chat span is labelled "chat". - env2.model = await (deps.applyModel ?? applyModel)( - env2.session, + environment.model = await (deps.applyModel ?? applyModel)( + environment.session, request.model, logger, { strict: strictModel }, ); - // Session-lifetime listeners: attach ONCE, demux into the active turn's sink. Non-throwing - // (the sandbox-agent registries are plain Sets; a thrown handler would corrupt the stream). - // Deliberate divergence from the old inline handler: a handler throw is swallowed + logged - // here instead of propagating, because the listener now outlives any single turn. - env2.session.onEvent((event: any) => { - try { - remountLocalCwdAfterRuntimeEnotconn(event); - const payload = event?.payload; - const update = payload?.params?.update ?? payload?.update; - if (!update) return; - // Record live ACP tool_call ids so a paused client_tool can correlate to Claude's bubble - // (session-scoped; a lookup CONSUMES its matched id). - env2.toolCallIndex.record(update); - const turn = env2.currentTurn; - if (turn) { - turn.handleUpdate(update); - } else { - // Between turns (parked/idle): no turn owns this event. Log and drop by decision. - logger(`[keepalive] between-turns event dropped`); - } - } catch (err) { - logger( - `session onEvent handler error: ${conciseError(err, plan.harness)}`, - ); - } - }); - env2.session.onPermissionRequest((req: any) => { - try { - const turn = env2.currentTurn; - if (turn?.onPermissionRequest) { - turn.onPermissionRequest(req); - return; - } - // Between turns: no turn owns this gate. A slice-2 approval park is recorded DURING the - // active turn (the gate fires while a prompt runs, routing through currentTurn), and a - // parked-on-approval session leaves its harness suspended on that gate — so nothing new - // fires here while parked. A gate that reaches this handler is therefore a genuine stray - // (e.g. a late teardown artifact): cancel by policy so it cannot hang. - logger( - `[keepalive] between-turns permission request, cancelling by policy id=${req?.id}`, - ); - void Promise.resolve( - env2.session?.respondPermission?.(req?.id, "reject"), - ).catch(() => {}); - } catch (err) { - logger( - `session onPermissionRequest handler error: ${conciseError(err, plan.harness)}`, - ); - } - }); + // Session-lifetime listeners: attach ONCE, each demuxing into the active turn's sink. They + // outlive any single turn, so the routing lives in dedicated non-throwing helpers below. + environment.session.onEvent((event: any) => + routeSessionEventToActiveTurn( + environment, + remountLocalCwdAfterRuntimeEnotconn, + event, + ), + ); + environment.session.onPermissionRequest((req: any) => + routePermissionRequestToActiveTurn(environment, req), + ); - return { ok: true, env: env2 }; + return { ok: true, env: environment }; } catch (err) { const error = conciseError(err, plan.harness, request.provider); // Mirror today's shared teardown: no otel exists yet during acquire, so there is no partial // trace to flush — just run the incrementally-registered finalizers and surface the error. - await env2.destroy(); + await environment.destroy(); return { ok: false, error }; } } +/** + * Route one harness event into the active turn's sink. + * + * Data flow: the ACP session emits an event -> we demux it -> the active turn + * (`environment.currentTurn`) consumes the update. The session listener is attached ONCE and + * outlives every turn, so this must never throw: the sandbox-agent registries are plain Sets and a + * thrown handler would corrupt the event stream, so any error is swallowed and logged. + * + * Steps: let the ENOTCONN watcher observe the raw event, extract the update payload (dropping events + * that carry none), record live tool_call ids for client-tool correlation, then hand the update to + * the active turn — or, between turns when no turn owns it, log and drop it. + */ +function routeSessionEventToActiveTurn( + environment: SessionEnvironment, + remountLocalCwdAfterRuntimeEnotconn: (event: unknown) => void, + event: any, +): void { + const { logger, plan } = environment; + try { + remountLocalCwdAfterRuntimeEnotconn(event); + const payload = event?.payload; + const update = payload?.params?.update ?? payload?.update; + if (!update) return; + // Record live ACP tool_call ids so a paused client_tool can correlate to Claude's bubble + // (session-scoped; a lookup CONSUMES its matched id). + environment.toolCallIndex.record(update); + const turn = environment.currentTurn; + if (turn) { + turn.handleUpdate(update); + } else { + // Between turns (parked/idle): no turn owns this event. Log and drop by decision. + logger(`[keepalive] between-turns event dropped`); + } + } catch (err) { + logger(`session onEvent handler error: ${conciseError(err, plan.harness)}`); + } +} + +/** + * Route one permission gate into the active turn's approval handler. + * + * Data flow: the harness raises a permission request -> the active turn (`environment.currentTurn`) + * decides it. Like the event listener this is attached ONCE and must never throw (a thrown handler + * would corrupt the sandbox-agent registries), so errors are swallowed and logged. + * + * Between turns no turn owns the gate. An approval park is always recorded DURING the active turn + * (the gate fires while a prompt runs, routing through currentTurn), and a parked-on-approval + * session leaves its harness suspended on that gate, so nothing new fires while parked. A gate that + * reaches here is therefore a genuine stray (e.g. a late teardown artifact): reject it by policy so + * it cannot hang. + */ +function routePermissionRequestToActiveTurn( + environment: SessionEnvironment, + req: any, +): void { + const { logger, plan } = environment; + try { + const turn = environment.currentTurn; + if (turn?.onPermissionRequest) { + turn.onPermissionRequest(req); + return; + } + logger( + `[keepalive] between-turns permission request, cancelling by policy id=${req?.id}`, + ); + void Promise.resolve( + environment.session?.respondPermission?.(req?.id, "reject"), + ).catch(() => {}); + } catch (err) { + logger( + `session onPermissionRequest handler error: ${conciseError(err, plan.harness)}`, + ); + } +} + /** * Run one turn against an acquired environment: start a fresh otel run, wire this turn's pause * controller / latch / decisions / responder into `env.currentTurn`, restart the tool relay,