Skip to content

fix(host): contain Windows Host descendants in an owned Job #7602

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
xyx2002OvO wants to merge 2 commits into
base: main
Choose a base branch
from
+390 −109
Open

fix(host): contain Windows Host descendants in an owned Job #7602

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e44f90e
fix(host): contain Windows Host descendants in an owned Job
xyx2002OvO Oct 28, 2026
b1d3dd8
Merge current main into Windows Host prerequisite
xyx2002OvO Oct 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
249 changes: 140 additions & 109 deletions loopx/control_plane/turn_driver/host_process_transport.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,15 @@ def _host_process_drain(record: dict[str, Any] | None, *, supervises: str,
return (HOST_PROCESS_NOT_LAUNCHED if record.get("process_group") is None
and "bridge_pid" not in record else HOST_PROCESS_UNATTRIBUTABLE)
if not hasattr(os, "killpg"):
if sys.platform == "win32" and "windows_job" in record:
from .windows_host_job import active_job_processes, valid_job_name
if not valid_job_name(record["windows_job"]):
return HOST_PROCESS_UNATTRIBUTABLE
try:
active = active_job_processes(record["windows_job"])
except OSError:
return HOST_PROCESS_UNATTRIBUTABLE
return HOST_PROCESS_DRAINING if active else HOST_PROCESS_DRAINED
# A launched Host on a platform without process groups is never proven
# drained. This is a platform boundary, not an attribution failure.
return HOST_PROCESS_UNSUPPORTED_PLATFORM
Expand Down Expand Up @@ -177,7 +186,11 @@ def execution_host_drain(record_path: Path, *, launch_possible: bool = False) ->
host = _host_process_drain(owned_record, supervises=HOST_PROCESS_SUPERVISES_HOST,
expected=True)
drain = next(state for state in _DRAIN_PRECEDENCE if state in {supervisor, host})
if not host_process_drain_supported() and (launch_possible or drain != HOST_PROCESS_NOT_LAUNCHED):
assigned_windows_job = sys.platform == "win32" and any(
"windows_job" in item for item in (owned_record or {}, supervisor_record or {})
)
if (not assigned_windows_job and not host_process_drain_supported()
and (launch_possible or drain != HOST_PROCESS_NOT_LAUNCHED)):
return HOST_PROCESS_UNSUPPORTED_PLATFORM
return drain

Expand Down Expand Up @@ -283,6 +296,131 @@ def run_host_process(
if delegated_lease is not None:
request["delegated_lease"] = delegated_lease
bridge = Path(__file__).with_name("host_process_bridge.ts")
bridge_environment, record_value, record_path, supervision, supervises = _bridge_record_scope(environment, delegated_lease)
launch_budget = max(1.0, timeout_seconds) if timeout_seconds is not None else 30.0
launch_deadline = time.monotonic() + launch_budget if launch_guard is not None else None
launch_arguments = (["--guarded-launch", str((time.time() + launch_budget) * 1000)]
if launch_guard is not None else [])
with ExitStack() as containment:
windows_job = None
if sys.platform == "win32":
from .windows_host_job import WindowsHostJob
windows_job = containment.enter_context(WindowsHostJob())
proc = subprocess.Popen(
[
_node_executable(),
"--no-warnings",
"--experimental-strip-types",
str(bridge),
*launch_arguments,
],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
encoding="utf-8",
errors="strict",
start_new_session=True,
env=bridge_environment,
)
assert proc.stdin is not None and proc.stdout is not None
result: dict[str, Any] | None = None
record = None
launch_scope = ExitStack()
guarded = (GuardedHostBridge(proc, deadline=launch_deadline,
cleanup_timeout=8 if delegated_lease is not None else 5)
if launch_deadline is not None else None)
try:
if windows_job is not None:
windows_job.assign(proc)
if launch_guard is not None:
launch_scope.enter_context(launch_guard())
record = _begin_host_process_record(record_path, record_value=record_value,
supervises=supervises, supervision=supervision, bridge_pid=proc.pid)
if record is not None and windows_job is not None:
record["windows_job"] = windows_job.name
_write_host_process_record(record_path, record)
frame = json.dumps(request, ensure_ascii=False, separators=(",", ":")) + "\n"
if guarded is not None:
guarded.send(frame)
else:
proc.stdin.write(frame)
proc.stdin.flush()
lines = iter(proc.stdout)
while True:
event = (guarded.next(deadline=launch_deadline) if guarded is not None
else json.loads(line) if (line := next(lines, None)) is not None else None)
if event is None:
break
kind = event.get("kind")
if kind == "spawned" and isinstance(event.get("pid"), int):
if record is not None:
assert record_path is not None
group = event.get("process_group")
record.update(phase="spawned", host_pid=event["pid"],
process_group=group if isinstance(group, int) else None)
_write_host_process_record(record_path, record)
launch_deadline = None
launch_scope.close()
elif kind in {"stdout", "stderr"} and isinstance(event.get("text"), str):
consume = on_stdout if kind == "stdout" else on_stderr
if consume is not None:
consume(event["text"])
elif kind == "result" and event.get("outcome") in {
"exited",
"timeout",
"cancelled",
"output_limit",
"spawn_failed",
}:
result = event
launch_deadline = None
launch_scope.close()
elif kind == "not_launched" and guarded is not None:
guarded.not_launched = True
raise TimeoutError("managed Host bridge launch deadline expired")
else:
raise RuntimeError("Invalid managed Host process observation")
finally:
try:
if guarded is not None and (launch_deadline is not None or sys.exc_info()[0] is not None):
# A failed handoff still owns the Goal fence. The TS bridge must
# report no launch or finish its Host cleanup before it releases.
guarded.cancel()
if record is not None and guarded.not_launched:
assert record_path is not None
record.update(phase=HOST_PROCESS_NOT_LAUNCHED, process_group=None)
record.pop("bridge_pid", None)
_write_host_process_record(record_path, record)
else:
# EOF asks the TS supervisor to clean its Host, also on a
# callback failure. Leased cleanup retains its original budget.
proc.stdin.close()
try:
proc.wait(timeout=8 if delegated_lease is not None else 5)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait()
finally:
if windows_job is not None:

@loopx-agent loopx-agent Oct 28, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Job finish 抛错时仍需执行其余 finally 清理。launch_guard 已进入、随后 record/setup 失败时，finish 内的 Query/Terminate 或观察超时可以再次抛错，使下方 launch_scope.close() 和 output cleanup 都不执行。真实 public run_host_process/Node 桥故障探针（仅 Windows Job IO 为替身）在基线释放 guard/关闭 stdout，在此 head 两者均未清理，内存中加一层 finally 后恢复。请用 finally 保证这两类资源释放，继续保留 Job 清理错误/unknown，并补 setup+finish 双失败回归；这项探针不宣称 Windows 内核资格。

windows_job.finish()
launch_scope.close()
if guarded is not None:
guarded.close_output()
else:
proc.stdout.close()
if record is not None and result is not None and proc.returncode == 0:
assert record_path is not None
# The supervisor returned only after cleaning its group; the group is still re-read.
record["phase"] = "finished"
_write_host_process_record(record_path, record)
if proc.returncode != 0 or result is None:
raise RuntimeError("Managed Host process supervision returned no result")
return result


def _bridge_record_scope(environment, delegated_lease):
"""Separate the supervisor's private record namespace from its child Host."""
# The bridge runs on the caller's environment, which may pin the release the
# Host must run and name the record its supervisor owns. Copy it instead of
# replacing it with the ambient one, and consume the record env from the
Expand All @@ -302,111 +440,4 @@ def run_host_process(
record_path = host_process_supervisor_record(record_path)
supervises = HOST_PROCESS_SUPERVISES_NESTED_HOST
supervision = "leased"
launch_budget = max(1.0, timeout_seconds) if timeout_seconds is not None else 30.0
launch_deadline = time.monotonic() + launch_budget if launch_guard is not None else None
launch_arguments = (["--guarded-launch", str((time.time() + launch_budget) * 1000)]
if launch_guard is not None else [])
proc = subprocess.Popen(
[
_node_executable(),
"--no-warnings",
"--experimental-strip-types",
str(bridge),
*launch_arguments,
],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
encoding="utf-8",
errors="strict",
start_new_session=True,
env=bridge_environment,
)
assert proc.stdin is not None and proc.stdout is not None
result: dict[str, Any] | None = None
record = None
launch_scope = ExitStack()
guarded = (GuardedHostBridge(proc, deadline=launch_deadline,
cleanup_timeout=8 if delegated_lease is not None else 5)
if launch_deadline is not None else None)
try:
if launch_guard is not None:
launch_scope.enter_context(launch_guard())
record = _begin_host_process_record(record_path, record_value=record_value,
supervises=supervises, supervision=supervision, bridge_pid=proc.pid)
frame = json.dumps(request, ensure_ascii=False, separators=(",", ":")) + "\n"
if guarded is not None:
guarded.send(frame)
else:
proc.stdin.write(frame)
proc.stdin.flush()
lines = iter(proc.stdout)
while True:
event = (guarded.next(deadline=launch_deadline) if guarded is not None
else json.loads(line) if (line := next(lines, None)) is not None else None)
if event is None:
break
kind = event.get("kind")
if kind == "spawned" and isinstance(event.get("pid"), int):
if record is not None:
assert record_path is not None
group = event.get("process_group")
record.update(phase="spawned", host_pid=event["pid"],
process_group=group if isinstance(group, int) else None)
_write_host_process_record(record_path, record)
launch_deadline = None
launch_scope.close()
elif kind in {"stdout", "stderr"} and isinstance(event.get("text"), str):
consume = on_stdout if kind == "stdout" else on_stderr
if consume is not None:
consume(event["text"])
elif kind == "result" and event.get("outcome") in {
"exited",
"timeout",
"cancelled",
"output_limit",
"spawn_failed",
}:
result = event
launch_deadline = None
launch_scope.close()
elif kind == "not_launched" and guarded is not None:
guarded.not_launched = True
raise TimeoutError("managed Host bridge launch deadline expired")
else:
raise RuntimeError("Invalid managed Host process observation")
finally:
try:
if guarded is not None and (launch_deadline is not None or sys.exc_info()[0] is not None):
# A failed handoff still owns the Goal fence. The TS bridge must
# report no launch or finish its Host cleanup before it releases.
guarded.cancel()
if record is not None and guarded.not_launched:
assert record_path is not None
record.update(phase=HOST_PROCESS_NOT_LAUNCHED, process_group=None)
record.pop("bridge_pid", None)
_write_host_process_record(record_path, record)
else:
# EOF asks the TS supervisor to clean its Host, also on a
# callback failure. Leased cleanup retains its original budget.
proc.stdin.close()
try:
proc.wait(timeout=8 if delegated_lease is not None else 5)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait()
finally:
launch_scope.close()
if guarded is not None:
guarded.close_output()
else:
proc.stdout.close()
if record is not None and result is not None and proc.returncode == 0:
assert record_path is not None
# The supervisor returned only after cleaning its group; the group is still re-read.
record["phase"] = "finished"
_write_host_process_record(record_path, record)
if proc.returncode != 0 or result is None:
raise RuntimeError("Managed Host process supervision returned no result")
return result
return bridge_environment, record_value, record_path, supervision, supervises
Loading
Loading