diff --git a/.agents/skills/agent-release-gate/resources/qa_product.py b/.agents/skills/agent-release-gate/resources/qa_product.py
index 637dc63f4f..39916e8c68 100644
--- a/.agents/skills/agent-release-gate/resources/qa_product.py
+++ b/.agents/skills/agent-release-gate/resources/qa_product.py
@@ -3167,12 +3167,19 @@ def _load_session_control_result(path: str) -> dict:
     skipped = sorted(name for name, status in statuses.items() if status == "SKIP")
     return {
         "path": str(result_path),
-        "status": "FAIL" if failed else "PASS",
+        "status": "FAIL" if failed else ("INCOMPLETE" if skipped else "PASS"),
         "failed": failed,
         "skipped": skipped,
     }
 
 
+def _session_control_result_label(result: dict) -> str:
+    label = f"recorded {result['status']}"
+    if result["skipped"]:
+        label += "; SKIPPED, UNTESTED: " + ", ".join(result["skipped"])
+    return label
+
+
 def main() -> int:
     # Declared here, not beside the assignments below, because the flag help strings read these
     # module defaults and a `global` statement must precede every use of the name in a function.
@@ -3453,7 +3460,7 @@ def main() -> int:
                     "MISSING — no such cell exists"
                     if cell in missing_cells
                     else (
-                        f"recorded {session_control_result['status']}"
+                        _session_control_result_label(session_control_result)
                         if cell == "session_control.py" and session_control_result
                         else "run it separately"
                     )
@@ -3558,7 +3565,7 @@ def main() -> int:
             if cell in CELLS:
                 here = "yes"
             elif cell == "session_control.py" and session_control_result:
-                here = f"recorded {session_control_result['status']}"
+                here = _session_control_result_label(session_control_result)
             else:
                 here = "no — run it separately"
             table += f"| {cell} | {here} | {', '.join(why)} |\n"
@@ -3593,7 +3600,7 @@ def main() -> int:
         for journey in cell["journeys"].values()
     )
     standalone_failed = bool(
-        session_control_result and session_control_result["status"] == "FAIL"
+        session_control_result and session_control_result["status"] != "PASS"
     )
     return 1 if failed or standalone_failed else 0
 
diff --git a/.agents/skills/agent-release-gate/resources/session_control.py b/.agents/skills/agent-release-gate/resources/session_control.py
index 2f370c2ba9..1042e930c7 100644
--- a/.agents/skills/agent-release-gate/resources/session_control.py
+++ b/.agents/skills/agent-release-gate/resources/session_control.py
@@ -2098,8 +2098,10 @@ def _judge_stop_approval(evidence: dict, *, pending_found: bool) -> dict:
             "pass --durable-stop on or off"
         )
     late = evidence["late_answer"]
