diff --git a/loopx/chat_ingress.py b/loopx/chat_ingress.py
index b05594b796..5ba04b72b8 100644
--- a/loopx/chat_ingress.py
+++ b/loopx/chat_ingress.py
@@ -72,6 +72,72 @@ class ChatIngressStore:
     def runtime_activation_lock(self):
         return chat_runtime_activation_lock(self.root.parent)
 
+    def queued_turns(self, session_id: str) -> list[dict[str, Any]]:
+        from .chat_store import _read_json, CHAT_TURN_SCHEMA_VERSION
+
+        session = self.load_session(session_id)
+        if session is None:
+            raise KeyError("chat session was not found")
+        active_turn_id = str(session.get("active_turn_id") or "")
+        turns_dir = self._session_dir(session_id) / "turns"
+        rows: list[dict[str, Any]] = []
+        for path in turns_dir.glob("*.json") if turns_dir.is_dir() else []:
+            if path.name.endswith(".events.json"):
+                continue
+            payload = _read_json(path)
+            if (
+                payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION
+                and payload.get("status") == "queued"
+                and str(payload.get("turn_id") or "") != active_turn_id
+            ):
+                rows.append(payload)
+        return sorted(
+            rows,
+            key=lambda item: (
+                str(item.get("created_at") or ""),
+                str(item.get("turn_id") or ""),
+            ),
+        )
+
+    def complete_recovered_terminalization(
+        self, session_id: str, turn_id: str,
+        session: dict[str, Any] | None, turn: dict[str, Any] | None,
+        *, recovery_messages: dict[str, str],
+    ) -> None:
+        """Finish the lifecycle owner's already-persisted failure, never dispatch.
+
+        State precedes the event, error message and claim release. If one of
+        those writes failed, retain the exact recovery result and idempotently
+        finish only its missing writes under the caller's execution fence.
+        """
+        from .chat_store import CHAT_SESSION_MODE_MANAGED, utc_now
+
+        if (session is None or turn is None
+                or session.get("session_mode") != CHAT_SESSION_MODE_MANAGED
+                or turn.get("status") != "failed"):
+            return
+        code = str(turn.get("error_code") or "")
+        message = recovery_messages.get(code)
+        if message is None or (code != "server_restarted" and turn.get("error") != message):
+            return  # Match this owner's recorded recovery, not arbitrary failures.
+        if not any(
+            event.get("kind") == "turn.failed"
+            and (event.get("payload") or {}).get("error_code") == code
+            and (event.get("payload") or {}).get("message") == message
+            for event in self.events_after(session_id, turn_id, None)
+        ):
+            self.append_event(session_id, turn_id, kind="turn.failed",
+                              payload={"error_code": code, "message": message})
+        if not any(
+            item.get("turn_id") == turn_id and item.get("role") == "error"
+            and item.get("text") == message for item in self.messages(session_id)
+        ):
+            self.append_message(session_id, role="error", text=message, turn_id=turn_id)
+        if session.get("active_turn_id") == turn_id:
+            self.release_active_turn(session_id, turn_id,
+                                     last_activity_at=str(turn.get("completed_at") or utc_now()),
+                                     last_error_code=code)
+
     def create_queued_turn(
         self,
         session_id: str,
diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py
index 6928dbf477..d9e8507e84 100644
--- a/loopx/chat_runtime.py
+++ b/loopx/chat_runtime.py
@@ -51,7 +51,10 @@
 from .control_plane.turn_driver.host_binding import MANAGED_TURN_HOST
 from .control_plane.turn_driver.execution_profile import managed_execution_profile
 from .control_plane.chat_turn_acceptance import managed_turn_recovery_allowed
-from .file_lock import LOCK_HOLDER_DEAD, lock_holder_liveness, try_exclusive_file_lock
+from .file_lock import (
+    LOCK_HOLDER_DEAD, LOCK_HOLDER_LIVE, LOCK_HOLDER_RELEASED,
+    _lock_id, lock_holder_host_label, lock_holder_liveness, try_exclusive_file_lock,
+)
 from .kiro_cli_goal_mode import (
     KIRO_CLI_BIN,
     KIRO_CLI_CHAT_AGENT_ID,
@@ -108,6 +111,11 @@
     "Your conversation is saved; this turn was not retried automatically. "
     "You can continue with a follow-up in this session."
 )
+WORKER_EXITED_MESSAGE = (
+    "The LoopX Chat execution worker exited before this turn finished. "
+    "Your conversation is saved; execution may have occurred. "
+    "This turn was not retried automatically."
+)
 
 
 class ChatTurnAcceptanceUnavailableError(Exception):
@@ -1420,6 +1428,13 @@ def _drain_session_queue(
                     return
                 active_turn_id = str(session.get("active_turn_id") or "")
                 if active_turn_id:
+                    try:
+                        self.recover_exited_turn(session_id, active_turn_id)
+                    except OSError:
+                        # Storage can still be unavailable. Retry only the
+                        # non-executing lifecycle repair, never the old input.
+                        self.closed.wait(0.25)
+                        continue
                     with self.lock:
                         attached = self.adapters.get(session_id)
                     if attached is None or not attached.healthcheck():
@@ -1474,14 +1489,20 @@ def _drain_session_queue(
                 done_event = threading.Event()
                 with self.lock:
                     self.turn_done_events[(session_id, turn_id)] = done_event
-                self._run_turn(
-                    session_id=session_id,
-                    turn_id=turn_id,
-                    message=str(turn.get("message") or ""),
-                    attachments=turn.get("attachments") or [],
-                    adapter=adapter,
-                    done_event=done_event,
-                )
+                try:
+                    self._run_turn(
+                        session_id=session_id,
+                        turn_id=turn_id,
+                        message=str(turn.get("message") or ""),
+                        attachments=turn.get("attachments") or [],
+                        adapter=adapter,
+                        done_event=done_event,
+                    )
+                except OSError:
+                    # _run_turn has released its execution fence/done event.
+                    # A failed error checkpoint must not kill this queue worker;
+                    # the next iteration reconciles the original durable state.
+                    self.closed.wait(0.25)
         finally:
             with self.lock:
                 worker = self.session_queue_threads.get(session_id)
@@ -1527,10 +1548,11 @@ def _fail_queue_preparation(
                         self.turn_done_events.pop(key, None)
 
     def recover_exited_turn(self, session_id: str, turn_id: str | None = None) -> None:
-        """Reconcile a killed local worker without starting a provider or replay.
+        """Reconcile a lost local execution without starting a provider or replay.
 
-        The execution lock records its host and PID for the worker's lifetime.
-        No record, a live/foreign holder, or an attached host proves no loss.
+        The kernel fence spans the worker, while the PID spans the service.
+        A failed terminal write can end one worker without ending that PID.
+        Unknown/foreign records and a held execution fence never prove loss.
         """
         session = self.store.load_session(session_id)
         if not session:
@@ -1539,13 +1561,23 @@ def recover_exited_turn(self, session_id: str, turn_id: str | None = None) -> No
         if not turn_id:
             return
         path = self.store.turn_execution_lock_path(session_id, turn_id)
-        if lock_holder_liveness(path)[0] != LOCK_HOLDER_DEAD:
+        state, observed = lock_holder_liveness(path)
+        if (state not in {LOCK_HOLDER_DEAD, LOCK_HOLDER_LIVE, LOCK_HOLDER_RELEASED}
+                or observed.get("host") != lock_holder_host_label()
+                or type(observed.get("pid")) is not int or observed["pid"] <= 0
+                or observed.get("lock_id") != _lock_id(path)
+                or observed.get("agent_id") != "loopx-chat"
+                or observed.get("operation") != "execute_chat_turn"
+                or observed.get("policy") != "single_flight"):
             return
         # Take the execution fence without consuming its death evidence, so
         # dependency failures remain retryable. Recheck under the fence: a
         # replacement may have changed the holder since the first liveness read.
         with try_exclusive_file_lock(path, preserve_holder=True) as ownership:
-            if ownership is None or lock_holder_liveness(path)[0] != LOCK_HOLDER_DEAD:
+            state, current = lock_holder_liveness(path)
+            if (ownership is None or current != observed
+                    or state not in {LOCK_HOLDER_DEAD, LOCK_HOLDER_LIVE, LOCK_HOLDER_RELEASED}
+                    or current.get("host") != lock_holder_host_label()):
                 return
             session = self.store.load_session(session_id)
             turn = self.store.load_turn(session_id, turn_id)
@@ -1553,8 +1585,10 @@ def recover_exited_turn(self, session_id: str, turn_id: str | None = None) -> No
                 session, turn, managed=bool(session and session.get("session_mode") == CHAT_SESSION_MODE_MANAGED),
                 owner_dead=True,
             ):
+                code, message = (("server_restarted", SERVER_RESTARTED_MESSAGE)
+                                 if state == LOCK_HOLDER_DEAD else ("runtime_error", WORKER_EXITED_MESSAGE))
                 self._fail_turn(
-                    session_id, turn_id, "server_restarted", SERVER_RESTARTED_MESSAGE,
+                    session_id, turn_id, code, message,
                     status="failed", expected_statuses={"starting", "running", "interrupting"},
                 )
             else:
@@ -1562,61 +1596,12 @@ def recover_exited_turn(self, session_id: str, turn_id: str | None = None) -> No
                 # and then lost its event, message or claim write. The typed
                 # owner rejects `failed`, so only this narrow completion may
                 # finish that result; it never reopens or replays the Turn.
-                self._complete_recovered_terminalization(
+                self.store.complete_recovered_terminalization(
                     session_id, turn_id, session, turn,
+                    recovery_messages={"server_restarted": SERVER_RESTARTED_MESSAGE,
+                                       "runtime_error": WORKER_EXITED_MESSAGE},
                 )
 
-    def _complete_recovered_terminalization(
-        self,
-        session_id: str,
-        turn_id: str,
-        session: dict[str, Any] | None,
-        turn: dict[str, Any] | None,
-    ) -> None:
-        """Idempotently finish a recovery that already persisted `failed`.
-
-        `_fail_turn` writes the state first and then the terminal event, error
-        message and claim release. A transient store error in those follow-up
-        writes leaves the exact recovery result partially applied, and the
-        typed owner no longer admits the Turn. Repair only that exact result.
-        """
-        if (
-            session is None
-            or turn is None
-            or session.get("session_mode") != CHAT_SESSION_MODE_MANAGED
-            or turn.get("status") != "failed"
-            or turn.get("error_code") != "server_restarted"
-        ):
-            return
-        message = SERVER_RESTARTED_MESSAGE
-        if not any(
-            event.get("kind") == "turn.failed"
-            and (event.get("payload") or {}).get("error_code") == "server_restarted"
-            for event in self.store.events_after(session_id, turn_id, None)
-        ):
-            self.store.append_event(
-                session_id,
-                turn_id,
-                kind="turn.failed",
-                payload={"error_code": "server_restarted", "message": message},
-            )
-        if not any(
-            item.get("turn_id") == turn_id
-            and item.get("role") == "error"
-            and item.get("text") == message
-            for item in self.store.messages(session_id)
-        ):
-            self.store.append_message(
-                session_id, role="error", text=message, turn_id=turn_id,
-            )
-        if session.get("active_turn_id") == turn_id:
-            self.store.release_active_turn(
-                session_id,
-                turn_id,
-                last_activity_at=str(turn.get("completed_at") or utc_now()),
-                last_error_code="server_restarted",
-            )
-
     def _run_turn(
         self,
         *,
diff --git a/loopx/chat_store.py b/loopx/chat_store.py
index fababacc89..84f1720a1b 100644
--- a/loopx/chat_store.py
+++ b/loopx/chat_store.py
@@ -208,7 +208,7 @@ def _replace_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
 class ChatSessionStore(ChatIngressStore):
     """Filesystem store kept outside project and public LoopX run history."""
 
-    def __init__(self, runtime_root: Path) -> None:
+    def __init__(self, runtime_root: Path, *, startup_maintenance: bool = True) -> None:
         self.root = runtime_root.expanduser().resolve() / "chat"
         self.sessions_root = self.root / "sessions"
         self._session_lock_guard = threading.Lock()
@@ -220,8 +220,11 @@ def __init__(self, runtime_root: Path) -> None:
         self.sessions_root.mkdir(parents=True, exist_ok=True, mode=0o700)
         os.chmod(self.root, 0o700)
         os.chmod(self.sessions_root, 0o700)
-        self.compact_completed_events()
-        self._recover_turn_completions()
+        # A scoped lifecycle repair must not compact unrelated history or finish
+        # other pending completions merely by opening their data owner.
+        if startup_maintenance:
+            self.compact_completed_events()
+            self._recover_turn_completions()
 
     def _session_dir(self, session_id: str) -> Path:
         token = _opaque_id(session_id, field="session_id")
@@ -1199,31 +1202,6 @@ def accept_managed_turn(
                     dispatch_reason=plan.dispatch_reason,
                 )
 
-    def queued_turns(self, session_id: str) -> list[dict[str, Any]]:
-        session = self.load_session(session_id)
-        if session is None:
-            raise KeyError("chat session was not found")
-        active_turn_id = str(session.get("active_turn_id") or "")
-        turns_dir = self._session_dir(session_id) / "turns"
-        rows: list[dict[str, Any]] = []
-        for path in turns_dir.glob("*.json") if turns_dir.is_dir() else []:
-            if path.name.endswith(".events.json"):
-                continue
-            payload = _read_json(path)
-            if (
-                payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION
-                and payload.get("status") == "queued"
-                and str(payload.get("turn_id") or "") != active_turn_id
-            ):
-                rows.append(payload)
-        return sorted(
-            rows,
-            key=lambda item: (
-                str(item.get("created_at") or ""),
-                str(item.get("turn_id") or ""),
-            ),
-        )
-
     def _settle_expired_queued_turns(
         self,
         session_id: str,
diff --git a/loopx/runtime_activation.py b/loopx/runtime_activation.py
index 03afc56157..cfb404b3db 100644
--- a/loopx/runtime_activation.py
+++ b/loopx/runtime_activation.py
@@ -130,12 +130,23 @@ def managed_chat_activation_guard(
     with ExitStack() as locks:
         for root in roots:
             locks.enter_context(chat_runtime_activation_lock(root, timeout_seconds=2.0))
-        for root in roots:
-            if chat_runtime_activation_state(root)["nonterminal_turns"]:
-                raise RuntimeActivationDeferred("chat_work_pending", labels)
         for plist, (label, pid) in (expected_services or {}).items():
             if plist not in plists or _managed_service_pid(plist, label=label) != pid:
                 raise ValueError("chat_service_route_unresolved")
+        for root in roots:
+            if chat_runtime_activation_state(root)["nonterminal_turns"]:
+                from .chat_runtime import ChatRuntimeController
+                from .chat_store import ChatSessionStore
+
+                # Data/lifecycle recovery only: this controller opens no adapter,
+                # starts no queue worker and never resends accepted input. Keep
+                # unrelated completion/compaction out of this scoped repair.
+                store = ChatSessionStore(root, startup_maintenance=False)
+                recovery = ChatRuntimeController(store=store, codex_bin="codex")
+                for session in store.list_sessions():
+                    recovery.recover_exited_turn(str(session["session_id"]))
+                if chat_runtime_activation_state(root)["nonterminal_turns"]:
+                    raise RuntimeActivationDeferred("chat_work_pending", labels)
         yield
 
 
diff --git a/tests/test_chat_runtime_activation.py b/tests/test_chat_runtime_activation.py
index 7f773953f2..e62a68b36a 100644
--- a/tests/test_chat_runtime_activation.py
+++ b/tests/test_chat_runtime_activation.py
@@ -4,6 +4,7 @@
 
 import hashlib
 import json
+import os
 from pathlib import Path
 import plistlib
 import shlex
@@ -17,6 +18,8 @@
 from loopx.chat_ingress import chat_runtime_activation_lock, chat_runtime_activation_state
 from loopx.runtime_activation import restart_services_for_runtime_activation
 from loopx.self_update import execute_update_plan, render_update_plan_markdown
+from loopx.chat_runtime import ChatRuntimeController
+from loopx.file_lock import lock_holder_path, try_exclusive_file_lock
 
 
 @pytest.fixture
@@ -71,6 +74,239 @@ def content_hashes(root):
     }
 
 
+def lost_execution(store, *, released=False):
+    sid = session(store)
+    turn, _ = store.create_turn(sid, client_turn_id="original", message="Keep the original request.")
+    tid = turn["turn_id"]
+    path = store.turn_execution_lock_path(sid, tid)
+    with try_exclusive_file_lock(path, agent_id="loopx-chat", operation="execute_chat_turn") as held:
+        assert held is not None
+        store.update_turn(sid, tid, status="running")
+        store.append_event(sid, tid, kind="answer.delta", payload={"delta": "Saved original progress."})
+    if not released:
+        # Disk exhaustion can also prevent advisory release metadata. The real
+        # kernel lock is gone while the owning service PID remains alive.
+        record = json.loads(lock_holder_path(path).read_text())
+        record.pop("released_at")
+        lock_holder_path(path).write_text(json.dumps(record))
+    return sid, tid, path
+
+
+@pytest.mark.parametrize("released", [False, True])
+def test_same_process_worker_exit_is_recovered_without_replay(tmp_path, released):
+    store = ChatSessionStore(tmp_path)
+    sid, tid, path = lost_execution(store, released=released)
+    record = lock_holder_path(path).read_bytes()
+    assert json.loads(record)["pid"] == os.getpid()
+    runtime = ChatRuntimeController(store=store, codex_bin="must-not-execute")
+    with mock.patch.object(runtime, "_start_adapter", side_effect=AssertionError("no provider")):
+        runtime.recover_exited_turn(sid, tid)
+        failed = store.load_turn(sid, tid)
+        assert failed["status"] == "failed" and failed["error_code"] == "runtime_error"
+        assert "execution worker exited" in failed["error"]
+        assert "execution may have occurred" in failed["error"]
+        assert "not retried automatically" in failed["error"]
+        assert store.load_session(sid)["active_turn_id"] is None
+        assert [m["text"] for m in store.messages(sid) if m["role"] == "user"] == ["Keep the original request."]
+        events = store.events_after(sid, tid, None)
+        assert any(e["kind"] == "answer.delta" for e in events)
+        assert sum(e["kind"] == "turn.failed" for e in events) == 1
+        runtime.recover_exited_turn(sid, tid)
+        assert store.events_after(sid, tid, None) == events
+        assert lock_holder_path(path).read_bytes() == record
+        assert not runtime.adapters and not runtime.session_queue_threads
+
+
+@pytest.mark.parametrize("fault", ["absent", "foreign", "pid", "identity", "operation", "queued", "attached"])
+def test_unproven_worker_exit_never_releases_accepted_work(tmp_path, fault):
+    store = ChatSessionStore(tmp_path)
+    if fault == "attached":
+        sid = store.create_session(goal_id="fixture", agent_id="fixture", adapter_kind="fake",
+            upstream_thread_id="fixture", session_mode="attached_host", host_surface="codex_app")["session_id"]
+        turn, _ = store.create_turn(sid, client_turn_id="attached", message="Keep attached work.")
+        tid, path = turn["turn_id"], store.turn_execution_lock_path(sid, turn["turn_id"])
+        with try_exclusive_file_lock(path, agent_id="loopx-chat", operation="execute_chat_turn"):
+            store.update_turn(sid, tid, status="running")
+    else:
+        sid, tid, path = lost_execution(store)
+    record = json.loads(lock_holder_path(path).read_text())
+    if fault == "absent":
+        lock_holder_path(path).unlink()
+    elif fault in {"foreign", "pid", "identity", "operation"}:
+        field, value = {"foreign": ("host", "another-host"), "pid": ("pid", "unknown"),
+                        "identity": ("lock_id", "another-turn"), "operation": ("operation", "another-operation")}[fault]
+        record[field] = value
+        lock_holder_path(path).write_text(json.dumps(record))
+    elif fault == "queued":
+        store.update_turn(sid, tid, status="queued")
+    before = content_hashes(store.root)
+    ChatRuntimeController(store=store, codex_bin="must-not-execute").recover_exited_turn(sid, tid)
+    assert content_hashes(store.root) == before
+
+
+def test_activation_recovers_only_unowned_execution_and_preserves_queued_input(services):
+    root, _agents, calls = services
+    store = ChatSessionStore(root)
+    sid, tid, _path = lost_execution(store)
+    queued, _ = store.create_queued_turn(sid, client_turn_id="saved-followup", message="Original queued follow-up.")
+    before = store.load_turn(sid, queued["turn_id"])
+    with mock.patch.object(ChatRuntimeController, "_start_adapter", side_effect=AssertionError("no provider")), \
+            mock.patch.object(ChatRuntimeController, "resume_session_queue", side_effect=AssertionError("no second consumer")):
+        result = activate()
+    assert result["activation_pending"] and result["restart_reason"] == "chat_work_pending"
+    assert not calls
+    assert store.load_turn(sid, tid)["status"] == "failed"
+    assert store.load_turn(sid, queued["turn_id"]) == before
+    assert [m["text"] for m in store.messages(sid) if m["role"] == "user"] == [
+        "Keep the original request.", "Original queued follow-up."]
+
+
+def test_activation_does_not_recover_a_live_execution_fence(services):
+    root, _agents, calls = services
+    store = ChatSessionStore(root)
+    sid = session(store)
+    turn, _ = store.create_turn(sid, client_turn_id="live", message="Still executing.")
+    with try_exclusive_file_lock(store.turn_execution_lock_path(sid, turn["turn_id"]),
+            agent_id="loopx-chat", operation="execute_chat_turn"):
+        store.update_turn(sid, turn["turn_id"], status="running")
+        before = content_hashes(store.root)
+        result = activate()
+        assert result["activation_pending"] and not calls
+        assert content_hashes(store.root) == before
+
+
+@pytest.mark.parametrize("method", ["append_event", "append_message", "release_active_turn"])
+def test_worker_exit_partial_terminalization_is_retryable(tmp_path, monkeypatch, method):
+    store = ChatSessionStore(tmp_path)
+    sid, tid, path = lost_execution(store)
+    original_record = lock_holder_path(path).read_bytes()
+    runtime = ChatRuntimeController(store=store, codex_bin="must-not-execute")
+    original = getattr(store, method)
+    with monkeypatch.context() as fault:
+        fault.setattr(store, method, mock.Mock(side_effect=OSError(28, "No space left on device")))
+        with pytest.raises(OSError):
+            runtime.recover_exited_turn(sid, tid)
+    assert store.load_turn(sid, tid)["status"] == "failed"
+    assert lock_holder_path(path).read_bytes() == original_record
+    runtime.recover_exited_turn(sid, tid)
+    assert store.load_session(sid)["active_turn_id"] is None
+    assert sum(e["kind"] == "turn.failed" for e in store.events_after(sid, tid, None)) == 1
+    assert sum(m["role"] == "error" for m in store.messages(sid)) == 1
+    assert getattr(store, method) == original
+
+
+def test_scoped_activation_recovery_skips_unrelated_startup_maintenance(services):
+    root, _agents, calls = services
+    store = ChatSessionStore(root)
+    sid, tid, _ = lost_execution(store)
+    with mock.patch.object(ChatSessionStore, "compact_completed_events", side_effect=AssertionError("unrelated history")), \
+            mock.patch.object(ChatSessionStore, "_recover_turn_completions", side_effect=AssertionError("unrelated completion")):
+        result = activate()
+    assert store.load_turn(sid, tid)["status"] == "failed"
+    assert not result["activation_pending"] and len(calls) == 2
+
+
+def test_activation_rechecks_new_managed_and_attached_sessions_in_same_runtime(services):
+    root, _agents, calls = services
+    assert not activate()["activation_pending"]
+    assert len(calls) == 2
+    store = ChatSessionStore(root)
+    sid, tid, _ = lost_execution(store)
+    attached = store.create_session(goal_id="fixture", agent_id="fixture", adapter_kind="fake",
+        upstream_thread_id="attached", session_mode="attached_host", host_surface="codex_app")["session_id"]
+    turn, _ = store.create_turn(attached, client_turn_id="attached", message="Independent attached work.")
+    store.update_turn(attached, turn["turn_id"], status="running")
+    original = store.session_snapshot(attached)
+    result = activate()
+    assert store.load_turn(sid, tid)["status"] == "failed"
+    assert store.session_snapshot(attached) == original
+    assert result["activation_pending"] and result["restart_reason"] == "chat_work_pending"
+    assert len(calls) == 2
+
+
+@pytest.mark.parametrize("patch", [{"active_turn_id": None}, {"status": "closed"}])
+def test_mutable_session_fields_cannot_hide_running_work_from_activation(services, patch):
+    root, _agents, calls = services
+    store = ChatSessionStore(root)
+    sid, tid, _ = lost_execution(store)
+    store.update_session(sid, **patch)
+    before = content_hashes(store.root)
+    result = activate()
+    assert result["activation_pending"] and result["restart_reason"] == "chat_work_pending"
+    assert not calls and content_hashes(store.root) == before
+    assert store.load_turn(sid, tid)["status"] == "running"
+
+
+def test_original_queue_survives_progress_and_cleanup_disk_failure(tmp_path, monkeypatch):
+    from loopx.chat_runtime import _TurnEventBuffer
+
+    store = ChatSessionStore(tmp_path / "runtime")
+    sid = session(store)
+    first, _ = store.create_queued_turn(sid, client_turn_id="original", message="Original input.")
+    second, _ = store.create_queued_turn(sid, client_turn_id="next", message="Saved follow-up.")
+    runtime = ChatRuntimeController(store=store, codex_bin="must-not-execute")
+    disk_full, failed_cleanup, followup_done = threading.Event(), threading.Event(), threading.Event()
+    calls = []
+    update, flush, init = store.update_turn, store.flush_events, _TurnEventBuffer.__init__
+
+    def checkpoint(*args, **kwargs):
+        if disk_full.is_set() and kwargs.get("status") == "running":
+            raise OSError(28, "No space left on device")
+        return update(*args, **kwargs)
+
+    def persist(*args, **kwargs):
+        if disk_full.is_set():
+            failed_cleanup.set()
+            raise OSError(28, "No space left on device")
+        return flush(*args, **kwargs)
+
+    def buffer_init(self, **kwargs):
+        init(self, **kwargs, metadata_checkpoint_interval_sec=0)
+
+    class OriginalAdapter:
+        upstream_thread_id = "fixture"
+
+        def healthcheck(self):
+            return True
+
+        def start_turn(self, message, sink):
+            calls.append(message)
+            sink("turn.started", {"upstream_turn_id": "original-native-turn"})
+            if message == "Original input.":
+                disk_full.set()
+                sink("agent.phase", {"label": "Observed original work"})
+                raise AssertionError("the failed progress write must escape")
+            followup_done.set()
+            return {"message": "Original service answered the follow-up."}
+
+        def close_session(self):
+            pass
+
+    adapter = OriginalAdapter()
+    monkeypatch.setattr(store, "update_turn", checkpoint)
+    monkeypatch.setattr(store, "flush_events", persist)
+    monkeypatch.setattr(_TurnEventBuffer, "__init__", buffer_init)
+    monkeypatch.setattr(runtime, "_ensure_adapter", lambda *a, **k: adapter)
+    runtime.adapters[sid] = adapter
+    try:
+        runtime.resume_session_queue(session_id=sid, work_dir=tmp_path, objective="fixture")
+        assert failed_cleanup.wait(5)
+        assert runtime.session_queue_threads[sid].is_alive()
+        # Restoring storage is the only recovery stimulus: no resubmit, wake,
+        # second controller, replacement consumer or second provider input.
+        disk_full.clear()
+        assert followup_done.wait(5)
+        result = runtime.wait_for_turn(session_id=sid, turn_id=second["turn_id"], timeout_sec=5)
+        assert result["status"] == "completed", result
+        assert store.load_turn(sid, first["turn_id"])["status"] == "failed"
+        assert calls == ["Original input.", "Saved follow-up."]
+        assert [m["text"] for m in store.messages(sid) if m["role"] == "user"] == calls
+        assert store.load_session(sid)["active_turn_id"] is None
+    finally:
+        disk_full.clear()
+        runtime.close()
+
+
 @pytest.mark.parametrize("state", ["queued", "running", "completed", "unknown"])
 def test_desktop_activation_uses_original_manager_and_accepted_work(services, monkeypatch, state):
     from loopx import dashboard_launcher, runtime_activation