From 8285dee352929ef3d296e40fefc1bef7b23f07e6 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 3 Nov 2026 08:12:16 +0800 Subject: [PATCH] fix(host): release blocked bridge readers after cleanup failures Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../turn_driver/host_bridge_transport.py | 21 ++++++++-- .../references/repair-patterns.md | 1 + tests/control_plane/test_host_process.py | 42 +++++++++++++++++++ 3 files changed, 61 insertions(+), 3 deletions(-) diff --git a/loopx/control_plane/turn_driver/host_bridge_transport.py b/loopx/control_plane/turn_driver/host_bridge_transport.py index 61cb305bbc..1dbed9c89a 100644 --- a/loopx/control_plane/turn_driver/host_bridge_transport.py +++ b/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() diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md index 1c1b7174ca..41ac3f59e0 100644 --- a/skills/loopx-self-repair/references/repair-patterns.md +++ b/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. | | `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. | | `vision_history_null_stride_decision` | Reusing the current Agent's complete Vision from `history` in `refresh-state --agent-vision-json` fails with `stride_decision must be an object`. | Compare the authored packet, compact history and status readback; retain the original operation identity and verify that rejection appended nothing. | The compact index represents an absent optional decision as JSON null while the write codec accepts only an omitted field. | Normalize null and omission as absence in the existing typed stride owner, shared by read and write codecs. Keep malformed proposals rejected and enabled stride reviews dependent on a current valid decision. Prove real File/SQLite history-to-writeback recovery, retained cumulative intent, unchanged-state refusals and off/on policy boundaries; do not rewrite historical receipts or count absence as a decision. | diff --git a/tests/control_plane/test_host_process.py b/tests/control_plane/test_host_process.py index 6e44644e77..5932d7e1d2 100644 --- a/tests/control_plane/test_host_process.py +++ b/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)