fix(host): release blocked bridge readers after cleanup failures

loopx/control_plane/turn_driver/host_bridge_transport.py

@@ -21,19 +21,27 @@ def __init__(self, process: subprocess.Popen[str], *, deadline: float, cleanup_t
         self.output = process.stdout
         self.process, self.deadline, self.cleanup_timeout = process, deadline, cleanup_timeout
         self.pending: queue.Queue[str | BaseException | None] = queue.Queue(maxsize=1)
+        self._stop_reading = threading.Event()
         self.requested = self.cancelled = self.not_launched = False
         self.result: dict[str, Any] | None = None
         self.reader = threading.Thread(target=self._read, name="loopx-host-bridge-reader", daemon=True)
         self.reader.start()
 
+    def _publish(self, item: str | BaseException | None) -> bool:
+        if self._stop_reading.is_set():
+            return False
+        self.pending.put(item)
+        return not self._stop_reading.is_set()
+
     def _read(self) -> None:
         try:
             for line in self.output:
-                self.pending.put(line)
+                if not self._publish(line):
+                    break
         except BaseException as error:
-            self.pending.put(error)
+            self._publish(error)
         finally:
-            self.pending.put(None)
+            self._publish(None)
 
     def next(self, *, deadline: float | None = None) -> dict[str, Any] | None:
         try:
@@ -94,6 +102,13 @@ def cancel(self) -> None:
             raise RuntimeError("managed Host bridge cleanup unconfirmed") from error
 
     def close_output(self) -> None:
+        self._stop_reading.set()
+        # No caller remains to consume frames after cleanup fails. Release the
+        # one blocked producer before joining; this does not certify Host exit.
+        try:
+            self.pending.get_nowait()
+        except queue.Empty:
+            pass
         self.reader.join(timeout=self.cleanup_timeout)
         if not self.reader.is_alive():
             self.output.close()

skills/loopx-self-repair/references/repair-patterns.md

@@ -5,6 +5,7 @@ teaches a reusable control-plane lesson.
 
 | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair |
 | --- | --- | --- | --- | --- |
+| `host_bridge_reader_orphan` | A rejected bridge observation leaves its reader thread and output pipe alive after the child has exited. | Original framing error, cleanup result, child exit, bounded-queue backpressure and reader/pipe state across repeated failures. | Cancellation rejects another malformed frame; the sole consumer leaves while the producer blocks on a full queue, and joining alone cannot release it. | Stop publication, release the waiting producer, join and close through the existing transport helper. Retain cleanup-unconfirmed results and original Host drain ownership; reclaiming a reader does not prove Host exit. Cover malformed JSON and invalid typed observations with real subprocesses, plus existing real Host cancellation and success paths. Do not increase timeouts or fabricate a successful result. |
 | `initial_preference_view_omission` | A fulfilled preference hook leaves corrections, retirement or advisory instructions absent from the recommended initial read. | The preference owner's complete current scoped view, actual selected reader/native input, empty and unavailable states, and same-capture continuation/recovery. | A hand-picked projection uses incomplete or stale child field names after the owner has already delivered the current view. | Reuse the complete owned current view in the existing initial selector; retain provenance, advisory authority, exact-scope freshness and immutable capture checks. Cover real File/SQLite lifecycle reads, empty replacements, unknown values and native consumer readback. Preserve changed-profile refusal and fresh recovery; disclose restored bytes as information cost, not compression or model benefit. |
 | `immutable_reader_parent_alias` | A warm native view refuses an unchanged reader reference after a verified capture read normalizes its parent path. | Cold descriptor bytes, warm baseline path and full digest, parent aliases versus distinct leaf names, and actual CLI recovery after compaction. | The reader accepts normal parent aliases, but immutable publication compares their literal path spellings. | Compare only equivalent parent spellings at the existing IO owner, retaining the exact leaf and every other metadata value; never rewrite the descriptor. Preserve full-digest, selection, permission and final-symlink checks. Cover both publication orders, distinct files and hardlinks, cold/warm native publication and registry-free current recovery; keep transport repair separate from retention and net-benefit claims. |
 | `todo_read_source_authority_label` | A canonical Todo read reports one provider in `authority_read` but labels the public result or projection source as File. | The same read's canonical provider revision, public source, envelope sources and Markdown across full, filtered, exact and compact views. | The shared formatter retained a provider-specific constant after provider routing moved to the canonical reader. | Reuse the existing canonical source authority for presentation and provenance. Exercise real CLI File/SQLite reads, missing results and source failure/recovery; preserve filtering, read-only behavior and fail-closed routing. A display correction does not select, migrate or qualify a provider. |

tests/control_plane/test_host_process.py

@@ -486,3 +486,45 @@ def test_default_turn_deadline_is_optional_and_real_transport_accepts_none(tmp_p
     )
     assert result["outcome"] == "exited"
     assert result["returncode"] == 0
+
+
+@pytest.mark.parametrize("broken_observation", ["not-json", "[]"])
+def test_host_bridge_parser_failure_closes_reader_without_claiming_cleanup(broken_observation):
+    from loopx.control_plane.turn_driver.host_bridge_transport import GuardedHostBridge
+
+    source = """
+import json, sys
+print(json.dumps({"kind": "ready"}), flush=True)
+sys.stdin.readline()
+for line in (sys.argv[1], sys.argv[1], '{"kind":"stdout","text":"diagnostic"}',
+             '{"kind":"result","outcome":"exited"}'):
+    print(line, flush=True)
+"""
+    with subprocess.Popen([sys.executable, "-u", "-c", source, broken_observation],
+                          stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                          text=True, encoding="utf-8") as process:
+        bridge = GuardedHostBridge(process, deadline=time.monotonic() + 5, cleanup_timeout=.2)
+        try:
+            bridge.send("{}\n")
+            process.wait(timeout=5)
+            with pytest.raises((json.JSONDecodeError, RuntimeError)):
+                bridge.next(deadline=time.monotonic() + 5)
+            with pytest.raises(RuntimeError, match="cleanup unconfirmed"):
+                bridge.cancel()
+            bridge.close_output()
+            assert not bridge.reader.is_alive()
+            assert bridge.output.closed
+            bridge.close_output()
+            # A dead bridge or reclaimed pipe cannot establish the Host's exit.
+            assert bridge.not_launched is False and bridge.result is None
+        finally:
+            # Keep a failing baseline from stranding its producer in the test runner.
+            import queue
+            for _ in range(4):
+                if not bridge.reader.is_alive():
+                    break
+                try:
+                    bridge.pending.get(timeout=.2)
+                except queue.Empty:
+                    pass
+                bridge.reader.join(timeout=.2)