From e32d4db3ceca9b4a8b7f3ea7a0dc7291e8edde48 Mon Sep 17 00:00:00 2001 From: "lusendong.6789" Date: Sat, 7 Nov 2026 06:08:48 +0800 Subject: [PATCH] refactor(runtime): defer process transport at core callers Signed-off-by: lusendong.6789 Co-authored-by: TRAE CLI --- .../rfcs/monorepo-distribution-split-v0.md | 42 ++++ .../monorepo-distribution-split-v0.zh-CN.md | 30 +++ loopx/contract.py | 40 +++- .../runtime/validation_command.py | 38 +++- .../turn_driver/host_bridge_transport.py | 44 ++++- loopx/doctor_git.py | 38 +++- tests/test_process_runtime_loading.py | 186 ++++++++++++++++++ 7 files changed, 408 insertions(+), 10 deletions(-) create mode 100644 tests/test_process_runtime_loading.py diff --git a/docs/architecture/rfcs/monorepo-distribution-split-v0.md b/docs/architecture/rfcs/monorepo-distribution-split-v0.md index c8e7ce79f9..cf4e6eed5c 100644 --- a/docs/architecture/rfcs/monorepo-distribution-split-v0.md +++ b/docs/architecture/rfcs/monorepo-distribution-split-v0.md @@ -824,6 +824,48 @@ Approval needs a reviewed decision, not implementation progress or silence. installation, R1 managed minimal or D1–D7. No preset, frontend operation, permission or persisted state changes; protected execution keeps its owner. +### 2026-11-07 — Caller process-transport loading proposal + +- **Baseline and boundary:** `7f521ff4505fa916a98a7401032922189eef414a` + loads `extensions.process_runtime` during the pure Turn-decision import. + Doctor Git, contract scanning, caller validation and Host bridge transport + now resolve their existing reexports on use. The original function objects, + signatures, direct/from/star imports, discovery, caller patch points, + preloaded-owner bindings and reload behavior remain available. This uses the + existing lazy-export pattern; the export-name set is local compatibility + metadata, not a shared state vocabulary. Python retains platform transport; + TypeScript retains decisions, deadlines, authority and receipts. No provider, + capability, default or process algorithm is added. +- **Executable evidence:** the four cold-caller oracles fail on the immutable + baseline. Focused compatibility checks and the existing architecture oracle + pass on the proposal; the Turn-preview unselected-module guard is retained. + Real isolated Git, validation and bridge tests cover timeout, cancellation, + exited-leader descendants, stdin policy, framing and unconfirmed cleanup. + A newly built monolithic wheel repeats compatibility and real-process checks + outside the checkout. Complete public discovery/star name sets and transport + signatures match the baseline. This is source/wheel proposal evidence, not an + upgrade or adoption result. +- **Measured scope:** pure Turn-decision import loads 337 LoopX modules instead + of 338. Pure import and help start no processes or connections; supported help + and unknown-command output/exit status remain identical. Shared synthetic + SQLite fixtures exercise accepted and rejected registered Turn plans: both + revisions start two Node processes and thirteen Git probes, with 62/61 + connection observations respectively. Their complete returned envelopes match + except the source-decision digest from fresh observations. Registered plans + still select process transport; the entire preview dependency closure is not + qualified as cold. One removed import does not establish latency, RSS, + complete-cost, managed-Turn or long-horizon benefit. +- **Remaining and rollback:** retain I3 for at least one minor release. D2, D5, + D6, I3, D3 and full R0/V10/M1/R1/context/model/install acceptance remain open. + Canary passes its six direct checks and 17 of 18 selected checks. + `first-connect-contract-smoke.py` fails with `KeyError: should_run` on both + baseline and proposal; its inline-field assumption needs separate repair. + This failure and existing broader lint/browser failures remain open. + The completed author proposal may enter normal review with disclosed + holds; protected runtime merge still requires maintainer authorization. + Revert this caller loading change to restore eager binding without state + migration; no unrelated language rewrite is required. + ## Appendix B: Decision log | Date | Decision | Owner / approval | Alternatives | Normative sections changed | diff --git a/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md b/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md index 5136da5181..913cb2513c 100644 --- a/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md +++ b/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md @@ -599,6 +599,36 @@ R0 可以先于目录移动交付，但不代表内核合格。R3/R4 可选， 不关闭完整 R0/V10、M1、拆包安装、R1 managed minimal 或 D1–D7。不改变 preset、 前端操作、权限或持久化状态；受保护执行仍由既有 owner 所有。 +### 2026-11-07 — 调用方进程传输按需加载提案 + +- **基线与边界：**`7f521ff4505fa916a98a7401032922189eef414a` 的纯 + Turn-decision import 会加载 `extensions.process_runtime`。Doctor Git、 + contract 扫描、调用方 validation 和 Host bridge transport 改为使用时解析既有 + reexport。保留原函数对象、签名、direct/from/star import、名称发现、调用方补丁点、 + 预加载 owner 的绑定和 reload 行为。复用现有 lazy-export 模式；导出名集合是本地 + 兼容元数据，不是共享状态词汇。Python 保留平台传输，TypeScript 保留决策、deadline、 + authority 与 receipt；不新增 provider、capability、default 或进程算法。 +- **可执行证据：**四个冷加载调用方 oracle 在不可变基线上失败。提案通过聚焦兼容 + 检查和既有架构 oracle，保留 Turn-preview 的未选模块隔离门。真实隔离的 Git、 + validation 和 bridge 测试覆盖 timeout、cancellation、leader 退出后的后代、stdin + 策略、framing 和未确认 cleanup。重新构建的单体 wheel 在源码目录外重复兼容与真实 + 进程检查；完整公开 discovery/star 名称集合及 transport 签名与基线一致。这是 + source/wheel 提案证据，不代表现有 host 升级或实际采用。 +- **实测范围：**纯 Turn-decision import 的 LoopX 模块数从 338 降至 337。 + 纯 import 和 help 没有进程启动或连接，支持的 help、unknown-command 输出与退出码 + 保持一致。共享合成 SQLite fixture 覆盖接受与拒绝的已注册 Turn plan：两个版本都 + 启动两个 Node 进程和十三个 Git probe，连接观察分别为 62/61 次。完整返回 envelope + 仅 fresh observation 的 source-decision digest 不同。已注册 plan 仍会选择进程 + transport，完整 preview 依赖闭包尚未通过冷加载资格。减少一个 import 不能证明 + latency、RSS、完整成本、managed-Turn 或长程收益。 +- **剩余与回滚：**I3 至少保留一个 minor release。D2、D5、D6、I3、D3，以及完整 + R0/V10/M1/R1/context/model/install 验收仍开放。Canary 六个 direct check 通过， + 18 个 selected check 中 17 个通过；`first-connect-contract-smoke.py` 在基线与 + 提案都报 `KeyError: should_run`，其 inline-field 假设需要独立修复。该失败与既有 + 全库 lint/browser 失败仍开放。完成的作者提案可披露 hold 后进入正常 review；受保护 runtime + 合入仍需 maintainer 授权。回退本调用方加载改动即可恢复 eager binding，无需状态 + 迁移或无关语言重写。 + ## 附录 B：决策日志 | 日期 | 决策 | 负责人 / 批准 | 备选 | 变更的规范章节 | diff --git a/loopx/contract.py b/loopx/contract.py index b310687f6e..070af451f9 100644 --- a/loopx/contract.py +++ b/loopx/contract.py @@ -4,6 +4,7 @@ import os import re import subprocess +import sys as _sys from collections.abc import Iterator, Mapping from dataclasses import dataclass from pathlib import Path @@ -36,7 +37,6 @@ from .control_plane.todos.active_state_editing import COMPLETED_WORK_ARCHIVE_HEADING from .control_plane.todos.active_state_todo_parser import active_state_incomplete_warning from .control_plane.todos.authoring_scope import todo_contract_diagnostics -from .extensions.process_runtime import terminate_process_tree from .history import ( RunHistoryAudit, build_run_history_audit, @@ -67,6 +67,36 @@ ) +# Keep the historical exports and caller patch points, but select the OS +# transport only when used. Decisions and deadlines remain with the caller. +_PROCESS_RUNTIME_EXPORTS = ("terminate_process_tree",) + + +def __getattr__(name: str) -> Any: + if name not in _PROCESS_RUNTIME_EXPORTS: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + from .extensions import process_runtime + + for export in _PROCESS_RUNTIME_EXPORTS: + globals().setdefault(export, getattr(process_runtime, export)) + return globals()[name] + + +def __dir__() -> list[str]: + return sorted(set(globals()) | set(_PROCESS_RUNTIME_EXPORTS)) + + +def _export(name: str) -> Any: + return getattr(_sys.modules[__name__], name) + + +# An already selected owner retains the former import-time binding, including +# owner replacements installed before this module was imported. +if "loopx.extensions.process_runtime" in _sys.modules: + for _name in _PROCESS_RUNTIME_EXPORTS: + globals()[_name] = getattr(_sys.modules["loopx.extensions.process_runtime"], _name) + + @dataclass(frozen=True) class LeakRule: pattern: re.Pattern[str] @@ -350,7 +380,7 @@ def _retire_owned_probe(process: subprocess.Popen[str]) -> None: """ try: - terminate_process_tree(process, grace_seconds=0.2) + _export("terminate_process_tree")(process, grace_seconds=0.2) except (OSError, RuntimeError): pass finally: @@ -1694,3 +1724,9 @@ def render_contract_markdown(payload: dict[str, Any]) -> str: lines.extend(["", f"## {title}"]) lines.extend(f"- {item}" for item in items) return "\n".join(lines) + + +# Star imports retain every former public name, including the lazy reexports. +__all__ = sorted( + {name for name in globals() if not name.startswith("_")} | set(_PROCESS_RUNTIME_EXPORTS) +) diff --git a/loopx/control_plane/runtime/validation_command.py b/loopx/control_plane/runtime/validation_command.py index 1e896e9b5b..b104d7bfcc 100644 --- a/loopx/control_plane/runtime/validation_command.py +++ b/loopx/control_plane/runtime/validation_command.py @@ -3,13 +3,41 @@ import os import shlex import subprocess +import sys as _sys from collections.abc import Iterator, Mapping from contextlib import contextmanager from contextvars import ContextVar from pathlib import Path from typing import Any -from ...extensions.process_runtime import terminate_process_tree +# Keep the historical exports and caller patch points, but select the OS +# transport only when used. Decisions and deadlines remain with the caller. +_PROCESS_RUNTIME_EXPORTS = ("terminate_process_tree",) + + +def __getattr__(name: str) -> Any: + if name not in _PROCESS_RUNTIME_EXPORTS: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + from ...extensions import process_runtime + + for export in _PROCESS_RUNTIME_EXPORTS: + globals().setdefault(export, getattr(process_runtime, export)) + return globals()[name] + + +def __dir__() -> list[str]: + return sorted(set(globals()) | set(_PROCESS_RUNTIME_EXPORTS)) + + +def _export(name: str) -> Any: + return getattr(_sys.modules[__name__], name) + + +# An already selected owner retains the former import-time binding, including +# owner replacements installed before this module was imported. +if "loopx.extensions.process_runtime" in _sys.modules: + for _name in _PROCESS_RUNTIME_EXPORTS: + globals()[_name] = getattr(_sys.modules["loopx.extensions.process_runtime"], _name) # Frozen wire id shared with capabilities/issue_fix. Kept verbatim for wire # compatibility with existing caller-repo-branch receipts. @@ -80,7 +108,7 @@ def run_caller_validation( # subprocess.run's immediate-kill cancellation rather than adding # time to the declared validation budget. It also handles an exited # POSIX leader whose surviving children still hold the output pipes. - terminate_process_tree(process, grace_seconds=0) + _export("terminate_process_tree")(process, grace_seconds=0) raise return { "schema_version": CALLER_VALIDATION_RECEIPT_SCHEMA_VERSION, @@ -99,3 +127,9 @@ def require_validation_passed(step: Mapping[str, Any]) -> None: raise RuntimeError( f"{step.get('command_label')} failed with exit code {step.get('exit_code')}" ) + + +# Star imports retain every former public name, including the lazy reexports. +__all__ = sorted( + {name for name in globals() if not name.startswith("_")} | set(_PROCESS_RUNTIME_EXPORTS) +) diff --git a/loopx/control_plane/turn_driver/host_bridge_transport.py b/loopx/control_plane/turn_driver/host_bridge_transport.py index 1dbed9c89a..d46f14c95a 100644 --- a/loopx/control_plane/turn_driver/host_bridge_transport.py +++ b/loopx/control_plane/turn_driver/host_bridge_transport.py @@ -6,11 +6,41 @@ import queue import signal import subprocess +import sys as _sys import threading import time from typing import Any -from ...extensions.process_runtime import force_terminate_process, write_process_stdin_before_deadline +# Keep the historical exports and caller patch points, but select the OS +# transport only when used. Decisions and deadlines remain with the caller. +_PROCESS_RUNTIME_EXPORTS = ( + "force_terminate_process", "write_process_stdin_before_deadline", +) + + +def __getattr__(name: str) -> Any: + if name not in _PROCESS_RUNTIME_EXPORTS: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + from ...extensions import process_runtime + + for export in _PROCESS_RUNTIME_EXPORTS: + globals().setdefault(export, getattr(process_runtime, export)) + return globals()[name] + + +def __dir__() -> list[str]: + return sorted(set(globals()) | set(_PROCESS_RUNTIME_EXPORTS)) + + +def _export(name: str) -> Any: + return getattr(_sys.modules[__name__], name) + + +# An already selected owner retains the former import-time binding, including +# owner replacements installed before this module was imported. +if "loopx.extensions.process_runtime" in _sys.modules: + for _name in _PROCESS_RUNTIME_EXPORTS: + globals()[_name] = getattr(_sys.modules["loopx.extensions.process_runtime"], _name) class GuardedHostBridge: @@ -61,7 +91,7 @@ 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, + _export("write_process_stdin_before_deadline")(self.process, frame, deadline=self.deadline, cancel_and_reap=self.cancel) def cancel(self) -> None: @@ -73,7 +103,7 @@ 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) + _export("force_terminate_process")(self.process, process_tree_owned=True) self.not_launched = True elif self.process.poll() is None: self.process.terminate() @@ -98,7 +128,7 @@ 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) + _export("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 +142,9 @@ def close_output(self) -> None: self.reader.join(timeout=self.cleanup_timeout) if not self.reader.is_alive(): self.output.close() + + +# Star imports retain every former public name, including the lazy reexports. +__all__ = sorted( + {name for name in globals() if not name.startswith("_")} | set(_PROCESS_RUNTIME_EXPORTS) +) diff --git a/loopx/doctor_git.py b/loopx/doctor_git.py index df09510587..d5f201a53f 100644 --- a/loopx/doctor_git.py +++ b/loopx/doctor_git.py @@ -5,13 +5,41 @@ 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 urllib.parse import urlsplit -from .extensions.process_runtime import terminate_process_tree +# Keep the historical exports and caller patch points, but select the OS +# transport only when used. Decisions and deadlines remain with the caller. +_PROCESS_RUNTIME_EXPORTS = ("terminate_process_tree",) + + +def __getattr__(name: str) -> Any: + if name not in _PROCESS_RUNTIME_EXPORTS: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + from .extensions import process_runtime + + for export in _PROCESS_RUNTIME_EXPORTS: + globals().setdefault(export, getattr(process_runtime, export)) + return globals()[name] + + +def __dir__() -> list[str]: + return sorted(set(globals()) | set(_PROCESS_RUNTIME_EXPORTS)) + + +def _export(name: str) -> Any: + return getattr(_sys.modules[__name__], name) + + +# An already selected owner retains the former import-time binding, including +# owner replacements installed before this module was imported. +if "loopx.extensions.process_runtime" in _sys.modules: + for _name in _PROCESS_RUNTIME_EXPORTS: + globals()[_name] = getattr(_sys.modules["loopx.extensions.process_runtime"], _name) # `doctor` inspects a checkout the operator may not control, so every Git read @@ -64,7 +92,7 @@ def _retire_owned_probe(process: subprocess.Popen[str]) -> None: """ try: - terminate_process_tree(process, grace_seconds=0.2) + _export("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 +352,9 @@ def trusted_release_ref_for_root( "git_ref": f"{remote}/{expected_ref}", } return None + + +# Star imports retain every former public name, including the lazy reexports. +__all__ = sorted( + {name for name in globals() if not name.startswith("_")} | set(_PROCESS_RUNTIME_EXPORTS) +) diff --git a/tests/test_process_runtime_loading.py b/tests/test_process_runtime_loading.py new file mode 100644 index 0000000000..ca185761d2 --- /dev/null +++ b/tests/test_process_runtime_loading.py @@ -0,0 +1,186 @@ +"""Existing transport exports stay compatible while pure callers stay cold.""" +from __future__ import annotations + +import os +import subprocess +import sys +import textwrap + +import pytest + +# Existing caller reexports, not a new transport API or state vocabulary. +CALLERS = { + "loopx.doctor_git": ("terminate_process_tree",), + "loopx.contract": ("terminate_process_tree",), + "loopx.control_plane.runtime.validation_command": ("terminate_process_tree",), + "loopx.control_plane.turn_driver.host_bridge_transport": ( + "force_terminate_process", "write_process_stdin_before_deadline", + ), +} + + +def _run(code: str) -> None: + completed = subprocess.run( + [sys.executable, "-c", textwrap.dedent(code)], + env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1", "LOOPX_USAGE_PING": "0"}, + capture_output=True, text=True, timeout=30, + ) + assert completed.returncode == 0, completed.stderr + + +@pytest.mark.parametrize("module_name", CALLERS) +def test_pure_caller_import_leaves_process_transport_unselected(module_name: str) -> None: + _run(rf""" + import importlib, sys + effects = [] + sys.addaudithook(lambda event, args: effects.append(event) + if event in ('subprocess.Popen', 'socket.connect') else None) + module = importlib.import_module({module_name!r}) + assert 'loopx.extensions.process_runtime' not in sys.modules + assert set({CALLERS[module_name]!r}) <= set(dir(module)) + assert effects == [] + try: + getattr(module, 'missing_transport_export') + except AttributeError: + pass + else: + raise AssertionError('unknown attribute accepted') + assert 'loopx.extensions.process_runtime' not in sys.modules + """) + + +@pytest.mark.parametrize("module_name", CALLERS) +@pytest.mark.parametrize("mode", ["attribute", "from", "star"]) +def test_explicit_transport_export_is_original_function(module_name: str, mode: str) -> None: + _run(rf""" + import importlib, inspect + module = importlib.import_module({module_name!r}) + names = {CALLERS[module_name]!r} + namespace = {{}} + if {mode!r} == 'from': + exec('from {module_name} import ' + ', '.join(names), namespace) + elif {mode!r} == 'star': + exec('from {module_name} import *', namespace) + else: + namespace.update((name, getattr(module, name)) for name in names) + owner = importlib.import_module('loopx.extensions.process_runtime') + for name in names: + value = namespace[name] + assert value is getattr(owner, name) + assert value is getattr(module, name) + assert inspect.signature(value) == inspect.signature(getattr(owner, name)) + if {mode!r} == 'star': + assert {{name for name in namespace if not name.startswith('_')}} == {{ + name for name in dir(module) if not name.startswith('_')}} + """) + + +@pytest.mark.parametrize("module_name", CALLERS) +def test_preloaded_owner_preserves_import_time_binding(module_name: str) -> None: + _run(rf""" + import importlib + owner = importlib.import_module('loopx.extensions.process_runtime') + names = {CALLERS[module_name]!r} + def before(*args, **kwargs): pass + def after(*args, **kwargs): pass + for name in names: + setattr(owner, name, before) + module = importlib.import_module({module_name!r}) + for name in names: + setattr(owner, name, after) + assert getattr(module, name) is before + """) + + +@pytest.mark.parametrize("module_name", CALLERS) +def test_caller_patch_points_remain_internal_bindings(module_name: str) -> None: + _run(rf""" + import importlib, io, time + from pathlib import Path + module = importlib.import_module({module_name!r}) + names = {CALLERS[module_name]!r} + calls = [] + def patched(*args, **kwargs): + calls.append((args, kwargs)) + for name in names: + setattr(module, name, patched) + class Process: + stdout = io.StringIO() + stderr = io.StringIO() + returncode = 0 + def __enter__(self): return self + def __exit__(self, *args): pass + def wait(self, **kwargs): return 0 + def poll(self): return 0 + def communicate(self, **kwargs): raise KeyboardInterrupt() + process = Process() + if {module_name!r} in ('loopx.doctor_git', 'loopx.contract'): + module._retire_owned_probe(process) + assert calls == [((process,), {{'grace_seconds': 0.2}})] + elif {module_name!r}.endswith('validation_command'): + module.subprocess.Popen = lambda *args, **kwargs: process + try: + module.run_caller_validation(Path.cwd(), validation_argv=['fixture'], + validation_label='fixture', timeout_seconds=1) + except KeyboardInterrupt: + pass + else: + raise AssertionError('cancellation lost') + assert calls == [((process,), {{'grace_seconds': 0}})] + else: + process.stdout = io.StringIO('{{"kind":"ready"}}\n') + bridge = module.GuardedHostBridge(process, + deadline=time.monotonic()+2, cleanup_timeout=1) + try: + bridge.send('fixture\n') + assert calls[0][0] == (process, 'fixture\n') + assert calls[0][1]['cancel_and_reap'] == bridge.cancel + finally: + bridge.close_output() + calls.clear() + process.stdout = io.StringIO() + bridge = module.GuardedHostBridge(process, + deadline=time.monotonic()+2, cleanup_timeout=1) + try: + bridge.cancel() + assert calls == [((process,), {{'process_tree_owned': True}})] + assert bridge.not_launched + finally: + bridge.close_output() + """) + + +@pytest.mark.parametrize("module_name", CALLERS) +def test_first_export_binds_the_original_owner_bundle(module_name: str) -> None: + _run(f""" + import importlib + module = importlib.import_module({module_name!r}) + names = {CALLERS[module_name]!r} + first = getattr(module, names[0]) + owner = importlib.import_module('loopx.extensions.process_runtime') + originals = {{name: getattr(owner, name) for name in names}} + def replacement(*args, **kwargs): pass + for name in names: + setattr(owner, name, replacement) + assert getattr(module, name) is originals[name] + assert first is originals[names[0]] + """) + + +@pytest.mark.parametrize("module_name", CALLERS) +def test_reload_rebinds_exports_from_the_selected_owner(module_name: str) -> None: + _run(f""" + import importlib + module = importlib.import_module({module_name!r}) + names = {CALLERS[module_name]!r} + getattr(module, names[0]) + owner = importlib.import_module('loopx.extensions.process_runtime') + def owner_patch(*args, **kwargs): pass + def caller_patch(*args, **kwargs): pass + for name in names: + setattr(owner, name, owner_patch) + setattr(module, name, caller_patch) + importlib.reload(module) + for name in names: + assert getattr(module, name) is owner_patch + """)