test(kanban): make progress-stall policy tests probe-deterministic (t_5457397a)

hermes_cli/kanban_db.py

@@ -14257,22 +14257,37 @@ def _worker_cpu_active(pid: int) -> bool:
     keepalive tickers. Unknown process state (``ps`` missing/failing/
     unparseable) returns True so it never authorizes a kill.
     """
-    import subprocess
     try:
-        ps = subprocess.run(
-            ["ps", "-A", "-o", "pid=,ppid=,pcpu="],
-            stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=2,
-            check=True,
-        )
-        for line in ps.stdout.splitlines():
-            fields = line.split()
-            if len(fields) != 3:
-                continue
-            proc, cpu = int(fields[0]), float(fields[2])
-            if proc == pid and cpu > 0:
-                return True
+        return _cpu_active_in_table(_process_cpu_table(), pid)
     except (OSError, ValueError, subprocess.SubprocessError):
         return True  # Unknown process state must not authorize a kill.
+
+
+def _process_cpu_table() -> str:
+    """Raw ``pid ppid pcpu`` table from ``ps`` (the probe's only I/O).
+
+    Split out so tests can feed a deterministic table instead of depending on
+    a loaded host's scheduler to make a real child read 0.0% in time.
+    """
+    return subprocess.run(
+        ["ps", "-A", "-o", "pid=,ppid=,pcpu="],
+        stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=2,
+        check=True,
+    ).stdout
+
+
+def _cpu_active_in_table(table: str, pid: int) -> bool:
+    """Pure parse: True iff ``pid`` itself shows CPU > 0 (children never veto).
+
+    Raises ValueError on an unparseable row so the caller treats it as unknown.
+    """
+    for line in table.splitlines():
+        fields = line.split()
+        if len(fields) != 3:
+            continue
+        proc, cpu = int(fields[0]), float(fields[2])
+        if proc == pid and cpu > 0:
+            return True
     return False
 
 

tests/hermes_cli/test_kanban_progress_stall.py

@@ -2,9 +2,21 @@
 
 The stall decision rides on the agent's own ``progress_at`` timestamp (stamped
 onto heartbeat events by the worker bridge). The process probe is a veto only.
-Every process-shaped test here drives the REAL probe against a REAL process.
+
+Probe determinism (t_5457397a): the reclaim/stall POLICY tests still run a REAL
+worker process (so reclaim really terminates it) but feed the probe a
+deterministic ``ps`` table through the ``kb._process_cpu_table`` seam. Waiting
+for the real ``ps`` to read a fresh child as 0.0% is scheduler-dependent -- on
+Linux procps ``pcpu`` is lifetime cputime/elapsed, so a child whose startup
+burned 30 ms needs ~30 s before it rounds to 0.0, far longer on a loaded
+runner -- and it ejected unrelated PRs from the merge queue. The table is still
+parsed by the production ``_cpu_active_in_table``, so the children-never-veto
+rule is exercised for real. The real probe keeps its own integration tests:
+busy process -> active (fail-safe direction, load-immune) and in-flight process
+-> idle (skips with the measured reason when the host cannot show idle).
 """
 import json
+import os
 import socket
 import subprocess
 import sys
@@ -40,18 +52,62 @@ def _reap_in_background(proc):
     return proc
 
 
-def _wait_idle(pid, timeout=30.0):
-    """Wait until the REAL probe reads idle on 6 consecutive samples."""
-    deadline, streak = time.monotonic() + timeout, 0
+class _FakePs:
+    """Deterministic ``ps`` table: every registered pid reads 0.0% CPU."""
+
+    def __init__(self):
+        self.rows = []  # (pid, ppid, pcpu)
+        self.reads = 0
+
+    def add(self, pid, ppid=None, pcpu=0.0):
+        self.rows.append((int(pid), int(ppid if ppid is not None else os.getpid()), pcpu))
+
+    def __call__(self):
+        self.reads += 1
+        return "".join(f"{p:>7} {pp:>7} {c:4.1f}\n" for p, pp, c in self.rows)
+
+
+@pytest.fixture
+def fake_ps(monkeypatch):
+    fake = _FakePs()
+    monkeypatch.setattr(kb, "_process_cpu_table", fake)
+    return fake
+
+
+def _wait_idle_real(pid, timeout=60.0):
+    """Wait until the REAL probe reads idle on 6 consecutive samples.
+
+    Only the real-probe integration test uses this. A miss caused by the host
+    (``ps`` failing/timing out, or load above the core count) is a SKIP with
+    the measured reason; a miss on an unloaded host with a working ``ps`` is a
+    genuine FAIL (e.g. the probe regressed to always-active).
+    """
+    deadline, streak, probe_errors = time.monotonic() + timeout, 0, []
     while time.monotonic() < deadline:
-        streak = 0 if kb._worker_cpu_active(pid) else streak + 1
+        try:
+            active = kb._cpu_active_in_table(kb._process_cpu_table(), pid)
+        except (OSError, ValueError, subprocess.SubprocessError) as exc:
+            probe_errors.append(type(exc).__name__)
+            active = True
+        streak = 0 if active else streak + 1
         if streak >= 6:
             return
         time.sleep(0.25)
-    pytest.fail(f"pid {pid} never read idle to the real probe")
+    load = os.getloadavg()[0] if hasattr(os, "getloadavg") else 0.0
+    cores = os.cpu_count() or 1
+    if probe_errors or load > cores:
+        pytest.skip(
+            f"host could not show pid {pid} idle within {timeout:.0f}s: "
+            f"probe errors={sorted(set(probe_errors)) or 'none'} "
+            f"({len(probe_errors)}x), loadavg1={load:.1f} on {cores} cores"
+        )
+    pytest.fail(
+        f"pid {pid} never read idle to the real probe on an unloaded host "
+        f"(loadavg1={load:.1f}, {cores} cores, no probe errors)"
+    )
 
 
-def _in_flight_worker(server):
+def _in_flight_worker(server, fake_ps=None):
     """Real process blocked on an in-flight request: 0% CPU, no children."""
     code = (
         "import socket,sys;s=socket.create_connection(('127.0.0.1',int(sys.argv[1])));"
@@ -63,7 +119,10 @@ def _in_flight_worker(server):
     conn, _ = server.accept()  # the request is now in flight and never answered
     conn.recv(64)
     _in_flight_worker.conns.append(conn)
-    _wait_idle(proc.pid)
+    if fake_ps is None:
+        _wait_idle_real(proc.pid)
+    else:
+        fake_ps.add(proc.pid)
     return proc
 
 
@@ -92,10 +151,10 @@ def _count(board, tid, kind):
     ).fetchone()[0]
 
 
-def test_run_7914_shape_stalls_at_15_reclaims_at_25_escalates_after_two(board, monkeypatch, silent_server):
+def test_run_7914_shape_stalls_at_15_reclaims_at_25_escalates_after_two(board, monkeypatch, silent_server, fake_ps):
     now = int(time.time())
     monkeypatch.setattr(kb.time, "time", lambda: now)
-    proc = _in_flight_worker(silent_server)
+    proc = _in_flight_worker(silent_server, fake_ps)
     # Last real progress = the API call start 15 min ago; the wrapper keeps
     # heartbeating (fresh last_heartbeat_at) with that same progress_at.
     tid = _running(board, now, proc.pid, progress_at=now - 900)
@@ -117,7 +176,7 @@ def test_run_7914_shape_stalls_at_15_reclaims_at_25_escalates_after_two(board, m
     assert kb.get_task(board, tid).status == "ready"
     proc.wait(timeout=10)  # the real worker was terminated
     for _ in range(2):
-        again = _in_flight_worker(silent_server)
+        again = _in_flight_worker(silent_server, fake_ps)
         claimed = kb.claim_task(board, tid)
         board.execute("UPDATE task_runs SET started_at=? WHERE id=?", (now - 1500, claimed.current_run_id))
         board.execute("UPDATE tasks SET worker_pid=? WHERE id=?", (again.pid, tid))
@@ -129,15 +188,27 @@ def test_run_7914_shape_stalls_at_15_reclaims_at_25_escalates_after_two(board, m
     assert board.execute(
         "SELECT COUNT(*) FROM task_runs WHERE task_id=? AND outcome='stalled'", (tid,)
     ).fetchone()[0] == 3
+    assert fake_ps.reads >= 5  # every stall decision consulted the probe veto
+
+
+def test_real_probe_reads_in_flight_worker_as_idle(silent_server):
+    """Integration proof for the REAL ``ps`` probe: a process blocked on an
+    unanswered request reads idle, exactly like a dead socket. This is why
+    the probe may only veto, never authorize, a reclaim."""
+    proc = _in_flight_worker(silent_server)  # waits on the real probe
+    try:
+        assert proc.poll() is None
+    finally:
+        proc.kill()
 
 
-def test_healthy_in_flight_wait_with_advancing_progress_is_never_reclaimed(board, monkeypatch, silent_server):
-    """Argus repro: real idle-but-in-flight process + advancing progress_at."""
+def test_healthy_in_flight_wait_with_advancing_progress_is_never_reclaimed(board, monkeypatch, silent_server, fake_ps):
+    """Argus repro: idle-but-in-flight process + advancing progress_at."""
     now = int(time.time())
     monkeypatch.setattr(kb.time, "time", lambda: now)
-    proc = _in_flight_worker(silent_server)
+    proc = _in_flight_worker(silent_server, fake_ps)
     try:
-        # The real probe cannot tell this from a dead socket...
+        # The probe cannot tell this from a dead socket...
         assert kb._worker_cpu_active(proc.pid) is False
         # ...but the agent reports progress (turns completing) each window.
         tid = _running(board, now, proc.pid, progress_at=now - 120, started_ago=3000)
@@ -152,7 +223,7 @@ def test_healthy_in_flight_wait_with_advancing_progress_is_never_reclaimed(board
         proc.kill()
 
 
-def _stalled_worker_with_persistent_child(server):
+def _stalled_worker_with_persistent_child(server, fake_ps):
     """7914 shape + one persistent idle child (execute_code kernel / LSP server)."""
     code = (
         "import socket,subprocess,sys;"
@@ -173,16 +244,18 @@ def _stalled_worker_with_persistent_child(server):
         kids = [int(p) for p, pp in (l.split() for l in ps.splitlines()) if int(pp) == proc.pid]
         time.sleep(0.1)
     assert kids  # the persistent idle child really exists
-    _wait_idle(proc.pid)
+    fake_ps.add(proc.pid)
+    for kid in kids:  # the child is in the table, parented to the worker
+        fake_ps.add(kid, ppid=proc.pid)
     return proc, kids
 
 
-def test_persistent_idle_child_does_not_hide_a_stall(board, monkeypatch, silent_server):
+def test_persistent_idle_child_does_not_hide_a_stall(board, monkeypatch, silent_server, fake_ps):
     """Argus round 2: an any-child veto blinded the detector to every worker
     holding a kernel/LSP child. Stale progress + idle child -> reclaimed at T+25."""
     now = int(time.time())
     monkeypatch.setattr(kb.time, "time", lambda: now)
-    proc, kids = _stalled_worker_with_persistent_child(silent_server)
+    proc, kids = _stalled_worker_with_persistent_child(silent_server, fake_ps)
     try:
         tid = _running(board, now, proc.pid, progress_at=now - 900)
         assert kb.detect_progress_stalls(board, stall_seconds=900, reclaim_seconds=1500) == []
@@ -224,7 +297,9 @@ def test_cpu_active_worker_vetoes_stale_progress(board, monkeypatch):
 def test_process_probe_failure_never_authorizes_reclaim(board, monkeypatch, silent_server, failure):
     now = int(time.time())
     monkeypatch.setattr(kb.time, "time", lambda: now)
-    proc = _in_flight_worker(silent_server)
+    # An UNINSTALLED table: skips the real-idle wait, but the probe itself stays
+    # real so it runs into the broken ``ps`` below.
+    proc = _in_flight_worker(silent_server, _FakePs())
     try:
         def broken(*_a, **_kw):
             raise failure
@@ -238,10 +313,10 @@ def broken(*_a, **_kw):
         proc.kill()
 
 
-def test_run_without_progress_signal_is_unknown_and_never_reclaimed(board, monkeypatch, silent_server):
+def test_run_without_progress_signal_is_unknown_and_never_reclaimed(board, monkeypatch, silent_server, fake_ps):
     now = int(time.time())
     monkeypatch.setattr(kb.time, "time", lambda: now)
-    proc = _in_flight_worker(silent_server)
+    proc = _in_flight_worker(silent_server, fake_ps)
     try:
         tid = _running(board, now, proc.pid, progress_at=None)
         rid = kb.get_task(board, tid).current_run_id
@@ -252,10 +327,10 @@ def test_run_without_progress_signal_is_unknown_and_never_reclaimed(board, monke
         proc.kill()
 
 
-def test_dispatch_tick_observes_progress_despite_fresh_heartbeats(board, monkeypatch, silent_server):
+def test_dispatch_tick_observes_progress_despite_fresh_heartbeats(board, monkeypatch, silent_server, fake_ps):
     now = int(time.time())
     monkeypatch.setattr(kb.time, "time", lambda: now)
-    proc = _in_flight_worker(silent_server)
+    proc = _in_flight_worker(silent_server, fake_ps)
     try:
         tid = _running(board, now, proc.pid, progress_at=now - 901)
         kb.dispatch_once(board, max_spawn=1, spawn_fn=lambda *args: 123)