-    if durable_stop == "on" and late.get("status") == 200:
-        return _fail("the late approval answer was accepted after the Stop settled it")
+    if durable_stop == "on" and late.get("status") != 409:
+        return _fail(
+            f"the late approval answer returned HTTP {late.get('status')}, expected 409"
+        )
     if durable_stop == "off":
         if late.get("status") != 200:
             return _fail(
diff --git a/.agents/skills/agent-release-gate/resources/test_qa_product_concurrency.py b/.agents/skills/agent-release-gate/resources/test_qa_product_concurrency.py
index 26f1628aaa..4638008b60 100644
--- a/.agents/skills/agent-release-gate/resources/test_qa_product_concurrency.py
+++ b/.agents/skills/agent-release-gate/resources/test_qa_product_concurrency.py
@@ -1,12 +1,12 @@
 # /// script
 # requires-python = ">=3.10"
-# dependencies = ["httpx>=0.27"]
+# dependencies = ["httpx>=0.27", "pytest>=8"]
 # ///
 """Offline tests for the `burst` and `crosstalk` journeys. No deployment, no network.
 
 Run either way:
 
-    uv run test_qa_product_concurrency.py     # standalone, prints a line per case
+    uv run test_qa_product_concurrency.py     # standalone, runs through pytest
     uv run --no-sync pytest test_qa_product_concurrency.py
 
 Every case fakes the wire. `invoke` is replaced with a function that builds a `Turn` by hand, so
@@ -436,6 +436,19 @@ def test_session_control_result_consumer_carries_a_failure(tmp_path):
     assert result["failed"] == ["stop-warm"]
 
 
+def test_session_control_result_consumer_marks_skips_incomplete(tmp_path):
+    path = tmp_path / "results.json"
+    path.write_text(json.dumps(_session_control_result("SKIP")))
+
+    result = qa._load_session_control_result(str(path))
+
+    assert result["status"] == "INCOMPLETE"
+    assert result["skipped"]
+    label = qa._session_control_result_label(result)
+    assert "SKIPPED, UNTESTED" in label
+    assert result["skipped"][0] in label
+
+
 def test_session_control_result_consumer_rejects_an_incomplete_run(tmp_path):
     payload = _session_control_result()
     del payload["cells"]["stop-warm"]
@@ -495,6 +508,30 @@ def test_driver_fails_for_a_failed_session_control_result(monkeypatch, tmp_path)
     assert qa.main() == 1
 
 
+def test_driver_fails_for_a_skipped_session_control_result(monkeypatch, tmp_path):
+    result_path = tmp_path / "session-control-results.json"
+    result_path.write_text(json.dumps(_session_control_result("SKIP")))
+    monkeypatch.setattr(qa, "RUNS", tmp_path / "runs")
+    monkeypatch.setitem(qa.JOURNEYS, "chat", lambda _cell: {"pass": True, "why": "ok"})
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "qa_product.py",
+            "--cell",
+            "C3",
+            "--only",
+            "chat",
+            "--changed-path",
+            "api/oss/src/core/sessions/service.py",
+            "--session-control-results",
+            str(result_path),
+        ],
+    )
+
+    assert qa.main() == 1
+
+
 def test_the_driver_forces_a_mandatory_journey_past_only(tmp_path=None):
     """End to end through main(), with every journey stubbed out."""
     import tempfile
@@ -958,12 +995,7 @@ def never_ends(session, messages, params, timeout=300.0, deadline=None):
 
 
 def main() -> int:
-    cases = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
-    for case in cases:
-        case()
-        print(f"PASS {case.__name__}")
-    print(f"\n{len(cases)} offline cases passed")
-    return 0
+    return pytest.main([__file__, "-q"])
 
 
 if __name__ == "__main__":
diff --git a/.agents/skills/agent-release-gate/resources/test_session_control.py b/.agents/skills/agent-release-gate/resources/test_session_control.py
index 2ad2ead351..148a15af09 100644
--- a/.agents/skills/agent-release-gate/resources/test_session_control.py
+++ b/.agents/skills/agent-release-gate/resources/test_session_control.py
@@ -85,10 +85,13 @@ def _stop_approval_evidence(*, durable_stop: str, late_status: int) -> dict:
 
 
 def test_stop_approval_durable_path_requires_late_answer_refusal():
-    accepted = _stop_approval_evidence(durable_stop="on", late_status=200)
     refused = _stop_approval_evidence(durable_stop="on", late_status=409)
 
-    assert sc._judge_stop_approval(accepted, pending_found=True)["pass"] is False
+    for status in (200, 202, 500):
+        unexpected = _stop_approval_evidence(durable_stop="on", late_status=status)
+        verdict = sc._judge_stop_approval(unexpected, pending_found=True)
+        assert verdict["pass"] is False
+        assert f"HTTP {status}, expected 409" in verdict["why"]
     verdict = sc._judge_stop_approval(refused, pending_found=True)
     assert verdict["pass"] is True
     assert "late answer was refused" in verdict["why"]
diff --git a/api/oss/src/apis/fastapi/sessions/router.py b/api/oss/src/apis/fastapi/sessions/router.py
index aa20b32872..b587e34fcb 100644
--- a/api/oss/src/apis/fastapi/sessions/router.py
+++ b/api/oss/src/apis/fastapi/sessions/router.py
@@ -551,6 +551,9 @@ async def heartbeat_session_stream(
         if not has_permission:
             raise FORBIDDEN_EXCEPTION
 
+        if payload.release_owner:
+            _assert_runner_token(request)
+
         heartbeat = await self._service.heartbeat(
             project_id=project_id,
             request=payload,
diff --git a/api/oss/src/utils/env.py b/api/oss/src/utils/env.py
index d4bcfa5135..b27dc8af64 100644
--- a/api/oss/src/utils/env.py
+++ b/api/oss/src/utils/env.py
@@ -537,11 +537,21 @@ class SessionsRecordsConfig(BaseModel):
 
     # How long a record message the worker failed to write sits unacknowledged before the
     # worker claims it back and tries again.
-    reclaim_idle_ms: int = int(os.getenv("AGENTA_RECORDS_RECLAIM_IDLE_MS") or 30_000)
+    reclaim_idle_ms: int = Field(
+        default_factory=lambda: int(
+            os.getenv("AGENTA_RECORDS_RECLAIM_IDLE_MS") or 30_000
+        ),
+        ge=0,
+        validate_default=True,
+    )
 
     # Deliveries after which a record message is dropped instead of retried forever. A message
     # Postgres never accepts would otherwise hold every later message in the group.
-    max_deliveries: int = int(os.getenv("AGENTA_RECORDS_MAX_DELIVERIES") or 5)
+    max_deliveries: int = Field(
+        default_factory=lambda: int(os.getenv("AGENTA_RECORDS_MAX_DELIVERIES") or 5),
+        ge=1,
+        validate_default=True,
+    )
 
     model_config = ConfigDict(extra="ignore")
 
diff --git a/api/oss/tests/pytest/unit/sessions/test_heartbeat_release_auth.py b/api/oss/tests/pytest/unit/sessions/test_heartbeat_release_auth.py
new file mode 100644
index 0000000000..96c0e42133
--- /dev/null
+++ b/api/oss/tests/pytest/unit/sessions/test_heartbeat_release_auth.py
@@ -0,0 +1,97 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import UUID
+
+import pytest
+from fastapi import HTTPException
+
+from oss.src.apis.fastapi.sessions import router as router_module
+from oss.src.apis.fastapi.sessions.router import SessionStreamsRouter
+from oss.src.core.sessions.streams.dtos import (
+    SessionHeartbeatRequest,
+    SessionHeartbeatResult,
+)
+from oss.src.utils.env import env
+
+
+_PROJECT = UUID("00000000-0000-0000-0000-0000000000aa")
+_USER = UUID("00000000-0000-0000-0000-0000000000bb")
+
+
+def _request(headers=None):
+    return SimpleNamespace(
+        state=SimpleNamespace(project_id=_PROJECT, user_id=_USER),
+        headers=headers or {},
+    )
+
+
+def _router(service):
+    return SessionStreamsRouter(
+        service=service,
+        interactions_service=SimpleNamespace(),
+    )
+
+
+@pytest.mark.asyncio
+async def test_release_owner_heartbeat_requires_the_runner_token(monkeypatch):
+    monkeypatch.setattr(env.runner, "token", "runner-secret")
+    monkeypatch.setattr(
+        router_module, "check_action_access", AsyncMock(return_value=True)
+    )
+    service = SimpleNamespace(heartbeat=AsyncMock())
+
+    with pytest.raises(HTTPException) as exc_info:
+        await _router(service).heartbeat_session_stream(
+            _request(),
+            SessionHeartbeatRequest(
+                session_id="session-1",
+                replica_id="replica-1",
+                release_owner=True,
+            ),
+        )
+
+    assert exc_info.value.status_code == 401
+    service.heartbeat.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_regular_heartbeat_keeps_user_authentication_only(monkeypatch):
+    monkeypatch.setattr(env.runner, "token", "runner-secret")
+    monkeypatch.setattr(
+        router_module, "check_action_access", AsyncMock(return_value=True)
+    )
+    service = SimpleNamespace(
+        heartbeat=AsyncMock(return_value=SessionHeartbeatResult(replica_id="replica-1"))
+    )
+    payload = SessionHeartbeatRequest(
+        session_id="session-1",
+        replica_id="replica-1",
+        turn_id="turn-1",
+    )
+
+    result = await _router(service).heartbeat_session_stream(_request(), payload)
+
+    assert result.replica_id == "replica-1"
+    service.heartbeat.assert_awaited_once_with(project_id=_PROJECT, request=payload)
+
+
+@pytest.mark.asyncio
+async def test_release_owner_accepts_the_shared_runner_token(monkeypatch):
+    monkeypatch.setattr(env.runner, "token", "runner-secret")
+    monkeypatch.setattr(
+        router_module, "check_action_access", AsyncMock(return_value=True)
+    )
+    service = SimpleNamespace(
+        heartbeat=AsyncMock(return_value=SessionHeartbeatResult(replica_id="replica-1"))
+    )
+    payload = SessionHeartbeatRequest(
+        session_id="session-1",
+        replica_id="replica-1",
+        release_owner=True,
+    )
+
+    await _router(service).heartbeat_session_stream(
+        _request({"X-Agenta-Runner-Token": "runner-secret"}), payload
+    )
+
+    service.heartbeat.assert_awaited_once_with(project_id=_PROJECT, request=payload)
diff --git a/api/oss/tests/pytest/unit/sessions/test_records_config.py b/api/oss/tests/pytest/unit/sessions/test_records_config.py
new file mode 100644
index 0000000000..a67cf8a3d6
--- /dev/null
+++ b/api/oss/tests/pytest/unit/sessions/test_records_config.py
@@ -0,0 +1,36 @@
+import pytest
+from pydantic import ValidationError
+
+from oss.src.utils.env import SessionsRecordsConfig
+
+
+def test_session_record_retry_bounds_accept_the_minimum_values():
+    config = SessionsRecordsConfig(reclaim_idle_ms=0, max_deliveries=1)
+
+    assert config.reclaim_idle_ms == 0
+    assert config.max_deliveries == 1
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [("reclaim_idle_ms", -1), ("max_deliveries", 0), ("max_deliveries", -1)],
+)
+def test_session_record_retry_bounds_reject_invalid_values(field, value):
+    with pytest.raises(ValidationError):
+        SessionsRecordsConfig(**{field: value})
+
+
+@pytest.mark.parametrize(
+    ("name", "value"),
+    [
+        ("AGENTA_RECORDS_RECLAIM_IDLE_MS", "-1"),
+        ("AGENTA_RECORDS_MAX_DELIVERIES", "0"),
+    ],
+)
+def test_session_record_retry_bounds_validate_environment_defaults(
+    monkeypatch, name, value
+):
+    monkeypatch.setenv(name, value)
+
+    with pytest.raises(ValidationError):
+        SessionsRecordsConfig()
diff --git a/docs/design/session-control-and-live-events/research.md b/docs/design/session-control-and-live-events/research.md
index fd2330744f..9abe25ff86 100644
--- a/docs/design/session-control-and-live-events/research.md
+++ b/docs/design/session-control-and-live-events/research.md
@@ -21,10 +21,15 @@ when a heartbeat returns `is_current_turn=false`, then aborts locally.
 `DELETE /sessions/streams/?session_id=...` is separate from normal Cancel. It contacts the runner
 and tears down the sandbox. The session remains resumable after Cancel but not after Kill.
 
-The direct kill client uses one configured `runner.internal_url`. Redis separately stores the
-logical owner `replica_id`. The current kill client does not resolve that identifier to a
-replica-specific address. Immediate Cancel cannot assume that logical owner identity already
-provides direct network routing.
+The v1 direct client uses one configured `runner.internal_url`. It is correct for a single runner,
+or when the URL fronts an owner-aware router. Redis separately stores the logical owner
+`replica_id`, but the direct client does not resolve that identity to a replica-specific address.
+A request that reaches the wrong replica returns not found and must not be treated as success.
+
+Immediate Cancel remains durable, so an unavailable owner can recover and apply it later or be
+settled as lost. Kill is best effort through the same configured URL. Until owner-aware forwarding
+exists, a multi-runner Kill cannot guarantee immediate teardown; authoritative session state is
+cleared and sandbox lease or orphan cleanup provides the fallback.
 
 ### Heartbeat
 
diff --git a/docs/design/session-control-and-live-events/rfc.md b/docs/design/session-control-and-live-events/rfc.md
index 6716361dd8..223fd5ad76 100644
--- a/docs/design/session-control-and-live-events/rfc.md
+++ b/docs/design/session-control-and-live-events/rfc.md
@@ -44,7 +44,7 @@ Idempotency-Key: <client-generated-key>
 {
   "type": "send",
   "message": "Explain this failure",
-  "delivery": "reject"
+  "on_busy": "reject"
 }
 ```
 
@@ -126,7 +126,7 @@ execution is already running:
 - `queue`: save the new message. Start it after current work stops normally.
 - `steer`: save the new message. Interrupt current work, then start the new message.
 
-When the session is idle, all accepted messages start normally. The contract may call this field
+When the session is idle, all accepted messages start normally. The contract calls this field
 `on_busy` so its purpose is clear.
 
 ### Visible pending messages
@@ -243,6 +243,10 @@ acknowledges it, and immediately opens the next request. A disconnected runner r
 claims commands that remain durable. Redis or Postgres notifications may wake API replicas
 internally, but the runner never connects to either system.
 
+Credential-bearing long polls require HTTPS with normal certificate validation. The client must
+disable redirects or reject any redirect whose origin differs from the configured API origin, and
+it must never forward runner credentials across origins.
+
 A persistent WebSocket or bidirectional stream can later reduce repeated requests and carry richer
 runner status. It is not required for the first contract. Direct API calls into runner pods and
 per-runner Redis subscriptions are poor fits for user-operated runners because they require inbound
@@ -303,8 +307,10 @@ execution: running -> stopping -> stopped
 
 The API accepts Stop by durably creating the command and moving the matching execution to
 `stopping` in one transaction. `expected_execution_id` remains optional. A command claim has a
-lease and can be delivered again after disconnection. The runner deduplicates by `command_id` and
-validates the execution ID and ownership generation before applying it.
+lease and can be delivered again after disconnection. In a fenced design, the runner deduplicates
+by `command_id` and validates both the execution ID and ownership generation before applying it.
+The v1 direct-delivery adapter has no generation token; it validates the target execution ID and
+requires the addressed runner replica to own that execution.
 
 Claiming or acknowledging a command does not prove that execution stopped. Public clients follow
 execution state. The runner normally reports the terminal outcome and the API settles the command
diff --git a/docs/design/session-control-and-live-events/slice-admission.md b/docs/design/session-control-and-live-events/slice-admission.md
index b8aa8da4b7..1aa2633389 100644
--- a/docs/design/session-control-and-live-events/slice-admission.md
+++ b/docs/design/session-control-and-live-events/slice-admission.md
@@ -71,10 +71,10 @@ point:
 The refusal streams as an `error` event carrying the code, then a failed terminal result. That is
 the path every runner failure already takes to the browser, so no new transport is involved.
 
-The coordinator change is a backstop, not the fix. The heartbeat fails open on a network or HTTP
-error, which is deliberate and unchanged: a transient API blip refusing every message would be a
-worse outage than the bug. In that window two turns can be admitted, and a `busy` pool entry is
-the more specific truth on this box, so the coordinator refuses rather than destroying.
+The first heartbeat is the admission decision and fails closed unless the coordination plane
+confirms ownership. Later heartbeat failures remain best effort for a turn that was already
+admitted. The coordinator stays as a same-runner backstop and refuses a competing `busy` pool entry
+without destroying the live environment.
 
 ### 2. The browser keeps the user's text (`bdd7116520`)
 
@@ -197,9 +197,8 @@ The four runner failures are **pre-existing**, all in
 `tests/unit/gateway-run-turn-composition.test.ts`. Confirmed by stashing this slice's changes and
 re-running: the same four fail on the branch tip.
 
-The 11 collection errors in the API run are an artifact of borrowing the live tree's virtual
-environment, which resolves `agenta` from `/home/mahmoud/code/agenta-2/sdks/python` rather than
-from this worktree. They are import errors in unrelated files.
+The 11 collection errors in the API run came from a virtual environment that resolved `agenta`
+from a different checkout. They are import errors in unrelated files.
 
 New tests:
 
@@ -207,9 +206,9 @@ New tests:
   driven over a socket against a fake platform API. Covers: a refused turn never calls `run()`,
   the error event carries the code, no interaction sweep or attachment claim happens, the end
   beat names the refused turn, an admitted turn proceeds, a resume-shaped request is admitted,
-  and an unreachable platform fails open.
+  and an unreachable platform fails closed before `run()`.
 - `services/runner/tests/unit/session-alive-interrupt.test.ts` (+4). `admitted` semantics: first
-  beat only, fail-open, and a later interruption does not un-admit.
+  beat only, fail-closed without confirmation, and a later interruption does not un-admit.
 - `services/runner/tests/unit/session-keepalive-dispatch.test.ts` (+1, 1 rewritten). A busy entry
   refuses with no eviction and no cold acquire; a destroyed entry still evicts.
 - `services/runner/tests/unit/session-steer-mount-loss.test.ts` (3 rewritten). These pinned the
@@ -236,33 +235,21 @@ that destroys a session also removes it.
 
 ### The stack
 
-A standalone EE dev stack built from this worktree, at **http://144.76.237.122:8680**.
+A standalone EE development stack built from this worktree at `<dev-host>:<port>`. The deployment
+used a current EE development environment file with isolated ports and project name. Dev-mode bind
+mounts confirmed that the containers ran this worktree's source.
 
-The brief named `hosting/docker-compose/ee/.env.ee.dev.local` as the base env file. That file is
-from 30 July and is missing `AGENTA_SERVICES_INTERNAL_KEY`, so compose refuses to start. The env
-file was rebased on `.env.ee.dev.toolkit.local` (29 August), which is the one Mahmoud's own stack
-runs, with every port, the project name and the env-file pointer changed. The four
-`agenta-ee-dev-*:latest` images were 15 minutes old, so `--build` was skipped as the brief
-directed; dev mode bind-mounts the source, so the containers run this worktree's code.
+When host and container users differ, dependency ownership can prevent the web entrypoint from
+updating generated binaries. Repair only the affected dependency or generated paths with targeted
+ownership or ACL changes, then restart the container. Never make the whole web tree world-writable.
 
-Two deployment notes worth keeping. First, the stale env file: compose fails immediately with
-`required variable AGENTA_SERVICES_INTERNAL_KEY is missing a value`, which names the problem
-clearly. Second, the web container 502s indefinitely if you have also run `pnpm install` in this
-worktree's `web/` from the host, as this slice did for lint and tests. The host install runs as
-uid 1000 and the container as uid 10001, so the container's own install and the api-client
-`prepare` build cannot overwrite those paths and the entrypoint retries forever. The log looks
-like a slow install; the real line is `[EACCES] ... .bin/tsc` thousands of lines up. Fix with
-`chmod -R a+rwX web/` in the worktree and restart the container, then poll `/w` rather than `/`,
-because `/` 308-redirects there and the first compile takes a few minutes.
-
-Sandbox provider: `local`. Harness: `pi_core`. Model: `gpt-5.6-luna` on the QA OpenAI key, added
-to the stack's own vault.
+Sandbox provider: `local`. Harness: `pi_core`. The model credential came from the stack's test
+vault; no key or secret is part of this record.
 
 ### The scenario
 
-Driver: `verify_admission.py`, wire level, asserting on SSE frame types and never on model prose.
-It is kept at
-`/tmp/claude-1000/-home-mahmoud-code-agenta-2/7c724667-82cd-41a6-ba0b-e47bc96b4f67/scratchpad/verify_admission.py`.
+The verification driver worked at wire level, asserted on SSE frame types, and never used model
+prose as evidence. Its environment-specific path and credentials are intentionally not recorded.
 
 1. Turn A starts on a fresh session and runs `sleep 40 && echo DONE_A` as a shell tool.
 2. Fifteen seconds in, turn B sends "What is 2 + 2?" to the same session.
@@ -287,7 +274,7 @@ Turn B's error frames, verbatim:
 {"type": "error", "errorText": "This session is already running a turn. Your message was not sent. Wait for the reply, or stop the turn, then send again."}
 ```
 
-Runner log for session `081a1fe7-9961-4a0e-bdb1-177a59a8bfd6`, in order:
+Runner log for the test session, in order:
 
 ```
 [sessions] stream sessionOwned=true sessionId=081a1fe7-… turnId=444d272b-… cred=present
@@ -314,15 +301,13 @@ Three things to read from that log:
   sandbox and the native harness session survived the second send. That is the constraint this
   slice was bound by, checked rather than assumed.
 
-The stack is left running. Teardown:
+Use the matching edition, image mode, and environment file to tear down the isolated stack:
 
 ```bash
-cd /home/mahmoud/code/agenta-2-worktrees/slice-admission
-bash ./hosting/docker-compose/run.sh --license ee --dev --env-file .env.ee.dev.admission --down
+bash ./hosting/docker-compose/run.sh --ee --dev --down
 ```
 
-Add `--nuke` to drop the volumes as well. That stack has its own Postgres on port 5441 and shares
-nothing with the other stacks on the box.
+Add `--nuke` only when the isolated volumes should also be removed.
 
 ### Not verified
 
@@ -377,10 +362,9 @@ this slice.
    the conversation. *Recommendation: move it there once someone looks at it in a browser.* The
    current bubble is honest but it sits in the transcript, which is where run failures live.
 
-4. **Is the fail-open on an unreachable API still the right default?** It is unchanged from
-   today, and the coordinator's busy check backs it up on a single runner.
-   *Recommendation: keep it.* Refusing every message during an API blip would be a worse outage
-   than the bug this closes, and with one runner the local check catches the real overlap.
+4. **Should initial admission fail closed when the API is unreachable?** *Decision: yes.* At-most-one
+   execution has to hold across replicas. Later watchdog failures remain best effort so an already
+   admitted healthy turn is not aborted by a transient API failure.
 
 5. **Should `--build` have been skipped?** The brief said to skip it if the images were under
    three hours old, and they were fifteen minutes old. The live results therefore depend on dev
diff --git a/docs/design/session-control-and-live-events/spike-a-sandbox-cancel.md b/docs/design/session-control-and-live-events/spike-a-sandbox-cancel.md
index 5344e3361f..c125b9533d 100644
--- a/docs/design/session-control-and-live-events/spike-a-sandbox-cancel.md
+++ b/docs/design/session-control-and-live-events/spike-a-sandbox-cancel.md
@@ -6,10 +6,10 @@ Status: the six questions are answered, the runner change is written and unit te
 scenario passed on the local sandbox for two harnesses. The Claude harness is not tested, because
 this stack has no Anthropic key.
 
-**One finding needs a decision before this ships: a stopped Codex turn leaves its shell command
-running inside the parked sandbox.** Pi kills its child; Codex does not. Before this change the
-sandbox was deleted, which killed the orphan, so parking is what makes it survive. Measured, both
-directions, in "What happens to the in-flight tool" below.
+**Codex process reaping is best effort after a settled Stop.** Pi kills its child; Codex does not.
+The runner attempts to reap the Codex child and records cleanup misses for QA. A cleanup miss does
+not revoke warm reuse or continuity; the 600-second stopped-session window bounds any leftover
+process.
 
 ## The answer in one paragraph
 
@@ -116,20 +116,18 @@ The Codex reading is unambiguous. One probe returned two leftovers at once, `sle
 seconds elapsed and `sleep 300` at 31 seconds elapsed, which are the cancelled turns of two
 different sessions, so the child survives its own turn AND the session that spawned it.
 
-**Parking made the original leak survive.** Running the same Codex scenario with the settle budget
-forced to 1 ms destroyed the environment and left no leftover. The runner now closes that gap in
+**Parking can expose the original leak.** The runner performs a best-effort cleanup in
 `reap-exec.ts`: after the cancelled prompt settles, it finds the `codex app-server` below this
 sandbox's daemon, selects only descendants started during the stopped turn, and checks that
 `kill -9` exits successfully before reporting them reaped. The turn-boundary test pins the order as
-cancel, process scan, reap, then park. The app server and older session processes remain alive, so
-the native session survives without a Daytona snapshot rebuild.
+cancel, process scan, reap, then park. Failed or unknown cleanup is recorded for QA, while the
+settled Stop still preserves the sandbox and native session for the 600-second stopped window.
 
 **What reaches the API.** The turn's `message`, `tool_call` and `tool_result` rows, a `usage` row,
-and the terminal `done` row, all present in the live runs. The terminal record now carries
-`stopReason: "cancelled"` (see below). The turn is still NOT marked complete in the turn ledger, and
-the runner drops the harness's continuity record, because a cancelled turn is not a faithful resume
-point for a COLD rebuild (`services/runner/src/engines/sandbox_agent/run-turn.ts:1429`). See the
-open issues.
+and the terminal `done` row were present in the live runs. The terminal record carries
+`stopReason: "cancelled"` (see below). When the harness confirms cancellation, the runner completes
+the turn ledger row and preserves the native-session continuity record. Reap outcomes do not alter
+that confirmation.
 
 ### 3b. A stopped turn is now distinguishable from a completed one
 
@@ -297,8 +295,8 @@ interval, which work package B replaces with long polling.
 
 ## The live test
 
-Stack `agenta-ee-dev-session-spike` on `http://144.76.237.122:8580`, built from the worktree
-`/home/mahmoud/code/agenta-2-worktrees/spike-a-cancel`, local sandbox provider, EE, dev image.
+An isolated EE development stack built from the spike branch used the local sandbox provider and
+development images.
 
 Protocol, driven by `spike_cancel_live.py` in the evidence folder:
 
@@ -387,28 +385,26 @@ Add one cell, run per harness and on both sandbox providers.
 3. Assert on the stream: the turn ends with `finish`, its open tool call settles as
    `tool-output-error`, and no `error` frame claims the run failed.
 4. Assert on the runner log: `stage=harness_cancel sent=true settled=true`, then, for Codex,
-   `stage=harness_reap killed=...`, then `prompt stopReason=cancelled`, then `park-cancelled`. Fail
-   the cell on `no-park:cancelled` or `stage=harness_reap ... skipped=kill-failed`.
+   `stage=harness_reap killed=...`, then `prompt stopReason=cancelled`, then `park-cancelled`. Record
+   `cleanup_miss=true` as QA evidence, but fail the warm-reuse cell only on `no-park:cancelled`.
 5. Send a second message on the same session, replaying the cancelled turn's assistant message.
 6. Assert on the runner log: `hit-continue` for the same pool key, and NO `stage=sandbox_start`
    between the two turns. On Daytona, additionally assert the sandbox id is unchanged.
 7. Assert the second turn's answer references something only turn 1 said.
 8. Assert the stopped turn's terminal `done` record carries `stopReason: "cancelled"` and the
    completed turn's does not.
-9. Assert no leftover process from the cancelled command survives into the second turn. This one
-   FAILS on Codex today, on purpose: it is the check that tells us when the bridge is fixed.
+9. When reaping succeeds, assert that no leftover process from the cancelled command survives into
+   the second turn. When reaping fails or is unknown, record the cleanup miss and still assert warm
+   parking and native-session continuity; the stopped TTL bounds the leftover process to 600 seconds.
 
 The negative leg is worth keeping too: with `AGENTA_RUNNER_HARNESS_CANCEL_SETTLE_MS=1` the same
 scenario must log `settled=false` and `no-park:cancelled`. That proves the guard still guards.
 
 ## Open questions for Mahmoud
 
-1. **A stopped Codex turn leaves its shell command running in the parked sandbox. Ship anyway, or
-   hold Codex back?** Recommendation: ship, and fix the bridge next. The orphan dies when the stopped
-   window closes. The stopped window is 600 s on Daytona, where the compute is billed, and holding
-   Codex back means Codex users keep paying a cold start on every Stop. The alternative, an env flag
-   that excludes one harness from parking, is machinery for a decision we would reverse within the
-   week.
+1. **How should a failed Codex reap affect parking?** Decision: keep the settled Stop parked. Reaping
+   is best effort, cleanup misses are QA evidence, and the 600-second stopped TTL bounds leftovers
+   without sacrificing warm reuse or native-session continuity.
 2. **Ten seconds for the settle budget?** Recommendation: yes, ship it. The measured cost is
    14 to 31 ms, so the budget is not a latency cost in the normal case, and it only ever delays a
    Stop that is already going badly.
@@ -421,6 +417,6 @@ scenario must log `settled=false` and `no-park:cancelled`. That proves the guard
    Codex result shows the interesting variation is in what the harness does with it, not whether it
    accepts it.
 
-Two things deliberately left as they are, flagged so nobody re-opens them by accident: a cancelled
-turn still drops its continuity record (decide with work package D, since it depends on the
-immutable-history choice), and `clientGone` still always destroys (a disconnect is not a Stop).
+Every settled Stop preserves the continuity row and native session, regardless of its best-effort
+Codex reap outcome. Only a harness cancel that does not settle invalidates continuity and falls back
+to cold replay. A plain `clientGone` still destroys because a disconnect is not a Stop.
diff --git a/services/runner/src/engines/sandbox_agent/cancel-turn.ts b/services/runner/src/engines/sandbox_agent/cancel-turn.ts
index 1f9d73fbf7..72e773adb3 100644
--- a/services/runner/src/engines/sandbox_agent/cancel-turn.ts
+++ b/services/runner/src/engines/sandbox_agent/cancel-turn.ts
@@ -98,16 +98,6 @@ export async function cancelHarnessTurn(
 
   const now = input.now ?? (() => Date.now());
   const startedAt = now();
-  try {
-    await cancelSession.call(input.sandbox, input.sessionId);
-  } catch (error) {
-    input.log(
-      "stage=harness_cancel sent=false error=" +
-        (error instanceof Error ? error.message : String(error)).slice(0, 160),
-    );
-    return unsettled;
-  }
-
   const timeoutMs = input.timeoutMs ?? resolveCancelSettleMs();
   const wait =
     input.wait ??
@@ -116,8 +106,27 @@ export async function cancelHarnessTurn(
         const handle = setTimeout(resolve, ms);
         handle.unref?.();
       }));
-
   const TIMED_OUT = Symbol("cancel-settle-timeout");
+
+  try {
+    const requested = await Promise.race([
+      cancelSession.call(input.sandbox, input.sessionId).then(() => true),
+      wait(timeoutMs).then(() => TIMED_OUT),
+    ]);
+    if (requested === TIMED_OUT) {
+      input.log(
+        `stage=harness_cancel sent=false reason=request-timeout budget_ms=${timeoutMs}`,
+      );
+      return unsettled;
+    }
+  } catch (error) {
+    input.log(
+      "stage=harness_cancel sent=false error=" +
+        (error instanceof Error ? error.message : String(error)).slice(0, 160),
+    );
+    return unsettled;
+  }
+
   // A RESOLVED prompt is the harness reporting its own `stopReason`. A REJECTED one means the
   // prompt died on the transport instead, which says nothing about whether the harness stopped,
   // so it counts as unsettled and the environment is destroyed.
diff --git a/services/runner/src/engines/sandbox_agent/reap-exec.ts b/services/runner/src/engines/sandbox_agent/reap-exec.ts
index 753f72e68e..fb489cad88 100644
--- a/services/runner/src/engines/sandbox_agent/reap-exec.ts
+++ b/services/runner/src/engines/sandbox_agent/reap-exec.ts
@@ -39,10 +39,8 @@
  * that was just stopped. An MCP server starts when the session is created, before the prompt, so
  * it is always older than the turn and is never selected.
  *
- * WHY A FAILURE IS NOT A DESTROY. The reap is best effort and cannot change the park decision. A
- * sandbox that would have been parked is still parked when the reap cannot run, because trading a
- * warm session away for a tidier process table is the wrong trade. The cost of not reaping is
- * bounded by the park window; the cost of destroying is a cold start on the user's next message.
+ * WHY A FAILURE DESTROYS. A parked sandbox must not retain a command from the stopped turn. Only a
+ * successful kill or a successful inspection that finds nothing to reap proves parking is safe.
  */
 
 /** One row of `ps -eo pid=,ppid=,etimes=,args=`. */
@@ -209,6 +207,15 @@ export interface ReapResult {
     | "kill-failed";
 }
 
+/** True when best-effort cleanup needs QA follow-up. */
+export function reapResultHasCleanupMiss(
+  result: ReapResult | undefined,
+): boolean {
+  return (
+    !result || (result.killed === 0 && result.skipped !== "nothing-to-reap")
+  );
+}
+
 /**
  * Best effort. Never throws, and every outcome is one log line the release gate can assert on.
  */
@@ -233,8 +240,7 @@ export async function reapLeakedExecChildren(
     rows = parseProcessTable(listing.stdout ?? "");
     if (rows.length === 0) throw new Error("no parseable rows");
   } catch (error) {
-    // A sandbox image without a `ps` that understands `-eo` lands here. That is a reason to leave
-    // the leak alone, never a reason to delete a sandbox the user is about to write to.
+    // A sandbox image without a compatible `ps` cannot prove that parking is safe.
     input.log(
       "stage=harness_reap killed=0 skipped=ps-failed error=" +
         (error instanceof Error ? error.message : String(error)).slice(0, 120),
diff --git a/services/runner/src/engines/sandbox_agent/run-turn.ts b/services/runner/src/engines/sandbox_agent/run-turn.ts
index e59027ae4d..dfa7f09702 100644
--- a/services/runner/src/engines/sandbox_agent/run-turn.ts
+++ b/services/runner/src/engines/sandbox_agent/run-turn.ts
@@ -70,7 +70,10 @@ import {
 import { noteExecutionSettled } from "../../sessions/execution-registry.ts";
 import { isUserStopAbort } from "../../sessions/stop-signal.ts";
 import { cancelHarnessTurn } from "./cancel-turn.ts";
-import { reapLeakedExecChildren } from "./reap-exec.ts";
+import {
+  reapLeakedExecChildren,
+  reapResultHasCleanupMiss,
+} from "./reap-exec.ts";
 import { sandboxAgentServerPort } from "./provider.ts";
 import { PAUSED, PendingApprovalPauseController } from "./pause.ts";
 import {
@@ -1125,8 +1128,7 @@ export async function runTurn(
     // from one the SESSION started earlier (an stdio MCP server). A resumed turn keeps the
     // resume's own start, which only ever makes the reap more conservative. See `reap-exec.ts`.
     let promptStartedAtMs = Date.now();
-    const approvalTransition =
-      opts.resume ?? opts.settleApprovalsThenPrompt;
+    const approvalTransition = opts.resume ?? opts.settleApprovalsThenPrompt;
     if (approvalTransition) {
       // The resume turn owns continued events; each decision answers one parked gate by id.
       // Carried gates keep the shared original prompt pending until a later answer.
@@ -1382,14 +1384,25 @@ export async function runTurn(
       // Codex leaves its shell child running inside the sandbox we are about to park; Pi and
       // Claude kill theirs. Reap it here, never in the bridge: the Codex shell is a child of a
       // vendored Rust binary the JS bridge holds no pid for, and a bridge patch would ship only
-      // through a Daytona snapshot rebuild. Best effort, and it cannot change the park decision.
+      // through a Daytona snapshot rebuild. This cleanup is best effort; the stopped TTL bounds
+      // leftovers without changing the harness-confirmed park and continuity decision.
       if (cancel.settled && plan.acpAgent === "codex") {
-        await reapLeakedExecChildren({
+        let reapError: unknown;
+        const reap = await reapLeakedExecChildren({
           sandbox: env.sandbox,
           sandboxAgentPort: sandboxAgentServerPort(env.sandbox?.sandboxId),
           turnElapsedMs: Date.now() - promptStartedAtMs,
           log: logger,
-        }).catch(() => undefined);
+        }).catch((error) => {
+          reapError = error;
+          return undefined;
+        });
+        if (reapResultHasCleanupMiss(reap)) {
+          logger(
+            `stage=harness_reap cleanup_miss=true skipped=${reap?.skipped ?? "unknown"}` +
+              (reapError ? ` error=${String(reapError).slice(0, 120)}` : ""),
+          );
+        }
       }
       // The harness has been asked to stop, so the Pi trace port and the environment teardown must
       // not ask again. Their `destroySession` also aborts `env.mcpAbort`, which belongs to the
diff --git a/services/runner/src/environment/mount-lifecycle.ts b/services/runner/src/environment/mount-lifecycle.ts
index c5b5e60e52..be45fce574 100644
--- a/services/runner/src/environment/mount-lifecycle.ts
+++ b/services/runner/src/environment/mount-lifecycle.ts
@@ -208,9 +208,9 @@ export async function mountLocalDurableCwd(
     creds,
     { log: ctx.log, signal: deps.signal },
   );
-  throwIfAcquireAborted(deps.signal);
   if (mounted) {
     ctx.commitLocalMount("cwd", plan.workspace.cwd, creds);
+    throwIfAcquireAborted(deps.signal);
     // Session-local links belong to the mount's lifecycle, not to first acquire: this mount is
     // object storage, which has no symlinks, so a remount hands back a 0-byte file where the link
     // was. Re-materialize the subscription Codex login link here, AFTER the mount is live
@@ -224,6 +224,7 @@ export async function mountLocalDurableCwd(
     }
     return true;
   }
+  throwIfAcquireAborted(deps.signal);
   // A false result means mountStorage stopped the attempt and CONFIRMED the path detached.
   ctx.markCwdDetachConfirmed();
   return false;
@@ -252,8 +253,8 @@ export async function mountLocalAgentCwd(
       rmSync(mountPath, { recursive: true, force: true });
       return false;
     }
-    throwIfAcquireAborted(deps.signal);
     ctx.commitLocalMount("agent", mountPath, creds);
+    throwIfAcquireAborted(deps.signal);
     await seedAgentReadme(mountPath, { log: ctx.log });
     await linkAgentFiles(plan.workspace.cwd, mountPath, { log: ctx.log });
     await activateAgentMountGuidance(ctx, deps);
diff --git a/services/runner/src/server.ts b/services/runner/src/server.ts
index 72eb8d62da..794548d600 100644
--- a/services/runner/src/server.ts
+++ b/services/runner/src/server.ts
@@ -794,12 +794,8 @@ async function runAndStreamWithApiBaseResolved(
     // A throw escaping run() itself (outside the engine's own try/catch) emitted no error
     // event — persist it here as the backstop.
     if (persistError) persistError(message);
-    if (
-      !terminalRecordEmitted &&
-      persistTerminal &&
-      isUserStopAbort(controller.signal)
-    ) {
-      persistTerminal("cancelled");
+    if (!terminalRecordEmitted && persistTerminal) {
+      persistTerminal(isUserStopAbort(controller.signal) ? "cancelled" : undefined);
     }
     if (flushPersist) await flushPersist().catch(() => {});
     result = { ok: false, error: message };
diff --git a/services/runner/src/sessions/alive.ts b/services/runner/src/sessions/alive.ts
index fad5c9a41b..eeeef93baf 100644
--- a/services/runner/src/sessions/alive.ts
+++ b/services/runner/src/sessions/alive.ts
@@ -136,8 +136,8 @@ export function ownedSessionCount(now: number = Date.now()): number {
  * uuid — the free gift of a call the runner already makes every turn, no new round-trip) and
  * `interrupted: true` when the API reports `is_current_turn: false` (a cancel/steer/kill took
  * this turn's alive/running lock since the last beat — W7.4, the control-signal path). A
- * network/HTTP failure yields `{ streamId: undefined, interrupted: false }` (fail-open: a
- * transient API blip must neither abort a healthy run nor fabricate a stream id).
+ * network/HTTP failure yields `confirmed: false`; callers use that to fail closed for initial
+ * admission while later watchdog beats remain best effort for a turn already admitted.
  */
 async function sendHeartbeat(
   sessionId: string,
@@ -145,7 +145,11 @@ async function sendHeartbeat(
   authorization: string,
   isRunning = true,
   proposal?: SessionProposal,
-): Promise<{ streamId: string | undefined; interrupted: boolean }> {
+): Promise<{
+  streamId: string | undefined;
+  interrupted: boolean;
+  confirmed: boolean;
+}> {
   try {
     const url = `${apiBase()}/sessions/streams/heartbeat`;
     const res = await fetch(url, {
@@ -168,7 +172,7 @@ async function sendHeartbeat(
     });
     if (!res.ok) {
       log(`heartbeat HTTP ${res.status} session=${sessionId} turn=${turnId}`);
-      return { streamId: undefined, interrupted: false };
+      return { streamId: undefined, interrupted: false, confirmed: false };
     }
     const body = (await res.json()) as {
       stream?: { id?: unknown } | null;
@@ -190,12 +194,12 @@ async function sendHeartbeat(
     log(
       `heartbeat OK session=${sessionId} turn=${turnId} running=${isRunning}${interrupted ? " INTERRUPTED" : ""}`,
     );
-    return { streamId, interrupted };
+    return { streamId, interrupted, confirmed: true };
   } catch (err) {
     log(
       `heartbeat failed session=${sessionId} turn=${turnId}: ${String(err instanceof Error ? err.message : err).slice(0, 120)}`,
     );
-    return { streamId: undefined, interrupted: false };
+    return { streamId: undefined, interrupted: false, confirmed: false };
   }
 }
 
@@ -265,9 +269,8 @@ export async function claimSessionOwnership(
  * touches the sandbox is what makes at-most-one-execution-per-session true: a refused turn stops
  * at the edge instead of reaching the keepalive pool and destroying the live turn's environment.
  *
- * `admitted` is false ONLY on an explicit `is_current_turn: false`. A network or HTTP failure
- * fails OPEN (`admitted: true`), matching every other use of this beat: a transient API blip must
- * not refuse a healthy turn. The keepalive pool's own busy check is the backstop for that window.
+ * Initial admission fails closed unless the coordination plane confirms this turn owns the lock.
+ * Later heartbeat failures remain best effort and do not abort an already-admitted healthy turn.
  *
  * `proposal` rides EVERY beat rather than only the first. The server fills each field once, so
  * repeating them is a no-op, and one payload for all beats beats a "was this the first?" flag.
@@ -348,9 +351,8 @@ export async function startAliveWatchdog(
   }
 
   return {
-    // Read from the FIRST beat only. A later interruption is a cancel, not a failed admission,
-    // and it travels the `onInterrupted` -> abort path instead.
-    admitted: !first.interrupted,
+    // Read from the FIRST beat only. Later interruptions travel the abort path instead.
+    admitted: first.confirmed && !first.interrupted,
     async release() {
       clearInterval(interval);
       credentialLease.release();
@@ -384,9 +386,14 @@ export async function releaseSessionOwnership(
   timeoutMs?: number,
 ): Promise<boolean> {
   try {
+    const runnerToken = process.env.AGENTA_RUNNER_TOKEN?.trim();
     const res = await fetch(`${apiBase()}/sessions/streams/heartbeat`, {
       method: "POST",
-      headers: { "content-type": "application/json", authorization },
+      headers: {
+        "content-type": "application/json",
+        authorization,
+        ...(runnerToken ? { "x-agenta-runner-token": runnerToken } : {}),
+      },
       body: JSON.stringify({
         session_id: sessionId,
         replica_id: REPLICA_ID,
diff --git a/services/runner/tests/unit/cancel-continuity.test.ts b/services/runner/tests/unit/cancel-continuity.test.ts
index 48bca6204d..61dd41122f 100644
--- a/services/runner/tests/unit/cancel-continuity.test.ts
+++ b/services/runner/tests/unit/cancel-continuity.test.ts
@@ -42,6 +42,8 @@ interface CancelFakeOpts {
   onPrompt?: () => void;
   /** Model the shell child Codex leaves behind after answering a cancelled prompt. */
   leakedCodexChild?: boolean;
+  /** Force Codex's best-effort post-cancel reap to fail in a known or unexpected way. */
+  codexReapFailure?: "failed" | "unknown";
 }
 
 /**
@@ -100,6 +102,9 @@ function fakeCancellableSandbox(opts: CancelFakeOpts = {}) {
     async runProcess(request: { command: string; args?: string[] }) {
       if (request.command === "ps") {
         calls.lifecycle.push("ps");
+        if (opts.codexReapFailure === "failed") {
+          throw new Error("ps unavailable");
+        }
         return {
           stdout: [
             "100 1 120 /x/bin/sandbox-agent server --port 3000",
@@ -118,6 +123,13 @@ function fakeCancellableSandbox(opts: CancelFakeOpts = {}) {
       return { stdout: "", exitCode: 0 };
     },
   };
+  if (opts.codexReapFailure === "unknown") {
+    Object.defineProperty(sandbox, "runProcess", {
+      get() {
+        throw new Error("reap inspection unavailable");
+      },
+    });
+  }
   if (opts.cancellable !== false) {
     sandbox.cancelSession = async (id: string) => {
       calls.lifecycle.push("cancel");
@@ -317,6 +329,32 @@ describe("a stopped turn's continuity record", () => {
     assert.deepEqual(fake.calls.lifecycle, ["cancel", "ps", "kill", "park"]);
   });
 
+  for (const codexReapFailure of ["failed", "unknown"] as const) {
+    it(`keeps a settled Codex Stop warm after a ${codexReapFailure} reap`, async () => {
+      const { calls, continuityStore, deps, signal } = fakeAbortingSandbox({
+        codexReapFailure,
+      });
+
+      const result = await runSandboxAgent(
+        { ...stopRequest, harness: "codex" },
+        undefined,
+        signal,
+        deps,
+      );
+
+      assert.equal(result.ok, true);
+      assert.equal(result.cancelSettled, true);
+      assert.equal(calls.paused, 1, "a settled Stop still parks");
+      assert.equal(calls.destroyed, 0);
+      assert.equal(calls.completed.length, 1, "continuity stays durable");
+      assert.equal(
+        continuityStore.get("sess-stop", "codex")?.agentSessionId,
+        AGENT_SESSION_ID,
+      );
+      assert.ok(calls.logs.some((line) => line.includes("cleanup_miss=true")));
+    });
+  }
+
   it("writes the record even when the abort was not a user Stop and the sandbox is deleted", async () => {
     // A disconnect deletes the sandbox, but the harness still confirmed it is idle and its
     // native session lives on the durable cwd, so the record stays worth keeping: the next turn
diff --git a/services/runner/tests/unit/control-command-apply.test.ts b/services/runner/tests/unit/control-command-apply.test.ts
index c55ce2a128..825d0bc3bf 100644
--- a/services/runner/tests/unit/control-command-apply.test.ts
+++ b/services/runner/tests/unit/control-command-apply.test.ts
@@ -292,7 +292,13 @@ describe("applyCommand", () => {
       /parked approval harness cancel did not settle/,
     );
 
-    assert.deepEqual(journal, ["reject", "cancel", "timeout", "teardown"]);
+    assert.deepEqual(journal, [
+      "reject",
+      "cancel",
+      "timeout",
+      "timeout",
+      "teardown",
+    ]);
     assert.equal(env.parkedApprovals.size, 1);
     assert.equal(env.sessionDestroyRequested, true);
   });
diff --git a/services/runner/tests/unit/harness-cancel-park.test.ts b/services/runner/tests/unit/harness-cancel-park.test.ts
index d647d3b553..8ccc1d8066 100644
--- a/services/runner/tests/unit/harness-cancel-park.test.ts
+++ b/services/runner/tests/unit/harness-cancel-park.test.ts
@@ -9,7 +9,7 @@
  *  3. The parked reason is on the teardown allowlist, so the sandbox is stopped, not deleted.
  */
 import assert from "node:assert/strict";
-import { describe, it } from "vitest";
+import { afterEach, beforeEach, describe, it } from "vitest";
 
 import {
   cancelHarnessTurn,
@@ -130,6 +130,21 @@ describe("cancelHarnessTurn", () => {
     assert.equal(result.settled, false);
   });
 
+  it("bounds a cancel request that never answers", async () => {
+    const logs: string[] = [];
+    const result = await cancelHarnessTurn({
+      sandbox: { cancelSession: never },
+      sessionId: "sess-1",
+      promptPromise: Promise.resolve({ stopReason: "cancelled" }),
+      timeoutMs: 5_000,
+      wait: async () => {},
+      log: (message) => logs.push(message),
+    });
+
+    assert.deepEqual(result, { settled: false, requested: false, elapsedMs: 0 });
+    assert.ok(logs.some((line) => line.includes("reason=request-timeout")));
+  });
+
   it("keeps a settle budget a user would wait through", () => {
     assert.ok(DEFAULT_CANCEL_SETTLE_MS > 0);
     assert.ok(DEFAULT_CANCEL_SETTLE_MS <= 30_000);
@@ -239,6 +254,29 @@ describe("the cancelled teardown reason", () => {
 });
 
 describe("the stopped-session park window", () => {
+  const ttlEnvNames = [
+    "AGENTA_RUNNER_SESSION_TTL_MS",
+    "AGENTA_RUNNER_SESSION_APPROVAL_TTL_MS",
+    "AGENTA_RUNNER_SESSION_STOPPED_TTL_MS",
+    "AGENTA_RUNNER_DAYTONA_SESSION_IDLE_TTL_MS",
+  ] as const;
+  let savedTtlEnv: Record<string, string | undefined>;
+
+  beforeEach(() => {
+    savedTtlEnv = Object.fromEntries(
+      ttlEnvNames.map((name) => [name, process.env[name]]),
+    );
+    for (const name of ttlEnvNames) delete process.env[name];
+  });
+
+  afterEach(() => {
+    for (const name of ttlEnvNames) {
+      const value = savedTtlEnv[name];
+      if (value === undefined) delete process.env[name];
+      else process.env[name] = value;
+    }
+  });
+
   // A settled Stop gets the same ten-minute human-response window on both providers. The
   // ordinary idle windows remain shorter and continue to govern clean completed turns.
   it("defaults a local stopped session to the approval window", () => {
@@ -256,16 +294,12 @@ describe("the stopped-session park window", () => {
 
   it("moves with its own env var, without touching the ordinary idle window", () => {
     process.env.AGENTA_RUNNER_SESSION_STOPPED_TTL_MS = "300000";
-    try {
-      const local = readKeepaliveConfig("local");
-      const daytona = readKeepaliveConfig("daytona");
-      assert.equal(local.stoppedTtlMs, 300_000);
-      assert.equal(local.ttlMs, 60_000);
-      assert.equal(daytona.stoppedTtlMs, 300_000);
-      assert.equal(daytona.ttlMs, 120_000);
-    } finally {
-      delete process.env.AGENTA_RUNNER_SESSION_STOPPED_TTL_MS;
-    }
+    const local = readKeepaliveConfig("local");
+    const daytona = readKeepaliveConfig("daytona");
+    assert.equal(local.stoppedTtlMs, 300_000);
+    assert.equal(local.ttlMs, 60_000);
+    assert.equal(daytona.stoppedTtlMs, 300_000);
+    assert.equal(daytona.ttlMs, 120_000);
   });
 });
 
diff --git a/services/runner/tests/unit/mount-lifecycle.test.ts b/services/runner/tests/unit/mount-lifecycle.test.ts
new file mode 100644
index 0000000000..69a590cc31
--- /dev/null
+++ b/services/runner/tests/unit/mount-lifecycle.test.ts
@@ -0,0 +1,98 @@
+import assert from "node:assert/strict";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { describe, it } from "vitest";
+
+import type { AcquireContext } from "../../src/environment/acquire-context.ts";
+import {
+  mountLocalAgentCwd,
+  mountLocalDurableCwd,
+  type MountDeps,
+} from "../../src/environment/mount-lifecycle.ts";
+
+const credentials = {
+  endpoint: "http://store",
+  region: "eu-central-1",
+  bucket: "bucket",
+  prefix: "prefix",
+  accessKey: "access",
+  secretKey: "secret",
+};
+
+const depsFor = (
+  signal: AbortSignal,
+  mountStorage: MountDeps["mountStorage"],
+): MountDeps => ({
+  mountStorage,
+  signMount: async () => null,
+  signAgentMount: async () => null,
+  daytonaPiDir: "/tmp/pi",
+  signal,
+});
+
+const contextFor = (cwd: string, commits: string[]): AcquireContext =>
+  ({
+    plan: {
+      acpAgent: "pi",
+      isDaytona: false,
+      workspace: { cwd },
+    },
+    env: {
+      mountCreds: credentials,
+      agentMountCreds: credentials,
+    },
+    sessionForMount: "session-1",
+    artifactId: "artifact-1",
+    log: () => {},
+    beginCwdMount: () => {},
+    markCwdDetachConfirmed: () => {},
+    commitLocalMount: (kind: string) => commits.push(kind),
+  }) as unknown as AcquireContext;
+
+describe("local mount cancellation", () => {
+  it("commits a durable cwd mount before observing an abort", async () => {
+    const cwd = mkdtempSync(join(tmpdir(), "agenta-mount-cwd-"));
+    const controller = new AbortController();
+    const commits: string[] = [];
+
+    try {
+      await assert.rejects(
+        mountLocalDurableCwd(
+          contextFor(cwd, commits),
+          depsFor(controller.signal, async () => {
+            controller.abort();
+            return true;
+          }),
+          "initial",
+        ),
+        { name: "AbortError" },
+      );
+      assert.deepEqual(commits, ["cwd"]);
+    } finally {
+      rmSync(cwd, { recursive: true, force: true });
+    }
+  });
+
+  it("commits an agent mount before its abort is handled", async () => {
+    const cwd = mkdtempSync(join(tmpdir(), "agenta-mount-agent-"));
+    const controller = new AbortController();
+    const commits: string[] = [];
+
+    try {
+      const mounted = await mountLocalAgentCwd(
+        contextFor(cwd, commits),
+        depsFor(controller.signal, async () => {
+          controller.abort();
+          return true;
+        }),
+      );
+
+      assert.equal(mounted, false);
+      assert.deepEqual(commits, ["agent"]);
+    } finally {
+      rmSync(cwd, { recursive: true, force: true });
+      rmSync(`${cwd}-agent`, { recursive: true, force: true });
+    }
+  });
+});
diff --git a/services/runner/tests/unit/reap-exec.test.ts b/services/runner/tests/unit/reap-exec.test.ts
index bbc1e1f6b0..6f190aa014 100644
--- a/services/runner/tests/unit/reap-exec.test.ts
+++ b/services/runner/tests/unit/reap-exec.test.ts
@@ -13,6 +13,7 @@ import {
   findSandboxAgentServerPid,
   parseProcessTable,
   reapLeakedExecChildren,
+  reapResultHasCleanupMiss,
   selectLeakedExecPids,
 } from "../../src/engines/sandbox_agent/reap-exec.ts";
 import {
@@ -22,6 +23,19 @@ import {
 
 const LIVE_PORT = 43_123;
 
+describe("reapResultHasCleanupMiss", () => {
+  it("flags failed and unknown cleanup for QA", () => {
+    expect(reapResultHasCleanupMiss({ killed: 1 })).toBe(false);
+    expect(
+      reapResultHasCleanupMiss({ killed: 0, skipped: "nothing-to-reap" }),
+    ).toBe(false);
+    expect(reapResultHasCleanupMiss({ killed: 0, skipped: "ps-failed" })).toBe(
+      true,
+    );
+    expect(reapResultHasCleanupMiss(undefined)).toBe(true);
+  });
+});
+
 /** The real tree, copied from the live probe on the integration stack (2026-09-03). */
 const LIVE_PS = [
   "    1     0  50000 /sbin/docker-init -- docker-entrypoint.sh sh -c node scripts/build-extension.mjs",
diff --git a/services/runner/tests/unit/server.test.ts b/services/runner/tests/unit/server.test.ts
index 4535502256..281cba9b69 100644
--- a/services/runner/tests/unit/server.test.ts
+++ b/services/runner/tests/unit/server.test.ts
@@ -882,6 +882,73 @@ describe("createAgentServer", () => {
     }
   });
 
+  it("persists one terminal done record when a session-owned run throws", async () => {
+    const s = await listen(async () => {
+      throw new Error("engine escaped");
+    });
+    const realFetch = globalThis.fetch.bind(globalThis);
+    const ingested: Array<Record<string, any>> = [];
+    const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
+    const fetchSpy = vi
+      .spyOn(globalThis, "fetch")
+      .mockImplementation(async (input, init) => {
+        const url = String(input);
+        if (url === `${s.url}/run`) return realFetch(input, init);
+        if (url.endsWith("/sessions/streams/heartbeat")) {
+          return Response.json({
+            stream: { id: "stream-escaped-run" },
+            is_current_turn: true,
+          });
+        }
+        if (url.endsWith("/sessions/records/ingest")) {
+          ingested.push(JSON.parse(String(init?.body)));
+        }
+        return Response.json({});
+      });
+
+    try {
+      const response = await fetchSpy(`${s.url}/run`, {
+        method: "POST",
+        headers: { accept: "application/x-ndjson", ...AUTH },
+        body: JSON.stringify({
+          harness: "pi_core",
+          sessionId: "session-escaped-run",
+          runContext: { project: { id: "project-1" } },
+          telemetry: {
+            exporters: {
+              otlp: {
+                endpoint: `${s.url}/otlp/v1/traces`,
+                headers: { authorization: "Test platform authorization" },
+              },
+            },
+          },
+          messages: [{ role: "user", content: "throw" }],
+        }),
+      });
+      const records = (await response.text())
+        .trim()
+        .split("\n")
+        .map((line) => JSON.parse(line) as Record<string, any>);
+
+      assert.deepEqual(
+        ingested
+          .filter((record) => ["error", "done"].includes(record.record_type))
+          .map((record) => record.record_type),
+        ["error", "done"],
+      );
+      assert.equal(
+        ingested.filter((record) => record.record_type === "done").length,
+        1,
+      );
+      assert.equal(records.filter((record) => record.kind === "result").length, 1);
+      assert.equal(records.at(-1)?.result.error, "engine escaped");
+    } finally {
+      fetchSpy.mockRestore();
+      errorSpy.mockRestore();
+      await s.close();
+    }
+  });
+
   it("rejects an over-cap session turn before persistence or attachment claiming", async () => {
     // Override the cap rather than generating a default-sized batch, so the case stays small.
     process.env.AGENTA_ATTACHMENTS_MAX_PER_TURN = "2";
diff --git a/services/runner/tests/unit/session-admission.test.ts b/services/runner/tests/unit/session-admission.test.ts
index d0f0e878d4..6829f267fc 100644
--- a/services/runner/tests/unit/session-admission.test.ts
+++ b/services/runner/tests/unit/session-admission.test.ts
@@ -457,10 +457,7 @@ describe("runner admission: an admitted turn proceeds", () => {
     }
   });
 
-  it("fails OPEN: an unreachable platform admits the turn rather than refusing it", async () => {
-    // The heartbeat has always failed open, and admission must not change that: a transient API
-    // blip refusing every message would be a worse outage than the bug this slice fixes. The
-    // keepalive pool's busy check is the backstop for the window this leaves.
+  it("fails closed when the coordination plane cannot confirm admission", async () => {
     process.env[INTERNAL_ENV] = "http://127.0.0.1:1";
     const runCalls: AgentRunRequest[] = [];
     const runner = await startRunner(async (request): Promise<AgentRunResult> => {
@@ -470,9 +467,10 @@ describe("runner admission: an admitted turn proceeds", () => {
     try {
       const { records } = await postRun(runner.url, sessionRequest());
 
-      assert.equal(runCalls.length, 1, "an unreachable arbiter does not refuse the turn");
+      assert.equal(runCalls.length, 0, "an unconfirmed turn must never reach run()");
       const terminal = records.find((r) => r.kind === "result");
-      assert.equal(terminal!.result!.ok, true);
+      assert.equal(terminal!.result!.ok, false);
+      assert.equal(terminal!.result!.error, SESSION_TURN_IN_USE_MESSAGE);
     } finally {
       await runner.close();
     }
diff --git a/services/runner/tests/unit/session-alive-interrupt.test.ts b/services/runner/tests/unit/session-alive-interrupt.test.ts
index 8b14ac0ad8..e466907d7b 100644
--- a/services/runner/tests/unit/session-alive-interrupt.test.ts
+++ b/services/runner/tests/unit/session-alive-interrupt.test.ts
@@ -164,14 +164,13 @@ describe("startAliveWatchdog admitted (single-turn admission)", () => {
     await watchdog.release();
   });
 
-  it("fails OPEN: an unreachable API admits the turn", async () => {
-    // A transient blip refusing every message would be a worse outage than the bug this closes.
-    // The keepalive pool's busy check is the backstop for the window this leaves open.
+  it("fails closed when the admission API is unreachable", async () => {
+    // Without an affirmative first heartbeat, the runner cannot prove it owns this turn.
     vi.stubGlobal("fetch", async () => {
       throw new Error("network down");
     });
     const watchdog = await startAliveWatchdog("sess-c", "turn-c", "proj-1");
-    assert.equal(watchdog.admitted, true);
+    assert.equal(watchdog.admitted, false);
     await watchdog.release();
   });
 
diff --git a/services/runner/tests/unit/session-ownership-release.test.ts b/services/runner/tests/unit/session-ownership-release.test.ts
index 246096b496..b24cc7601b 100644
--- a/services/runner/tests/unit/session-ownership-release.test.ts
+++ b/services/runner/tests/unit/session-ownership-release.test.ts
@@ -15,7 +15,11 @@
 import { describe, it, beforeEach, afterEach, vi } from "vitest";
 import assert from "node:assert/strict";
 
-const fetchCalls: Array<{ url: string; body: any }> = [];
+const fetchCalls: Array<{
+  url: string;
+  body: any;
+  headers?: RequestInit["headers"];
+}> = [];
 let fetchImpl: (
   url: string,
   init?: RequestInit,
@@ -24,7 +28,7 @@ let fetchImpl: (
 
 vi.stubGlobal("fetch", async (url: string, init?: RequestInit) => {
   const body = init?.body ? JSON.parse(init.body as string) : undefined;
-  fetchCalls.push({ url, body });
+  fetchCalls.push({ url, body, headers: init?.headers });
   return fetchImpl(url, init);
 });
 
@@ -46,6 +50,7 @@ const ownedBy = (replica: string) => async () =>
 beforeEach(() => {
   fetchCalls.length = 0;
   fetchImpl = ownedBy(REPLICA_ID);
+  process.env.AGENTA_RUNNER_TOKEN = "runner-secret";
 });
 
 afterEach(async () => {
@@ -54,6 +59,7 @@ afterEach(async () => {
     forgetOwnedSession(id);
   }
   vi.restoreAllMocks();
+  delete process.env.AGENTA_RUNNER_TOKEN;
 });
 
 describe("learning which sessions this replica owns", () => {
@@ -118,6 +124,10 @@ describe("the shutdown release", () => {
       assert.ok(call.url.endsWith("/sessions/streams/heartbeat"));
       assert.equal(call.body.release_owner, true);
       assert.equal(call.body.replica_id, REPLICA_ID);
+      assert.equal(
+        (call.headers as Record<string, string>)["x-agenta-runner-token"],
+        "runner-secret",
+      );
       assert.equal(
         call.body.turn_id,
         undefined,
diff --git a/web/oss/src/components/AgentChatSlice/AgentConversation.tsx b/web/oss/src/components/AgentChatSlice/AgentConversation.tsx
index db5b2697c2..1a30c244ae 100644
--- a/web/oss/src/components/AgentChatSlice/AgentConversation.tsx
+++ b/web/oss/src/components/AgentChatSlice/AgentConversation.tsx
@@ -56,6 +56,7 @@ import {TEMPLATE_STRIP_MODE} from "@/oss/components/pages/agent-home/assets/cons
 import {isAgentFileUploadsEnabled} from "./assets/constants"
 import {CONTENT_VISIBILITY_ENABLED} from "./assets/conversationLayout"
 import {runWithInFlightSubmit} from "./assets/inFlightSubmit"
+import {restoreHeldRefusedSend} from "./assets/refusedMessageRecovery"
 import AgentComposerDock from "./components/AgentComposerDock"
 import AgentTranscript from "./components/AgentTranscript"
 import AgentTurn from "./components/AgentTurn"
@@ -242,6 +243,7 @@ const AgentConversation = ({
         attachmentsSettled,
         isDragging,
         addFiles,
+        restoreAttachments,
     } = attachments
 
     // Playground-native onboarding: the hero, Create-agent / Continue-in-IDE, the template strip
@@ -430,20 +432,27 @@ const AgentConversation = ({
             }),
         [messages],
     )
-    // Single-turn admission (#6417, #5539, #5538): the backend refuses a message sent while
-    // another turn is already running on this session. Nothing ran and nothing was sent, so the
-    // user's text goes back into the composer instead of vanishing. Without this the refusal is
-    // worse than the bug for the person typing: they lose what they wrote and have no way to get
-    // it back.
-    //
-    // The rAF mirrors the edit-stash restore above it: `submitEditorAsMarkdown` clears the editor
-    // synchronously after `onSubmit` returns, so a restore has to land after that clear.
+    const refusedSendRef = useRef<QueuedMessage | undefined>(undefined)
+    const restoreRefusedSend = useCallback(
+        () => restoreHeldRefusedSend(refusedSendRef, richInputRef.current, restoreAttachments),
+        [restoreAttachments],
+    )
+    // Restore a refused send after the editor's synchronous submit clear.
     useEffect(() => {
         if (!error || !isSessionBusyRefusal(error)) return
-        const sent = takeLastSent()
-        if (!sent?.text) return
-        requestAnimationFrame(() => richInputRef.current?.setMarkdown(sent.text))
-    }, [error, takeLastSent])
+        if (!refusedSendRef.current) refusedSendRef.current = takeLastSent()
+        requestAnimationFrame(() => {
+            restoreRefusedSend()
+        })
+    }, [error, restoreRefusedSend, takeLastSent])
+
+    const handleComposerChange = useCallback(
+        (text: string) => {
+            composer.handleComposerChange(text)
+            if (!text.trim()) restoreRefusedSend()
+        },
+        [composer.handleComposerChange, restoreRefusedSend],
+    )
 
     useEffect(() => {
         const status: SessionRunStatus = error
@@ -539,11 +548,12 @@ const AgentConversation = ({
         trimmed: string,
         fileParts: FileUIPart[] | undefined,
         consumedUids: string[],
+        stagedFiles: typeof files,
     ) => {
         if (editingId) {
             // A rewrite of a held message: nothing is sent, so the transcript must not move.
             // The input clears itself on submit, so the displaced draft goes back after that.
-            const draft = commitEdit({text: trimmed, fileParts})
+            const draft = commitEdit({text: trimmed, fileParts, stagedFiles})
             if (draft) requestAnimationFrame(() => richInputRef.current?.setMarkdown(draft))
         } else {
             // Glide to the bottom; the min-h-full active turn makes that show the new question at the
@@ -552,7 +562,7 @@ const AgentConversation = ({
             scrollIntent.armGlide()
             setStopped(false)
             // One path: `submit` sends now or queues behind held messages via the shared release gate.
-            submit({text: trimmed, fileParts})
+            submit({text: trimmed, fileParts, stagedFiles})
         }
         // The message left the composer — drop its persisted draft (and any pending capture).
         composer.clearDraft()
@@ -593,7 +603,7 @@ const AgentConversation = ({
                     }
                     fileParts = parts
                 }
-                finishSubmit(trimmed, fileParts, stagedUids)
+                finishSubmit(trimmed, fileParts, stagedUids, files)
                 return
             }
 
@@ -606,7 +616,7 @@ const AgentConversation = ({
             const fileParts = outboundFiles.length
                 ? stagedFilesToParts(outboundFiles, sessionId)
                 : undefined
-            finishSubmit(trimmed, fileParts, stagedUids)
+            finishSubmit(trimmed, fileParts, stagedUids, outboundFiles)
         })
 
     handleSubmitRef.current = handleSubmit
@@ -864,7 +874,7 @@ const AgentConversation = ({
                                         onStop={handleStop}
                                         stopping={stopping}
                                         richInputRef={richInputRef}
-                                        composer={composer}
+                                        composer={{...composer, handleComposerChange}}
                                         attachments={attachments}
                                         onboardingChat={onboardingChat}
                                         voice={voice}
diff --git a/web/oss/src/components/AgentChatSlice/assets/refusedMessageRecovery.test.ts b/web/oss/src/components/AgentChatSlice/assets/refusedMessageRecovery.test.ts
new file mode 100644
index 0000000000..9c4b51f9e6
--- /dev/null
+++ b/web/oss/src/components/AgentChatSlice/assets/refusedMessageRecovery.test.ts
@@ -0,0 +1,86 @@
+import {describe, expect, it, vi} from "vitest"
+
+import {
+    canRestoreRefusedSend,
+    restoreRefusedDraft,
+    restoreHeldRefusedSend,
+    restoreRefusedSend,
+} from "./refusedMessageRecovery"
+
+describe("restoreRefusedDraft", () => {
+    it("restores a refused message only into an empty composer", () => {
+        const setMarkdown = vi.fn()
+        const editor = {getMarkdown: () => "", setMarkdown} as never
+
+        expect(restoreRefusedDraft(editor, "try again")).toBe(true)
+        expect(setMarkdown).toHaveBeenCalledWith("try again")
+    })
+
+    it("does not overwrite a newer draft", () => {
+        const setMarkdown = vi.fn()
+        const editor = {getMarkdown: () => "new draft", setMarkdown} as never
+
+        expect(restoreRefusedDraft(editor, "old refused message")).toBe(false)
+        expect(setMarkdown).not.toHaveBeenCalled()
+    })
+
+    it("allows attachment recovery only while the composer is still empty", () => {
+        expect(canRestoreRefusedSend({getMarkdown: () => "", setMarkdown: vi.fn()} as never)).toBe(
+            true,
+        )
+        expect(
+            canRestoreRefusedSend({getMarkdown: () => "new draft", setMarkdown: vi.fn()} as never),
+        ).toBe(false)
+    })
+
+    it("leaves a refused send with staged attachments untouched behind a newer draft", () => {
+        const setMarkdown = vi.fn()
+        const restoreAttachments = vi.fn()
+        const stagedFiles = [{uid: "file-1", name: "brief.pdf"}]
+        const editor = {getMarkdown: () => "newer draft", setMarkdown} as never
+
+        expect(
+            restoreRefusedSend(editor, {text: "refused message", stagedFiles}, restoreAttachments),
+        ).toBe(false)
+        expect(setMarkdown).not.toHaveBeenCalled()
+        expect(restoreAttachments).not.toHaveBeenCalled()
+    })
+
+    it("captures a refusal before deferred placement and restores it once", () => {
+        let markdown = "newer draft"
+        const setMarkdown = vi.fn((next: string) => {
+            markdown = next
+        })
+        const restoreAttachments = vi.fn()
+        const stagedFiles = [{uid: "file-1", name: "brief.pdf"}]
+        const refused = {text: "refused message", stagedFiles}
+        const newer = {text: "newer draft", stagedFiles: []}
+        let lastSent: typeof refused | undefined = refused
+        const takeLastSent = () => {
+            const sent = lastSent
+            lastSent = undefined
+            return sent
+        }
+        const slot: {current: typeof refused | undefined} = {current: undefined}
+        const editor = {getMarkdown: () => markdown, setMarkdown} as never
+        const frames: (() => boolean)[] = []
+
+        expect(slot.current).toBeUndefined()
+        if (!slot.current) slot.current = takeLastSent()
+        frames.push(() => restoreHeldRefusedSend(slot, editor, restoreAttachments))
+
+        lastSent = newer
+        markdown = ""
+        expect(frames.shift()?.()).toBe(true)
+
+        expect(setMarkdown).toHaveBeenCalledTimes(1)
+        expect(setMarkdown).toHaveBeenCalledWith("refused message")
+        expect(restoreAttachments).toHaveBeenCalledTimes(1)
+        expect(restoreAttachments).toHaveBeenCalledWith(stagedFiles)
+        expect(lastSent).toBe(newer)
+
+        expect(restoreHeldRefusedSend(slot, editor, restoreAttachments)).toBe(false)
+        expect(setMarkdown).toHaveBeenCalledTimes(1)
+        expect(restoreAttachments).toHaveBeenCalledTimes(1)
+    })
+})
diff --git a/web/oss/src/components/AgentChatSlice/assets/refusedMessageRecovery.ts b/web/oss/src/components/AgentChatSlice/assets/refusedMessageRecovery.ts
new file mode 100644
index 0000000000..3219d1fca6
--- /dev/null
+++ b/web/oss/src/components/AgentChatSlice/assets/refusedMessageRecovery.ts
@@ -0,0 +1,43 @@
+import type {RichChatInputHandle} from "@agenta/ui/rich-chat-input"
+
+export const canRestoreRefusedSend = (editor: RichChatInputHandle | null): boolean =>
+    Boolean(editor && editor.getMarkdown() === "")
+
+export const restoreRefusedDraft = (editor: RichChatInputHandle | null, text: string): boolean => {
+    if (!editor || !text || !canRestoreRefusedSend(editor)) return false
+    editor.setMarkdown(text)
+    return true
+}
+
+interface RefusedSend<TAttachment> {
+    text: string
+    stagedFiles?: TAttachment[]
+}
+
+interface RefusedSendSlot<TAttachment> {
+    current: RefusedSend<TAttachment> | undefined
+}
+
+export const restoreRefusedSend = <TAttachment>(
+    editor: RichChatInputHandle | null,
+    sent: RefusedSend<TAttachment>,
+    restoreAttachments: (files: TAttachment[]) => void,
+): boolean => {
+    if (!canRestoreRefusedSend(editor)) return false
+    if (sent.text && !restoreRefusedDraft(editor, sent.text)) return false
+    if (sent.stagedFiles?.length) restoreAttachments(sent.stagedFiles)
+    return true
+}
+
+export const restoreHeldRefusedSend = <TAttachment>(
+    slot: RefusedSendSlot<TAttachment>,
+    editor: RichChatInputHandle | null,
+    restoreAttachments: (files: TAttachment[]) => void,
+): boolean => {
+    const sent = slot.current
+    if (!sent) return false
+    slot.current = undefined
+    if (restoreRefusedSend(editor, sent, restoreAttachments)) return true
+    slot.current = sent
+    return false
+}
diff --git a/web/oss/src/components/AgentChatSlice/components/AgentMessage.tsx b/web/oss/src/components/AgentChatSlice/components/AgentMessage.tsx
index cc2e83f398..443ddb59c5 100644
--- a/web/oss/src/components/AgentChatSlice/components/AgentMessage.tsx
+++ b/web/oss/src/components/AgentChatSlice/components/AgentMessage.tsx
@@ -163,12 +163,7 @@ const RETRYABLE_CODES = new Set([
     "execution_lost",
 ])
 
-/**
- * Single-turn admission refused the message because another turn already owns the session
- * (#6417). Nothing ran and nothing failed, so the failure header would be a lie. The composer
- * already has the user's text back (see AgentConversation's restore effect), which is why there is
- * no retry button either: sending again is one keystroke away and only the user knows when.
- */
+// An admission refusal means the message was not sent, not that an agent run failed.
 const NOT_SENT_CODES = new Set([SESSION_TURN_IN_USE_CODE])
 
 /** The ONE rule driving both the clamp and the toggle — they can't disagree and hide text (#5350). */
diff --git a/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts b/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts
index 139fb28465..f6572babe0 100644
--- a/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts
+++ b/web/packages/agenta-chat/src/hooks/useAgentChatQueue.ts
@@ -5,10 +5,15 @@ import {canReleaseQueuedMessage, isHitlPending} from "@agenta/playground/agent-c
 import {generateId} from "@agenta/shared/utils"
 import type {FileUIPart, UIMessage} from "ai"
 
+import {latestTurnId} from "../assets/agentTurn"
+
+import type {ComposerAttachment} from "./useComposerAttachments"
+
 export interface QueuedMessage {
     id: string
     text: string
     fileParts?: FileUIPart[]
+    stagedFiles?: ComposerAttachment[]
 }
 
 interface UseAgentChatQueueArgs {
@@ -83,30 +88,25 @@ export const useAgentChatQueue = ({
         queuedRef.current = queued
     }, [queued])
 
-    /**
-     * The message this mount sent immediately, held until something claims it.
-     *
-     * A QUEUED message survives a failed turn on its own — it is in `queued`, which the dock
-     * renders and the store mirrors. An immediately-sent one had nowhere to live: `submit` handed
-     * it to `sendQueued` and dropped the object, so a send the backend refuses lost the user's
-     * text with no trace. `takeLastSent` is how the host gets it back and puts it in the composer.
-     *
-     * NOT re-queued automatically: the queue releases on a settled `"error"` status, which for a
-     * refusal ("another turn is running") would re-send and be refused again in a tight loop. The
-     * user decides when to send again.
-     */
+    // Retained until admission so a refused immediate send can return to the composer.
     const lastSentRef = useRef<QueuedMessage | undefined>(undefined)
 
-    /** Take back the last immediately-sent message, once. */
-    const takeLastSent = useCallback(() => {
+    const admittedTurnId = latestTurnId(messages)
+    useEffect(() => {
+        if (admittedTurnId) lastSentRef.current = undefined
+    }, [admittedTurnId])
+
+    /** Take back the last sent message only after an optional placement succeeds. */
+    const takeLastSent = useCallback((place?: (message: QueuedMessage) => boolean) => {
         const message = lastSentRef.current
+        if (!message || (place && !place(message))) return undefined
         lastSentRef.current = undefined
         return message
     }, [])
 
     // Send now only if idle, unlatched, and the queue is empty; otherwise append (FIFO).
     const submit = useCallback(
-        (item: {text: string; fileParts?: FileUIPart[]}) => {
+        (item: {text: string; fileParts?: FileUIPart[]; stagedFiles?: ComposerAttachment[]}) => {
             const message: QueuedMessage = {...item, id: generateId()}
             if (!releasingRef.current && queuedRef.current.length === 0 && canReleaseNow) {
                 releasingRef.current = true
@@ -164,7 +164,7 @@ export const useAgentChatQueue = ({
      * so the text the session displaced has to come back here too or it is lost for good.
      */
     const commitEdit = useCallback(
-        (item: {text: string; fileParts?: FileUIPart[]}) => {
+        (item: {text: string; fileParts?: FileUIPart[]; stagedFiles?: ComposerAttachment[]}) => {
             const id = editingId
             setEditingId(null)
             const draft = takeStash()
@@ -174,6 +174,7 @@ export const useAgentChatQueue = ({
                 return draft
             }
             const fileParts = [...(target.fileParts ?? []), ...(item.fileParts ?? [])]
+            const stagedFiles = [...(target.stagedFiles ?? []), ...(item.stagedFiles ?? [])]
             // Edited down to nothing and carrying no files: there is no message left to hold.
             if (!item.text.trim() && fileParts.length === 0) {
                 setQueued((q) => q.filter((m) => m.id !== id))
@@ -186,6 +187,7 @@ export const useAgentChatQueue = ({
                               ...m,
                               text: item.text,
                               fileParts: fileParts.length ? fileParts : undefined,
+                              stagedFiles: stagedFiles.length ? stagedFiles : undefined,
                           }
                         : m,
                 ),
@@ -209,8 +211,7 @@ export const useAgentChatQueue = ({
         releasingRef.current = true
         const [head, ...rest] = queued
         setQueued(rest)
-        // Reclaimable for the same reason as the immediate path: the release removed it from the
-        // queue, so a refusal would otherwise lose it.
+        // A released head also needs refusal recovery because it has left the queue.
         lastSentRef.current = head
         sendQueued(head)
     }, [settled, canReleaseNow, queued, sendQueued])
diff --git a/web/packages/agenta-chat/src/hooks/useComposerAttachments.ts b/web/packages/agenta-chat/src/hooks/useComposerAttachments.ts
index ad45639de5..1a722de212 100644
--- a/web/packages/agenta-chat/src/hooks/useComposerAttachments.ts
+++ b/web/packages/agenta-chat/src/hooks/useComposerAttachments.ts
@@ -14,7 +14,8 @@ import {attachmentsBySession} from "../state/sessionEphemera"
 
 import {removeUploadFile, useAttachmentUploads} from "./useAttachmentUploads"
 
-type StagedFile = UploadFile<SessionAttachmentResponse>
+export type ComposerAttachment = UploadFile<SessionAttachmentResponse>
+type StagedFile = ComposerAttachment
 
 /** Convert settled upload-tray entries into reference `file` parts via the neutral builder. */
 export const stagedFilesToParts = (files: StagedFile[], sessionId: string) =>
@@ -289,12 +290,12 @@ export const useComposerAttachments = ({
      * than through `addFiles`, which would re-upload them as second attachments. Idempotent:
      * anything already back in the tray is left where it is.
      */
-    const restoreAttachments = (restored: StagedFile[]) => {
+    const restoreAttachments = useCallback((restored: StagedFile[]) => {
         setFiles((prev) => [
             ...restored.filter((file) => !prev.some((row) => row.uid === file.uid)),
             ...prev,
         ])
-    }
+    }, [])
 
     return {
         uploadsEnabled,
diff --git a/web/packages/agenta-chat/src/model/error.ts b/web/packages/agenta-chat/src/model/error.ts
index f5c15c6688..61d4bb6ee1 100644
--- a/web/packages/agenta-chat/src/model/error.ts
+++ b/web/packages/agenta-chat/src/model/error.ts
@@ -46,36 +46,17 @@ export const isTransportFailure = (raw: string): boolean => {
     return TRANSPORT_MESSAGES.includes(bare)
 }
 
-/**
- * The runner refuses a message sent while another turn is already running on the same session,
- * so at most one execution runs per session (#6417, #5539, #5538). Nothing ran, nothing was
- * destroyed, and the message was never sent — so this is NOT a run failure, and the client keeps
- * the user's text instead of losing it.
- *
- * The message text is the contract with the runner. It is produced in exactly one place,
- * `services/runner/src/sessions/admission.ts`, and reaches the browser verbatim: the SDK's
- * `sanitize_runner_error` passes a clean one-line message through unchanged, and the Vercel
- * egress puts it on the stream as `errorText`. Keep the two constants byte-identical.
- */
+// Keep this refusal contract byte-identical to the runner message.
 export const SESSION_TURN_IN_USE_CODE = "session_turn_in_use"
 
 export const SESSION_TURN_IN_USE_MESSAGE =
     "This session is already running a turn. Your message was not sent. Wait for the reply, or stop the turn, then send again."
 
-/**
- * True when a `useChat` stream error is the single-turn admission refusal.
- *
- * Matched on the message rather than on the stream's `data-agent-error` code because the `error`
- * object is the only thing available at the moment the client has to decide whether to give the
- * user their text back. The code still travels on the message part and drives how the bubble
- * renders (`getMessageRunErrorCode`).
- */
+/** True when a `useChat` error is the single-turn admission refusal. */
 export const isSessionBusyRefusal = (err: unknown): boolean =>
     parseAgentRunError(err).message.trim() === SESSION_TURN_IN_USE_MESSAGE
 
-// Copied verbatim from web/oss/src/components/AgentChatSlice/AgentConversation.tsx
-// (2026-07-25); the OSS original remains authoritative for the desktop chat until the
-// re-plumb PR deletes it. Keep byte-parity if either side changes.
+// Keep byte parity with the desktop parser until its duplicate is removed.
 /**
  * Best-effort human reason from a useChat stream error: a plain string or a `{status:{…}}`
  * envelope. An engine's own wording is translated — "Failed to fetch" under "The agent run
@@ -107,8 +88,7 @@ export const parseAgentRunError = (err: unknown): ParsedRunError => {
         // Carry the class so the bubble can say "not sent" rather than "the agent run failed".
         return {message: fallback, code: SESSION_TURN_IN_USE_CODE}
     }
-    // After the envelope: a server that reports those words means them, and its code is worth more
-    // than this translation. A bare engine string has no envelope to lose.
+    // A server envelope outranks transport-phrase translation.
     if (isTransportFailure(fallback)) return {message: TRANSPORT_ERROR_MESSAGE, transport: true}
     return {message: fallback}
 }
diff --git a/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts b/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts
index 6456017047..41b8aa1116 100644
--- a/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts
+++ b/web/packages/agenta-chat/tests/unit/hooks/useAgentChatQueue.test.ts
@@ -364,11 +364,7 @@ describe("useAgentChatQueue", () => {
 })
 
 describe("useAgentChatQueue: reclaiming a sent message", () => {
-    // Single-turn admission (#6417) refuses a message sent while another turn owns the session.
-    // A QUEUED message survives that on its own — it is still in `queued`. An immediately-sent one
-    // had nowhere to live: `submit` handed it to `sendQueued` and dropped it, so a refused send
-    // lost the user's text with no trace. `takeLastSent` is how the host puts it back in the
-    // composer.
+    // The host reclaims an immediate send only until the runner confirms admission.
 
     it("hands back the message that was sent immediately", () => {
         const {result} = setup(settledEmpty)
@@ -387,6 +383,55 @@ describe("useAgentChatQueue: reclaiming a sent message", () => {
         expect(result.current.takeLastSent()).toBeUndefined()
     })
 
+    it("keeps an attachment-only refused send recoverable", () => {
+        const {result} = setup(settledEmpty)
+        const stagedFiles = [{uid: "file-1", name: "brief.pdf", status: "done"}] as never
+        act(() => {
+            result.current.submit({text: "", stagedFiles})
+        })
+
+        expect(result.current.takeLastSent()).toMatchObject({text: "", stagedFiles})
+    })
+
+    it("retains a refused send when the composer cannot place it", () => {
+        const {result} = setup(settledEmpty)
+        const stagedFiles = [{uid: "file-1", name: "brief.pdf", status: "done"}] as never
+        act(() => {
+            result.current.submit({text: "refused message", stagedFiles})
+        })
+
+        expect(result.current.takeLastSent(() => false)).toBeUndefined()
+        expect(result.current.takeLastSent()).toMatchObject({
+            text: "refused message",
+            stagedFiles,
+        })
+    })
+
+    it("does not clear recovery when dispatch only changes the stream status", () => {
+        const {result, rerender} = setup(settledEmpty)
+        act(() => {
+            result.current.submit({text: "sent"})
+        })
+        rerender({status: "streaming", messages: [userTurn("u1", "sent")], stopped: false})
+        expect(result.current.takeLastSent()?.text).toBe("sent")
+    })
+
+    it("clears recovery after a runner turn id confirms admission", () => {
+        const {result, rerender} = setup(settledEmpty)
+        act(() => {
+            result.current.submit({text: "admitted"})
+        })
+        rerender({
+            status: "streaming",
+            messages: [
+                userTurn("u2", "admitted"),
+                {...assistantText("a2", ""), metadata: {turnId: "turn-2"}},
+            ],
+            stopped: false,
+        })
+        expect(result.current.takeLastSent()).toBeUndefined()
+    })
+
     it("has nothing to hand back for a message that only QUEUED", () => {
         // A queued message is already safe: it is rendered by the dock and mirrored per session.
         const {result, sendQueued} = setup({status: "streaming", messages: [], stopped: false})
diff --git a/web/packages/agenta-chat/tests/unit/model/error.test.ts b/web/packages/agenta-chat/tests/unit/model/error.test.ts
index 6ba635c5a3..d1df74f894 100644
--- a/web/packages/agenta-chat/tests/unit/model/error.test.ts
+++ b/web/packages/agenta-chat/tests/unit/model/error.test.ts
@@ -84,10 +84,7 @@ describe("parseAgentRunError", () => {
 })
 
 describe("single-turn admission refusal", () => {
-    // The runner refuses a message sent while another turn owns the session (#6417, #5539, #5538).
-    // Nothing ran and nothing was sent, so the client keeps the user's text instead of losing it.
-    // The message text is the contract with `services/runner/src/sessions/admission.ts`; it reaches
-    // the browser verbatim through the SDK's `sanitize_runner_error` and the Vercel egress.
+    // The runner refusal message is the browser recovery contract.
 
     it("recognises the refusal and carries its stable class", () => {
         expect(parseAgentRunError(new Error(SESSION_TURN_IN_USE_MESSAGE))).toEqual({
diff --git a/web/packages/agenta-chat/tests/unit/model/userStop.test.ts b/web/packages/agenta-chat/tests/unit/model/userStop.test.ts
index 398e24acfb..5af63de766 100644
--- a/web/packages/agenta-chat/tests/unit/model/userStop.test.ts
+++ b/web/packages/agenta-chat/tests/unit/model/userStop.test.ts
@@ -47,18 +47,16 @@ const reduce = (
 
 describe("user stopped state", () => {
     it("keeps a remounted turn guarded until its durable Stop settles", () => {
-        expect(
-            isSessionTurnStopping({currentTurnId: "turn-1", stoppingTurnId: "turn-1"}),
-        ).toBe(true)
-        expect(isSessionTurnStopping({currentTurnId: "turn-1", stoppingTurnId: null})).toBe(
-            false,
+        expect(isSessionTurnStopping({currentTurnId: "turn-1", stoppingTurnId: "turn-1"})).toBe(
+            true,
         )
+        expect(isSessionTurnStopping({currentTurnId: "turn-1", stoppingTurnId: null})).toBe(false)
     })
 
     it("does not apply a stale Stop marker to a newer turn", () => {
-        expect(
-            isSessionTurnStopping({currentTurnId: "turn-2", stoppingTurnId: "turn-1"}),
-        ).toBe(false)
+        expect(isSessionTurnStopping({currentTurnId: "turn-2", stoppingTurnId: "turn-1"})).toBe(
+            false,
+        )
     })
 
     it("maps a stream-delivered cancelled ending to the neutral state", () => {