refactor(turn): defer DSH builders and process helpers to selected paths

loopx/cli_commands/turn_run_once.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import argparse
+import sys as _sys
 from collections.abc import Callable, Mapping
 from pathlib import Path
+from typing import TYPE_CHECKING as _TYPE_CHECKING
 from typing import Any
 
 from ..capabilities.explore.composition_frontier import (
@@ -42,15 +44,37 @@
 from .turn_cadence import managed_cadence_start
 from .turn_host_request import build_codex_cli_host_runner, prepare_turn_host_request
 from .turn_decision import collect_turn_status_payload, current_turn_host_launch_decision
-from .turn_dsh_host import build_dsh_host_runner, build_dsh_operation_host_runner
 from .turn_rendering import build_turn_error_payload
 from .turn_post_writeback import drain_committed_turn_post_writeback_hooks
 from .turn_todo_writeback import captured_todo_text_sha256, write_turn_validated_completion
 
+if _TYPE_CHECKING:
+    from .turn_dsh_host import (
+        build_dsh_host_runner as build_dsh_host_runner,
+        build_dsh_operation_host_runner as build_dsh_operation_host_runner,
+    )
+
 EXACT_SETTLEMENT_READBACK_NOT_FOUND = (
     "exact settlement readback unexpectedly returned not-found"
 )
 
+# Keep the existing import and patch seams without loading an unselected host.
+_DSH_HOST_EXPORTS = ("build_dsh_host_runner", "build_dsh_operation_host_runner")
+
+
+def __getattr__(name: str) -> Any:
+    if name not in _DSH_HOST_EXPORTS:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    from . import turn_dsh_host
+
+    value = getattr(turn_dsh_host, name)
+    globals()[name] = value
+    return value
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | set(_DSH_HOST_EXPORTS))
+
 
 def execute_turn_run_once(
     args: argparse.Namespace,
@@ -816,7 +840,7 @@ def resolve_built_in_session_binding(
             session_binding_resolver = resolve_built_in_session_binding
         elif args.host == "dsh":
             if getattr(args, "dsh_operation_module", None):
-                host_runner = build_dsh_operation_host_runner(
+                host_runner = getattr(_sys.modules[__name__], "build_dsh_operation_host_runner")(
                     args, runtime_root=runtime_root, registry_path=registry_path,
                     workspace=project, environ=operator_environ,
                     goal_admission=goal_admission,
@@ -829,7 +853,7 @@ def resolve_dsh_operation_session(envelope: Mapping[str, Any]) -> dict[str, str]
 
                 session_binding_resolver = resolve_dsh_operation_session
             else:
-                host_runner = build_dsh_host_runner(
+                host_runner = getattr(_sys.modules[__name__], "build_dsh_host_runner")(
                     args, workspace=project, environ=operator_environ,
                 )
 
@@ -937,3 +961,6 @@ def on_managed_start_admitted() -> None:
             execution_started=execution_started,
             journal_readback=journal_readback,
         )
+
+
+__all__ = [name for name in globals() if not name.startswith("_")] + list(_DSH_HOST_EXPORTS)

loopx/control_plane/turn_driver/host_bridge_transport.py

@@ -6,11 +6,35 @@
 import queue
 import signal
 import subprocess
+import sys as _sys
 import threading
 import time
 from typing import Any
+from typing import TYPE_CHECKING as _TYPE_CHECKING
 
-from ...extensions.process_runtime import force_terminate_process, write_process_stdin_before_deadline
+if _TYPE_CHECKING:
+    from ...extensions.process_runtime import (
+        force_terminate_process as force_terminate_process,
+        write_process_stdin_before_deadline as write_process_stdin_before_deadline,
+    )
+
+# Preview needs the bridge type; process helpers belong to execution or an
+# explicit caller import. Keep their existing identity and patch seams.
+_PROCESS_EXPORTS = ("force_terminate_process", "write_process_stdin_before_deadline")
+
+
+def __getattr__(name: str) -> Any:
+    if name not in _PROCESS_EXPORTS:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    from ...extensions import process_runtime
+
+    value = getattr(process_runtime, name)
+    globals()[name] = value
+    return value
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | set(_PROCESS_EXPORTS))
 
 
 class GuardedHostBridge:
@@ -61,8 +85,8 @@ def send(self, frame: str) -> None:
         if self.next(deadline=self.deadline) != {"kind": "ready"}:
             raise RuntimeError("managed Host bridge did not become ready")
         self.requested = True
-        write_process_stdin_before_deadline(self.process, frame, deadline=self.deadline,
-            cancel_and_reap=self.cancel)
+        write_stdin = getattr(_sys.modules[__name__], "write_process_stdin_before_deadline")
+        write_stdin(self.process, frame, deadline=self.deadline, cancel_and_reap=self.cancel)
 
     def cancel(self) -> None:
         if self.cancelled:
@@ -73,7 +97,8 @@ def cancel(self) -> None:
             if not self.requested:
                 # No request bytes have been sent, so this bridge cannot own a
                 # Host. This also handles suspension before JS handlers exist.
-                force_terminate_process(self.process, process_tree_owned=True)
+                getattr(_sys.modules[__name__], "force_terminate_process")(
+                    self.process, process_tree_owned=True)
                 self.not_launched = True
             elif self.process.poll() is None:
                 self.process.terminate()
@@ -98,7 +123,8 @@ def cancel(self) -> None:
             # Killing the bridge alone cannot certify an unobserved detached
             # Host. Retain the launching record and fail instead of claiming it.
             if self.process.poll() is None:
-                force_terminate_process(self.process, process_tree_owned=True)
+                getattr(_sys.modules[__name__], "force_terminate_process")(
+                    self.process, process_tree_owned=True)
             raise RuntimeError("managed Host bridge cleanup unconfirmed") from error
 
     def close_output(self) -> None:
@@ -112,3 +138,6 @@ def close_output(self) -> None:
         self.reader.join(timeout=self.cleanup_timeout)
         if not self.reader.is_alive():
             self.output.close()
+
+
+__all__ = [name for name in globals() if not name.startswith("_")] + list(_PROCESS_EXPORTS)

loopx/doctor_git.py

@@ -5,13 +5,31 @@
 import os
 import re
 import subprocess
+import sys as _sys
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
 from typing import Any
+from typing import TYPE_CHECKING as _TYPE_CHECKING
 from urllib.parse import urlsplit
 
-from .extensions.process_runtime import terminate_process_tree
+if _TYPE_CHECKING:
+    from .extensions.process_runtime import terminate_process_tree as terminate_process_tree
+
+
+# Successful read-only Git probes do not need process cleanup. Keep the former
+# helper export and patch point, resolving its owner only when requested.
+def __getattr__(name: str) -> Any:
+    if name != "terminate_process_tree":
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    from .extensions.process_runtime import terminate_process_tree
+
+    globals()[name] = terminate_process_tree
+    return terminate_process_tree
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | {"terminate_process_tree"})
 
 
 # `doctor` inspects a checkout the operator may not control, so every Git read
@@ -64,7 +82,7 @@ def _retire_owned_probe(process: subprocess.Popen[str]) -> None:
     """
 
     try:
-        terminate_process_tree(process, grace_seconds=0.2)
+        getattr(_sys.modules[__name__], "terminate_process_tree")(process, grace_seconds=0.2)
     except (OSError, RuntimeError):
         # The answer is already unknown; an unconfirmed cleanup must not turn a
         # read-only diagnostic into a traceback.
@@ -324,3 +342,6 @@ def trusted_release_ref_for_root(
                 "git_ref": f"{remote}/{expected_ref}",
             }
     return None
+
+
+__all__ = [name for name in globals() if not name.startswith("_")] + ["terminate_process_tree"]

tests/architecture/test_control_plane_import_boundaries.py

@@ -324,6 +324,31 @@ def test_read_adapters_do_not_load_writeback_facades_for_leaf_helpers() -> None:
         assert completed.returncode == 0, completed.stderr
 
 
+def test_turn_run_once_defers_dsh_builders_and_preserves_exports() -> None:
+    completed = subprocess.run(
+        [sys.executable, "-c", """
+import sys
+import loopx.cli_commands.turn_run_once as command
+assert 'loopx.cli_commands.turn_dsh_host' not in sys.modules
+assert 'loopx.dsh_goal_mode.turn_host_adapter' not in sys.modules
+names = {'build_dsh_host_runner', 'build_dsh_operation_host_runner'}
+assert names <= set(dir(command))
+assert not hasattr(command, 'unknown_host_builder')
+from loopx.cli_commands.turn_run_once import build_dsh_host_runner, build_dsh_operation_host_runner
+from loopx.cli_commands import turn_dsh_host
+assert build_dsh_host_runner is turn_dsh_host.build_dsh_host_runner
+assert build_dsh_operation_host_runner is turn_dsh_host.build_dsh_operation_host_runner
+exports = {}
+exec('from loopx.cli_commands.turn_run_once import *', exports)
+assert exports['build_dsh_host_runner'] is build_dsh_host_runner
+assert exports['build_dsh_operation_host_runner'] is build_dsh_operation_host_runner
+assert exports['execute_turn_run_once'] is command.execute_turn_run_once
+"""],
+        cwd=REPOSITORY_ROOT, capture_output=True, check=False, text=True,
+    )
+    assert completed.returncode == 0, completed.stderr
+
+
 def test_settlement_plan_import_does_not_load_native_capture_io() -> None:
     # A command template needs the native output profile, not capture binding
     # or the complete quota/Goal routing graph. Keep the typed kernel usable
@@ -373,6 +398,54 @@ def test_turn_preview_import_does_not_select_report_delivery_or_lark_hosts() ->
     assert completed.returncode == 0, completed.stderr
 
 
+def test_host_bridge_defers_process_helpers_and_preserves_exports() -> None:
+    completed = subprocess.run(
+        [sys.executable, "-c", """
+import sys
+import loopx.control_plane.turn_driver.host_bridge_transport as bridge
+assert 'loopx.extensions.process_runtime' not in sys.modules
+names = {'force_terminate_process', 'write_process_stdin_before_deadline'}
+assert names <= set(dir(bridge))
+assert not hasattr(bridge, 'unknown_process_helper')
+from loopx.control_plane.turn_driver.host_bridge_transport import force_terminate_process, write_process_stdin_before_deadline
+from loopx.extensions import process_runtime
+assert force_terminate_process is process_runtime.force_terminate_process
+assert write_process_stdin_before_deadline is process_runtime.write_process_stdin_before_deadline
+exports = {}
+exec('from loopx.control_plane.turn_driver.host_bridge_transport import *', exports)
+assert exports['force_terminate_process'] is force_terminate_process
+assert exports['write_process_stdin_before_deadline'] is write_process_stdin_before_deadline
+assert exports['GuardedHostBridge'] is bridge.GuardedHostBridge
+"""],
+        cwd=REPOSITORY_ROOT, capture_output=True, check=False, text=True,
+    )
+    assert completed.returncode == 0, completed.stderr
+
+
+def test_doctor_git_defers_cleanup_and_preserves_exports() -> None:
+    completed = subprocess.run(
+        [sys.executable, "-c", """
+import sys
+from pathlib import Path
+from loopx import doctor_git
+assert 'loopx.extensions.process_runtime' not in sys.modules
+assert doctor_git._run_git(Path.cwd(), 'rev-parse', '--show-toplevel').returncode == 0
+assert 'loopx.extensions.process_runtime' not in sys.modules
+assert 'terminate_process_tree' in dir(doctor_git)
+assert not hasattr(doctor_git, 'unknown_process_helper')
+from loopx.doctor_git import terminate_process_tree
+from loopx.extensions import process_runtime
+assert terminate_process_tree is process_runtime.terminate_process_tree
+exports = {}
+exec('from loopx.doctor_git import *', exports)
+assert exports['terminate_process_tree'] is terminate_process_tree
+assert exports['git_source_metadata'] is doctor_git.git_source_metadata
+"""],
+        cwd=REPOSITORY_ROOT, capture_output=True, check=False, text=True,
+    )
+    assert completed.returncode == 0, completed.stderr
+
+
 def test_lazy_report_and_lark_exports_preserve_original_owner_identity() -> None:
     # Compatibility includes from-imports, attribute access, dir(), star
     # exports for report facades, and the existing Lark command patch seam.

tests/control_plane/test_host_process.py

@@ -488,6 +488,60 @@ def test_default_turn_deadline_is_optional_and_real_transport_accepts_none(tmp_p
     assert result["returncode"] == 0
 
 
+@pytest.mark.parametrize("operation", ["send", "cancel"])
+def test_host_bridge_preserves_process_helper_patch_points(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, operation: str,
+) -> None:
+    from loopx.control_plane.turn_driver import host_bridge_transport as bridge_module
+
+    calls: list[str] = []
+    original_write = bridge_module.write_process_stdin_before_deadline
+    original_terminate = bridge_module.force_terminate_process
+
+    def write(*args, **kwargs):
+        calls.append("send")
+        return original_write(*args, **kwargs)
+
+    def terminate(*args, **kwargs):
+        assert kwargs["process_tree_owned"] is True
+        calls.append("cancel")
+        return original_terminate(*args, **kwargs)
+
+    monkeypatch.setattr(bridge_module, "write_process_stdin_before_deadline", write)
+    monkeypatch.setattr(bridge_module, "force_terminate_process", terminate)
+    source = """
+import json, sys
+print(json.dumps({"kind": "ready"}), flush=True)
+assert json.loads(sys.stdin.readline()) == {"message": "bridge λ😀"}
+print(json.dumps({"kind": "result", "outcome": "exited"}), flush=True)
+"""
+    with subprocess.Popen(
+        [sys.executable, "-u", "-c", source], cwd=tmp_path,
+        stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True,
+        encoding="utf-8", start_new_session=True,
+    ) as process:
+        bridge = bridge_module.GuardedHostBridge(
+            process, deadline=time.monotonic() + 5, cleanup_timeout=2,
+        )
+        try:
+            if operation == "send":
+                bridge.send(json.dumps({"message": "bridge λ😀"}, ensure_ascii=False) + "\n")
+                process.wait(timeout=5)
+            bridge.cancel()
+            bridge.cancel()
+            assert calls == [operation]
+            assert process.poll() is not None
+            assert bridge.not_launched is (operation == "cancel")
+            if operation == "send":
+                assert bridge.result == {"kind": "result", "outcome": "exited"}
+        finally:
+            if process.poll() is None:
+                process.kill()
+                process.wait(timeout=5)
+            bridge.close_output()
+        assert not bridge.reader.is_alive() and bridge.output.closed
+
+
 @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

tests/test_doctor_git_probe_budget.py

@@ -102,8 +102,9 @@ def test_doctor_finishes_when_git_blocks(
 
 
 @pytest.mark.skipif(os.name != "posix", reason="core.fsmonitor needs a POSIX hook")
+@pytest.mark.parametrize("cleanup_error", [False, True])
 def test_a_blocked_query_takes_the_helpers_it_started(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, cleanup_error: bool,
 ) -> None:
     """Git's own helpers are in its group, and the budget has to retire them.
 
@@ -127,9 +128,20 @@ def test_a_blocked_query_takes_the_helpers_it_started(
     )
 
     monkeypatch.setattr(doctor_git, "GIT_PROBE_TIMEOUT_SECONDS", 0.8)
+    cleanup_calls: list[float] = []
+    original_terminate = doctor_git.terminate_process_tree
+
+    def terminate(process, *, grace_seconds):
+        cleanup_calls.append(grace_seconds)
+        original_terminate(process, grace_seconds=grace_seconds)
+        if cleanup_error:
+            raise RuntimeError("fixture cleanup could not confirm retirement")
+
+    monkeypatch.setattr(doctor_git, "terminate_process_tree", terminate)
     started = time.monotonic()
     assert doctor_git._run_git(repo, "status", "--porcelain") is None
     assert time.monotonic() - started < 6
+    assert cleanup_calls == [0.2]
 
     retired = heartbeats.stat().st_size
     assert retired > 0, "the fixture hook never ran, so this proves nothing"