diff --git a/tests/control_plane/host_process_fixture.py b/tests/control_plane/host_process_fixture.py new file mode 100644 index 0000000000..58d0f12809 --- /dev/null +++ b/tests/control_plane/host_process_fixture.py @@ -0,0 +1,28 @@ +from __future__ import annotations + + +COUNTER_PROCESS_SOURCE = """ +import os +import signal +import sys +import time +from pathlib import Path + +marker = Path(sys.argv[1]) +pid_path = Path(sys.argv[2]) +interval_seconds = float(sys.argv[3]) +pause_path = Path(sys.argv[4]) if len(sys.argv) == 5 else None + +signal.signal(signal.SIGTERM, signal.SIG_IGN) +pid_path.write_text(str(os.getpid()), encoding="utf-8") +staged = marker.with_name(f".{marker.name}.{os.getpid()}.tmp") +counter = 0 +while True: + staged.write_text(str(counter), encoding="utf-8") + if pause_path is not None: + pause_path.touch() + signal.pause() + os.replace(staged, marker) + counter += 1 + time.sleep(interval_seconds) +""" diff --git a/tests/control_plane/test_host_process.py b/tests/control_plane/test_host_process.py index 7da98714ef..27cb3b612f 100644 --- a/tests/control_plane/test_host_process.py +++ b/tests/control_plane/test_host_process.py @@ -16,6 +16,7 @@ HostOutputLines, run_host_process, ) +from tests.control_plane.host_process_fixture import COUNTER_PROCESS_SOURCE def test_host_output_lines_bound_storage_and_use_lf() -> None: @@ -76,23 +77,53 @@ def reject(_text: str) -> None: assert time.monotonic() - started < 8 +@pytest.mark.skipif(os.name == "nt", reason="POSIX process interruption contract") +def test_counter_process_fixture_publishes_atomically(tmp_path: Path) -> None: + marker = tmp_path / "counter" + pid_path = tmp_path / "pid" + pause_path = tmp_path / "before-publish" + marker.write_text("published", encoding="utf-8") + process = subprocess.Popen( + [ + sys.executable, + "-c", + COUNTER_PROCESS_SOURCE, + str(marker), + str(pid_path), + "0.02", + str(pause_path), + ] + ) + try: + deadline = time.monotonic() + 5 + while not pause_path.exists() and time.monotonic() < deadline: + time.sleep(0.02) + assert pause_path.exists(), "Counter process never reached the publication fence" + process.kill() + process.wait(timeout=5) + assert marker.read_text(encoding="utf-8") == "published" + finally: + if process.poll() is None: + process.kill() + process.wait(timeout=5) + + @pytest.mark.skipif(os.name == "nt", reason="POSIX process-group cancellation contract") def test_disappearing_python_owner_cancels_real_host(tmp_path: Path) -> None: marker = tmp_path / "counter" pid_path = tmp_path / "pid" - host = f""" -import os,time,signal -from pathlib import Path -signal.signal(signal.SIGTERM, signal.SIG_IGN) -Path({str(pid_path)!r}).write_text(str(os.getpid())) -i=0 -while True: - Path({str(marker)!r}).write_text(str(i));i+=1;time.sleep(.02) -""" + host_argv = [ + sys.executable, + "-c", + COUNTER_PROCESS_SOURCE, + str(marker), + str(pid_path), + "0.02", + ] launcher = f""" from pathlib import Path from loopx.control_plane.turn_driver.host_process_transport import run_host_process -run_host_process({[sys.executable, "-c", host]!r}, project=Path({str(tmp_path)!r}), input_text='', timeout_seconds=30) +run_host_process({host_argv!r}, project=Path({str(tmp_path)!r}), input_text='', timeout_seconds=30) """ owner = subprocess.Popen( [sys.executable, "-c", launcher], @@ -125,16 +156,15 @@ def test_disappearing_python_owner_cancels_real_host(tmp_path: Path) -> None: def test_generic_host_timeout_stops_real_descendant(tmp_path: Path) -> None: marker = tmp_path / "child-work" pid_path = tmp_path / "child-pid" - child = f""" -import os,time,signal -from pathlib import Path -signal.signal(signal.SIGTERM, signal.SIG_IGN) -Path({str(pid_path)!r}).write_text(str(os.getpid())) -i=0 -while True: - Path({str(marker)!r}).write_text(str(i));i+=1;time.sleep(.02) -""" - host = f"import subprocess,sys,time;subprocess.Popen({[sys.executable, '-c', child]!r});time.sleep(30)" + child_argv = [ + sys.executable, + "-c", + COUNTER_PROCESS_SOURCE, + str(marker), + str(pid_path), + "0.02", + ] + host = f"import subprocess,time;subprocess.Popen({child_argv!r});time.sleep(30)" try: result = _run_host( {}, argv=[sys.executable, "-c", host], project=tmp_path, timeout_seconds=1 diff --git a/tests/test_loopx_turn_codex_cli.py b/tests/test_loopx_turn_codex_cli.py index 17a751a49a..b466e35dd5 100644 --- a/tests/test_loopx_turn_codex_cli.py +++ b/tests/test_loopx_turn_codex_cli.py @@ -34,6 +34,7 @@ from loopx.control_plane.turn_driver.subagent_execution_topology import ( OPAQUE_REF_PATTERN, ) +from tests.control_plane.host_process_fixture import COUNTER_PROCESS_SOURCE FAILURE_ENVELOPE_FIXTURES = ( @@ -113,8 +114,7 @@ def _request( def _fake_codex(tmp_path: Path) -> tuple[Path, Path]: executable = tmp_path / "fake-codex" log_path = tmp_path / "codex-argv.jsonl" - executable.write_text( - """#!/usr/bin/env python3 + source = """#!/usr/bin/env python3 import json import os import pathlib @@ -168,11 +168,14 @@ def _fake_codex(tmp_path: Path) -> tuple[Path, Path]: raise SystemExit(9) if os.environ.get("FAKE_CODEX_CHILD_MARKER"): marker = os.environ["FAKE_CODEX_CHILD_MARKER"] - child = subprocess.Popen([sys.executable, "-c", - "import pathlib,signal,time;signal.signal(signal.SIGTERM,signal.SIG_IGN);" - "p=pathlib.Path(" + repr(marker) + ");n=0\\n" - "while True:\\n p.write_text(str(n));n+=1;time.sleep(.01)"]) - pathlib.Path(marker + ".pid").write_text(str(child.pid)) + child = subprocess.Popen([ + sys.executable, + "-c", + __COUNTER_PROCESS_SOURCE__, + marker, + marker + ".pid", + ".01", + ]) while not pathlib.Path(marker).exists(): time.sleep(.01) if os.environ.get("FAKE_CODEX_SLEEP"): @@ -191,7 +194,12 @@ def _fake_codex(tmp_path: Path) -> tuple[Path, Path]: "vision_unchanged_reason": "The fixture objective remains unchanged.", "summary": "One public fixture advanced." }), encoding="utf-8") -""", +""" + executable.write_text( + source.replace( + "__COUNTER_PROCESS_SOURCE__", + repr(COUNTER_PROCESS_SOURCE), + ), encoding="utf-8", ) executable.chmod(0o755)