diff --git a/autoresearch/prefill/supervisor.py b/autoresearch/prefill/supervisor.py index f6150b5..09c22ff 100644 --- a/autoresearch/prefill/supervisor.py +++ b/autoresearch/prefill/supervisor.py @@ -12,6 +12,7 @@ import shutil import socket import subprocess +import threading import time import urllib.request from pathlib import Path @@ -242,17 +243,46 @@ def run_gan_experiment( "--candidate-file", str(candidate_path), "--state-file", str(state_path), ] - result = subprocess.run( + process = subprocess.Popen( command, - input="/continue\n/quit\n", + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, text=True, - capture_output=True, - timeout=timeout_s, + bufsize=1, cwd=repo, ) - output = result.stdout + result.stderr - if result.returncode != 0: - raise RuntimeError(f"GAN experiment failed ({result.returncode}): {output[-4000:]}") + assert process.stdin is not None + assert process.stdout is not None + process.stdin.write("/continue\n/quit\n") + process.stdin.flush() + process.stdin.close() + timed_out = threading.Event() + + def terminate_on_timeout() -> None: + timed_out.set() + process.kill() + + timer = threading.Timer(timeout_s, terminate_on_timeout) + timer.daemon = True + timer.start() + lines: list[str] = [] + try: + for line in process.stdout: + print(line, end="", flush=True) + lines.append(line) + returncode = process.wait() + finally: + timer.cancel() + output = "".join(lines) + if timed_out.is_set(): + raise TimeoutError( + f"GAN experiment exceeded {timeout_s}s: {output[-4000:]}", + ) + if returncode != 0: + raise RuntimeError( + f"GAN experiment failed ({returncode}): {output[-4000:]}", + ) matches = re.findall(r"run=(br_[0-9a-f]+)", output) if not matches: raise RuntimeError("GAN experiment produced no benchmark run id") diff --git a/tests/inference_engine/bench/test_autoresearch_supervisor.py b/tests/inference_engine/bench/test_autoresearch_supervisor.py index 3a00471..b1ec2be 100644 --- a/tests/inference_engine/bench/test_autoresearch_supervisor.py +++ b/tests/inference_engine/bench/test_autoresearch_supervisor.py @@ -115,3 +115,20 @@ def test_supervisor_predeploys_before_real_strategy_proposal(): ) assert "phase=predeploy-current" in body assert "phase=strategy-proposal real-gemma" in body + + +def test_gan_subprocess_output_is_streamed_not_captured(): + source = ( + Path(__file__).resolve().parents[3] + / "autoresearch" + / "prefill" + / "supervisor.py" + ).read_text() + body = source[ + source.index("def run_gan_experiment"): + source.index("def read_results") + ] + assert "subprocess.Popen(" in body + assert "stderr=subprocess.STDOUT" in body + assert 'print(line, end="", flush=True)' in body + assert "capture_output=True" not in body