fix(OMN-14641): stream coverage sweep subprocess output

scripts/ci/run_coverage_sweep_gate.py

@@ -50,6 +50,7 @@
 import json
 import subprocess
 import sys
+import time
 from pathlib import Path
 
 # Add src/ to path so omnimarket imports work in CI without editable install
@@ -68,6 +69,7 @@ def generate_coverage_json(
     test_path: str = "tests/",
     marker_expr: str = "not kafka",
     timeout_s: int = 1800,
+    heartbeat_s: int = 60,
 ) -> tuple[bool, str]:
     """Run the real test suite under coverage and write ``coverage.json``.
 
@@ -94,15 +96,34 @@ def generate_coverage_json(
         f"--cov={target_dir / 'src'}",
         f"--cov-report=json:{coverage_json}",
     ]
+    print(
+        "coverage-sweep-gate: running coverage generation command: " + " ".join(cmd),
+        flush=True,
+    )
     try:
-        proc = subprocess.run(
+        proc = subprocess.Popen(
             cmd,
             cwd=str(target_dir),
-            capture_output=True,
-            text=True,
-            timeout=timeout_s,
-            check=False,
         )
+        started_at = time.monotonic()
+        next_heartbeat_at = started_at + heartbeat_s
+        deadline_at = started_at + timeout_s
+
+        while proc.poll() is None:
+            now = time.monotonic()
+            if now >= deadline_at:
+                proc.kill()
+                proc.wait()
+                return False, f"coverage generation timed out after {timeout_s}s"
+            if now >= next_heartbeat_at:
+                elapsed_s = int(now - started_at)
+                print(
+                    "coverage-sweep-gate: coverage generation still running "
+                    f"after {elapsed_s}s",
+                    flush=True,
+                )
+                next_heartbeat_at = now + heartbeat_s
+            time.sleep(min(1.0, max(0.1, next_heartbeat_at - now)))
     except subprocess.TimeoutExpired as exc:
         return False, f"coverage generation timed out after {timeout_s}s: {exc}"
     except OSError as exc:
@@ -111,8 +132,7 @@ def generate_coverage_json(
     if not coverage_json.is_file():
         return False, (
             f"coverage generation exited {proc.returncode} but produced no "
-            f"coverage.json at {coverage_json}. stderr(tail): "
-            f"{proc.stderr[-2000:]}"
+            f"coverage.json at {coverage_json}."
         )
     return True, f"generated {coverage_json}"
 

tests/ci/test_run_coverage_sweep_gate.py

@@ -11,7 +11,6 @@
 from __future__ import annotations
 
 import json
-import subprocess
 import sys
 from pathlib import Path
 from unittest.mock import patch
@@ -165,33 +164,95 @@ def test_generate_helper_invokes_pytest_cov(self, tmp_path: Path) -> None:
 
         captured_cmd: list[str] = []
 
-        def _fake_run(cmd, **kwargs):  # type: ignore[no-untyped-def]
+        captured_kwargs: dict[str, object] = {}
+
+        class _FakeProcess:
+            returncode = 0
+
+            def __init__(self) -> None:
+                self._poll_count = 0
+
+            def poll(self) -> int | None:
+                self._poll_count += 1
+                if self._poll_count == 1:
+                    return None
+                (tmp_path / "coverage.json").write_text(json.dumps({"files": {}}))
+                return 0
+
+        def _fake_popen(cmd, **kwargs):  # type: ignore[no-untyped-def]
             captured_cmd.extend(cmd)
-            (tmp_path / "coverage.json").write_text(json.dumps({"files": {}}))
-            return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
+            captured_kwargs.update(kwargs)
+            return _FakeProcess()
 
-        with patch("run_coverage_sweep_gate.subprocess.run", side_effect=_fake_run):
-            ok, message = generate_coverage_json(tmp_path)
+        with (
+            patch("run_coverage_sweep_gate.subprocess.Popen", side_effect=_fake_popen),
+            patch("run_coverage_sweep_gate.time.sleep"),
+        ):
+            ok, message = generate_coverage_json(tmp_path, heartbeat_s=1)
 
         assert ok is True
         assert "coverage.json" in message
         assert "pytest" in captured_cmd
         assert any("--cov-report=json" in part for part in captured_cmd)
+        assert "capture_output" not in captured_kwargs
+        assert "stdout" not in captured_kwargs
+        assert "stderr" not in captured_kwargs
 
     def test_generate_helper_reports_failure_without_swallowing(
         self, tmp_path: Path
     ) -> None:
         from run_coverage_sweep_gate import generate_coverage_json
 
-        def _fake_run(cmd, **kwargs):  # type: ignore[no-untyped-def]
-            return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="boom")
+        class _FakeProcess:
+            returncode = 1
+
+            def poll(self) -> int:
+                return 1
+
+        def _fake_popen(cmd, **kwargs):  # type: ignore[no-untyped-def]
+            return _FakeProcess()
 
-        with patch("run_coverage_sweep_gate.subprocess.run", side_effect=_fake_run):
+        with patch("run_coverage_sweep_gate.subprocess.Popen", side_effect=_fake_popen):
             ok, message = generate_coverage_json(tmp_path)
 
         assert ok is False
         assert "boom" in message or "coverage.json" in message
 
+    def test_generate_helper_emits_parent_heartbeat(
+        self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        from run_coverage_sweep_gate import generate_coverage_json
+
+        class _FakeProcess:
+            returncode = 0
+
+            def __init__(self) -> None:
+                self._poll_count = 0
+
+            def poll(self) -> int | None:
+                self._poll_count += 1
+                if self._poll_count < 3:
+                    return None
+                (tmp_path / "coverage.json").write_text(json.dumps({"files": {}}))
+                return 0
+
+        times = iter([0.0, 1.0, 1.0, 2.0, 2.0])
+
+        with (
+            patch(
+                "run_coverage_sweep_gate.subprocess.Popen", return_value=_FakeProcess()
+            ),
+            patch(
+                "run_coverage_sweep_gate.time.monotonic",
+                side_effect=lambda: next(times),
+            ),
+            patch("run_coverage_sweep_gate.time.sleep"),
+        ):
+            ok, _ = generate_coverage_json(tmp_path, heartbeat_s=1)
+
+        assert ok is True
+        assert "coverage generation still running after 1s" in capsys.readouterr().out
+
     def test_generation_failure_exits_two(self, tmp_path: Path) -> None:
         from run_coverage_sweep_gate import main