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
diff --git a/services/runner/src/engines/sandbox_agent.ts b/services/runner/src/engines/sandbox_agent.ts
index 4b0005a066..1d54ad08ef 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<unknown>;
+}
+
+/** 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<unknown>;
+}
+
+/** 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<void>;
@@ -572,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,
@@ -603,72 +665,81 @@ export async function acquireEnvironment(
     closeToolMcp: undefined,
     currentTurn: undefined,
     lastTurnToolCallIds: [],
+    parkedApproval: undefined,
+    approvalGateCount: 0,
     destroyed: false,
     destroy: async () => {},
     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<boolean> => {
-    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;
@@ -700,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)}`,
       );
@@ -726,7 +801,7 @@ export async function acquireEnvironment(
       deps.startSandboxAgent ??
       ((options: Parameters<typeof SandboxAgent.start>[0]) =>
         SandboxAgent.start(options));
-    env2.sandbox = await startSandboxAgent({
+    environment.sandbox = await startSandboxAgent({
       sandbox: (deps.buildSandboxProvider ?? buildSandboxProvider)(
         plan.sandboxId,
         env,
@@ -747,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,
       });
@@ -761,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
       )({
@@ -773,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,
@@ -787,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,
         });
@@ -814,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.
@@ -849,82 +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: 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.
-        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,
@@ -937,13 +1060,18 @@ export async function runTurn(
   request: AgentRunRequest,
   emit?: EmitEvent,
   signal?: AbortSignal,
-  opts: { continuation?: boolean } = {},
+  opts: RunTurnOptions = {},
 ): Promise<AgentRunResult> {
   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 +1130,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 +1247,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 +1311,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 +1362,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<unknown>;
+    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 +1414,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..a4556622fc 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
@@ -268,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;
 }
 
 /**
@@ -380,10 +439,35 @@ export class SessionPool<E = unknown> {
   }
 
   /**
-   * 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<E> | 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<E>,
@@ -393,21 +477,30 @@ export class SessionPool<E = unknown> {
       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,13 +510,20 @@ export class SessionPool<E = unknown> {
    * 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<E>, ttlMs: number): boolean {
+  async park(
+    input: ParkInput<E>,
+    ttlMs: number,
+    state: "idle" | "awaiting_approval" = "idle",
+  ): Promise<boolean> {
     // 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()) {
@@ -439,23 +539,38 @@ export class SessionPool<E = unknown> {
       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<E>,
+    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..be5216311f 100644
--- a/services/runner/src/server.ts
+++ b/services/runner/src/server.ts
@@ -35,13 +35,16 @@ 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,
+  credentialEpochMismatch,
+  mountCredentialsExpired,
   expectedNextHistoryFingerprint,
   historyFingerprint,
   poolKeyFor,
@@ -50,6 +53,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 +240,7 @@ export interface KeepaliveEngine {
     request: AgentRunRequest,
     emit: EmitEvent | undefined,
     signal: AbortSignal | undefined,
-    opts: { continuation?: boolean },
+    opts: RunTurnOptions,
   ): Promise<AgentRunResult>;
   /**
    * Today's cold path (acquire -> runTurn -> teardown). Used when a request must not park.
@@ -326,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)) {
@@ -351,13 +369,135 @@ 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<void> => {
+    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 (
+        !(await pool.park(input, config.approvalTtlMs, "awaiting_approval"))
+      ) {
+        await env.destroy();
+      } else {
+        watchParkedPrompt(env);
+      }
+    } else if (shouldPark(result, signal, clientGone)) {
+      if (!(await 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<SessionEnvironment>,
+    result: AgentRunResult,
+  ): Promise<void> => {
+    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<AgentRunResult> => {
     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, trackedEmit, signal, {
+        approvalParkMode: true,
+      });
     } catch (err) {
       await env.destroy();
       return {
@@ -365,27 +505,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;
   };
 
@@ -393,11 +513,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) {
@@ -413,54 +538,163 @@ export async function runWithKeepalive(
       klog(`hit-continue key=${key}`);
       let result: AgentRunResult;
       try {
-        result = await engine.runTurn(live.environment, request, emit, signal, {
-          continuation: true,
-        });
+        // A continuation can itself raise an approval gate, so it runs in park mode too.
+        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();
       }
-      live.environment.clearTurn();
-      if (shouldPark(result, signal, clientGone)) {
-        const reparked = pool.repark(
-          live,
+      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. 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)
+      : 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 (priorFp !== existing.historyFingerprint) {
+      mismatch = "history";
+    } else if (mountCredentialsExpired(existing.credentialEpoch)) {
+      mismatch = "credentials-expired";
+    }
+
+    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,
+          trackedEmit,
+          signal,
           {
-            configFingerprint: cfgFp,
-            historyFingerprint: expectedNextHistoryFingerprint(
-              request.messages ?? [],
-              live.environment.lastTurnToolCallIds ?? [],
-            ),
-            credentialEpoch: incomingEpoch,
+            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) {
+        // 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) {
+        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);
       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/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-approval.test.ts b/services/runner/tests/unit/session-keepalive-approval.test.ts
new file mode 100644
index 0000000000..fc5b1ded24
--- /dev/null
+++ b/services/runner/tests/unit/session-keepalive-approval.test.ts
@@ -0,0 +1,1330 @@
+/**
+ * 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<void> {
+  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<void>;
+  destroy: () => Promise<void>;
+}
+
+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<number, () => 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<AgentRunResult> => {
+    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<void>((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<KeepaliveConfig> = {},
+  clientGone?: () => boolean,
+) {
+  const config: KeepaliveConfig = {
+    enabled: true,
+    ttlMs: 60_000,
+    approvalTtlMs: 600_000,
+    poolMax: 8,
+    ...overrides,
+  };
+  const pool = new SessionPool<SessionEnvironment>(
+    { 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> = {},
+): 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,
+    mountOpts: { expiresAt?: string } = {},
+  ) {
+    const { engine, calls } = makeApprovalEngine(
+      [
+        {
+          approvalPause: {
+            permissionId: "perm-1",
+            toolCallId: "tc-gate",
+            toolName: "commit",
+          },
+          toolCallIds: ["tc-gate"],
+        },
+        { result: { ok: true, output: "cold", stopReason: "complete" } },
+      ],
+      mountOpts,
+    );
+    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("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",
+    );
+  });
+
+  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 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();
+    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<string>;
+  /** 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<string, any> => {
+    const run: FakeRun & Record<string, any> = {
+      id: calls.runs.length + 1,
+      handled: [],
+      emitted: [],
+      open: new Set<string>(),
+      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<string, unknown>) {
+  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-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<AgentRunResult> => {
     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 d4cd182ce2..8f8514bdd9 100644
--- a/services/runner/tests/unit/session-pool.test.ts
+++ b/services/runner/tests/unit/session-pool.test.ts
@@ -8,9 +8,12 @@ import assert from "node:assert/strict";
 
 import type { AgentRunRequest } from "../../src/protocol.ts";
 import {
+  approvalDecisionForToolCall,
   computeCredentialEpoch,
   configFingerprint,
+  credentialEpochMismatch,
   credentialEpochValid,
+  mountCredentialsExpired,
   expectedNextHistoryFingerprint,
   historyFingerprint,
   poolKeyFor,
@@ -52,6 +55,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",
@@ -326,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", () => {
@@ -341,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);
@@ -378,13 +471,146 @@ 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");
     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(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 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");
@@ -392,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",
     );
@@ -491,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<void>((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) => {