From f484e2cf34e2f52e055b7649bbe7c7e58a342466 Mon Sep 17 00:00:00 2001
From: Gergo Magyar <gergomagyar0@gmail.com>
Date: Sun, 2 Aug 2026 12:12:08 +0000
Subject: [PATCH] fix(eval): drain with read1 so progress surfaces before the
 child exits
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The live proof run showed the driver's own lines streaming correctly and
every one of the sweep's 24 cell lines sharing one timestamp
(12:05:29.96) four hours after the sweep began — the exact symptom
echo_stdout was added to remove, still present for the phase that
actually takes the fifteen hours.

`_drain` read with `pipe.read(8192)`. On a BufferedReader that blocks
until it has all 8192 bytes or the pipe closes; it does not return short
reads. A sweep emits a couple of short lines per ~45-minute cell and
never fills 8 KB, so everything sat in the buffer until the process
exited. The tail and the capture were unaffected — they only need the
bytes eventually — which is why nothing caught it before.

The existing echo test could not have: its child wrote one line and
exited immediately, so EOF made `read` return. The new test makes the
child refuse to exit until the echoed line has been observed, so an
implementation that only flushes at EOF deadlocks and fails on the
timeout instead of passing on a technicality. Verified it fails with
`read` and passes with `read1`.
---
 eval/tests/test_process_control.py     | 43 ++++++++++++++++++++++++++
 eval/workflow_bench/process_control.py |  6 +++-
 2 files changed, 48 insertions(+), 1 deletion(-)

diff --git a/eval/tests/test_process_control.py b/eval/tests/test_process_control.py
index c5a335aa8..c3fdf1976 100644
--- a/eval/tests/test_process_control.py
+++ b/eval/tests/test_process_control.py
@@ -9,6 +9,7 @@ import sys
 import time
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -141,6 +142,48 @@ def test_echo_stdout_streams_child_progress_and_stays_off_by_default(capfd) -> N
     assert "starting" in echoed.stdout_tail
 
 
+def test_echo_reaches_the_log_while_the_child_is_still_running(tmp_path: Path, monkeypatch) -> None:
+    """Streaming has to be prompt, not merely eventual.
+
+    A sweep emits one line every ~45 minutes. Draining with `read(8192)` still
+    delivers every byte, so the tail and the capture look correct — but nothing
+    surfaces until the pipe closes, which turns a 15-hour job into a silent one
+    and is the whole reason this passthrough exists.
+
+    The child here refuses to exit until the echoed line has been observed, so
+    an implementation that only flushes at EOF deadlocks and fails on the
+    timeout rather than passing on a technicality.
+    """
+    released = tmp_path / "echo-observed"
+
+    class Sink:
+        def write(self, data: bytes) -> int:
+            if b"first-line" in data:
+                released.write_text("go")
+            return len(data)
+
+        def flush(self) -> None:
+            pass
+
+    monkeypatch.setattr(process_control.sys, "stderr", SimpleNamespace(buffer=Sink()))
+    script = """
+import pathlib, sys, time
+sys.stdout.write('first-line\\n')
+sys.stdout.flush()
+target = pathlib.Path(%r)
+for _ in range(400):
+    if target.exists():
+        break
+    time.sleep(0.05)
+""" % str(released)
+
+    result = run_managed([PYTHON, "-c", script], timeout=15, echo_stdout=True)
+
+    assert released.exists(), "the line never reached the echo sink while the child ran"
+    assert result.ok
+    assert "first-line" in result.stdout_tail
+
+
 def test_incomplete_stdin_delivery_cannot_report_success() -> None:
     result = run_managed(
         [PYTHON, "-c", "import os,time; os.close(0); time.sleep(0.05)"],
diff --git a/eval/workflow_bench/process_control.py b/eval/workflow_bench/process_control.py
index ea64be75e..b9df301e7 100644
--- a/eval/workflow_bench/process_control.py
+++ b/eval/workflow_bench/process_control.py
@@ -128,7 +128,11 @@ def _drain(
     echo: BinaryIO | None = None,
 ) -> None:
     try:
-        while chunk := pipe.read(8192):
+        # read1, not read: on a BufferedReader, read(n) blocks until it has all
+        # n bytes or the pipe closes. A child that emits a line every 45 minutes
+        # never fills 8 KB, so its output would surface only when it exits —
+        # which is precisely what echo_stdout exists to avoid.
+        while chunk := pipe.read1(8192):
             tail.append(chunk)
             if capture is not None:
                 capture.append(chunk)