fix(process): a supervisor a broken-pipe write dropped is still restarted by its waiter

Gate (2c): while the waiter thread sits in the untimed wait() it holds the Popen's waitpid lock, so
poll() keeps answering "running" for a supervisor that has died. A registration change in that
window writes into the dead pipe, gets BrokenPipe and drops the supervisor (_supervisor = None);
the waiter then saw "not current" and only reaped it, so the registered groups stayed uncovered
until the next registration change. The waiter now returns early only when another supervisor
already replaced it; a dropped one is restarted and replayed like a current one (a deliberate
release still ends at the empty-set check).

Test: kill the supervisor and drop it under the lock, as the broken-pipe caller does; a replacement
starts and kills the replayed group on EOF. Fails with the old check.

hermes_cli/process_identity.py

@@ -870,8 +870,10 @@ def _watch_supervisor(proc) -> None:
         global _supervisor
         proc.wait()
         with _supervisor_lock:
-            if _supervisor is not proc:
-                return  # released, or already replaced after a broken pipe: only reaped
+            if _supervisor is not None and _supervisor is not proc:
+                return  # already replaced: only reaped
+            # Still current, or dropped by a caller whose write hit its broken pipe: restart below if
+            # groups are still registered (a deliberate release leaves the set empty).
             _supervisor = None
             _prune_dead_supervised_pgids()
             if not _supervised_pgids:

tests/hermes_cli/test_process_identity_supervisor.py

@@ -890,3 +890,24 @@ def test_a_supervisor_killed_on_its_own_is_restarted_with_every_group_replayed(m
     finally:
         _kill(child.pid)
         child.wait(timeout=10)
+
+
+def test_a_supervisor_dropped_on_a_broken_pipe_is_still_restarted_by_its_waiter(monkeypatch):
+    """A caller can hit the dead supervisor's pipe before its waiter wakes (the waiter holds the
+    Popen's waitpid lock, so poll() still says "running"); that caller drops it. The waiter must still
+    restart coverage, or the registered groups wait for the next registration change."""
+    monkeypatch.setattr(process_identity, "_SUPERVISOR_RESTART_MIN_UPTIME_S", 0.0)
+    child = subprocess.Popen(_VICTIM, start_new_session=True)
+    try:
+        process_identity.own_process_groups([child.pid])
+        first = process_identity._supervisor
+        with process_identity._supervisor_lock:  # the broken-pipe caller, before the waiter runs
+            first.kill()
+            process_identity._supervisor = None
+        wait_until(lambda: process_identity._supervisor not in (None, first), timeout=10,
+                   what="a replacement started for the supervisor a broken pipe dropped")
+        process_identity._supervisor.stdin.close()
+        wait_until(lambda: child.poll() is not None, timeout=15, what="the replayed group killed")
+    finally:
+        _kill(child.pid)
+        child.wait(timeout=10)