fix(gateway): -p <profile> stop counts its /bg agents and refuses when the host doesn't answer

The profile-scoped active-work answer left out detached /bg and /btw side agents, which
unserving the profile interrupts, and read authority state on the socket thread after the
loop callback returned; a race there raised, the CLI read the error as no work and went
ahead. Everything is now collected inside the loop callback, side agents are counted, and a
missing answer from the host that serves the profile refuses instead of allowing.

gateway/run_active_work_verb.py

@@ -39,32 +39,41 @@ def _profile_units(runner, profile: str, units: list[dict[str, Any]]) -> list[di
     home = (getattr(runner, "_served_profile_homes", None) or {}).get(profile)
     registry = getattr(runner, "session_authorities", None)
     authority = registry.for_home(home) if registry is not None and home is not None else None
+    from gateway.session_runtime_workers import side_agents
+
     managed = sum(not w.closed.is_set() for w in getattr(authority, "_managed_workers", {}).values())
     crons = len(getattr(authority, "_cron_cancellations", None) or {})
-    return mine + [{"kind": "managed", "pid": os.getpid()}] * managed + [{"kind": "cron", "job_id": "?", "pid": os.getpid()}] * crons
+    sides = len(side_agents(authority)) if authority is not None else 0
+    pid = os.getpid()
+    return (mine + [{"kind": "managed", "pid": pid}] * managed + [{"kind": "cron", "job_id": "?", "pid": pid}] * crons
+            + [{"kind": "side_agent", "pid": pid}] * sides)
 
 
 def active_work_verb(runner, loop) -> Callable[[], dict[str, Any]]:
     """Handler for the control socket: runs on its executor thread, reads the runner on the loop."""
 
     def _handler(params: dict[str, Any]) -> dict[str, Any]:
-        box: list[list] = []
+        profile = params.get("profile")
+        scoped = isinstance(profile, str) and bool(profile)
+        box: list[dict[str, Any]] = []
         done = threading.Event()
 
         def _collect() -> None:
+            # Everything that reads live authority state runs here, on the loop: a read on the socket
+            # thread can race a worker being added and raise, and an error reads as "no work".
             try:
-                box.append(runner._describe_active_work())
+                units = list(runner._describe_active_work()) + _delegation_units(time.time())
+                if scoped:
+                    units = _profile_units(runner, profile, units)
+                    box.append({"active_agents": len(units), "active_work": units, "profile": profile})
+                else:
+                    box.append({"active_agents": runner._active_work_count(), "active_work": units})
             finally:
                 done.set()
 
         loop.call_soon_threadsafe(_collect)
-        done.wait(timeout=5.0)
-        units = list(box[0]) if box else []
-        units += _delegation_units(time.time())
-        profile = params.get("profile")
-        if isinstance(profile, str) and profile:
-            units = _profile_units(runner, profile, units)
-            return {"active_agents": len(units), "active_work": units, "pid": os.getpid(), "profile": profile}
-        return {"active_agents": runner._active_work_count(), "active_work": units, "pid": os.getpid()}
+        if not done.wait(timeout=5.0) or not box:
+            raise RuntimeError("active-work: the gateway loop did not answer")
+        return {**box[0], "pid": os.getpid()}
 
     return _handler

hermes_cli/gateway_lifecycle_guards.py

@@ -70,14 +70,16 @@ def _candidate_homes(system: bool = False) -> list[Path]:
 def refusal_lines(verb: str, homes: list[Path], *, flag_hint: str = f"hermes gateway {{verb}} {INTERRUPT_FLAG}",
                   profile: Optional[str] = None) -> list[str]:
     """Refusal text for ``verb`` while the first answering gateway among ``homes`` has live work, else []."""
-    from hermes_cli.update_cmd_drain_report import describe_active_work_unit
-
     for home in homes:
         work = live_work(home, profile)
         if work is not None:
-            break
-    else:
-        return []
+            return _work_lines(verb, home, work, flag_hint)
+    return []
+
+
+def _work_lines(verb: str, home: Path, work: dict[str, Any], flag_hint: str) -> list[str]:
+    from hermes_cli.update_cmd_drain_report import describe_active_work_unit
+
     units = [u for u in work.get("active_work") or [] if isinstance(u, dict)]
     count = max(len(units), int(work.get("active_agents") or 0))
     if not count:
@@ -96,8 +98,15 @@ def refuse_while_sessions_live(verb: str, args, *, host_home: Optional[Path] = N
     ``host_home``/``profile``: a served profile's stop/restart unserves only that profile in the host,
     so only its work counts."""
     if profile:
-        lines = refusal_lines(verb, [Path(host_home)], profile=profile,
-                              flag_hint=f"hermes -p {profile} gateway {{verb}} {INTERRUPT_FLAG}")
+        hint = f"hermes -p {profile} gateway {{verb}} {INTERRUPT_FLAG}"
+        work = live_work(Path(host_home), profile)
+        if work is None:
+            # The host serving this profile is up; no answer is not "idle", so fail closed.
+            lines = [f"Refusing to {verb} the gateway: profile '{profile}' did not report its live work.",
+                     "  Retry in a moment, or interrupt whatever it is running deliberately:",
+                     f"    {hint.format(verb=verb)}"]
+        else:
+            lines = _work_lines(verb, Path(host_home), work, hint)
     else:
         lines = refusal_lines(verb, _candidate_homes(getattr(args, "system", False)))
     if not lines:

tests/hermes_cli/test_gateway_lifecycle_guards.py

@@ -279,3 +279,38 @@ def test_active_work_verb_scoped_to_a_profile_names_only_its_turns(monkeypatch):
     scoped = handler({"profile": "coder"})
     assert scoped["active_agents"] == 1 and [u["session"] for u in scoped["active_work"]] == ["agent:coder:telegram:dm:7"]
 
+
+
+def test_active_work_verb_counts_a_profiles_side_agent_and_reads_it_on_the_loop(monkeypatch):
+    """A profile whose only work is a running /bg side agent is not idle (unserving it interrupts the
+    agent), and every live read happens inside the loop callback, never on the socket thread."""
+    from gateway.run_active_work_verb import active_work_verb
+    from tools import async_delegation
+    monkeypatch.setattr(async_delegation, "list_async_delegations", list)
+    on_loop = []
+    authority = SimpleNamespace(_managed_workers={}, _cron_cancellations={},
+                                _side_agents={object(): [object()]})
+    reads = []
+    registry = SimpleNamespace(for_home=lambda home: reads.append(on_loop[-1]) or authority)
+    runner = SimpleNamespace(_describe_active_work=list, _active_work_count=lambda: 1,
+                             _served_profile_homes={"coder": "/h/coder"}, session_authorities=registry)
+
+    def _call_soon(fn):
+        on_loop.append(True)
+        fn()
+        on_loop.append(False)
+
+    loop = SimpleNamespace(call_soon_threadsafe=_call_soon)
+    answer = active_work_verb(runner, loop)({"profile": "coder"})
+    assert answer["active_agents"] == 1 and answer["active_work"][0]["kind"] == "side_agent"
+    assert reads == [True], "the authority must be read inside the loop callback"
+
+
+def test_profile_stop_fails_closed_when_the_host_does_not_report(monkeypatch, capsys):
+    """No answer from the host that serves the profile is not "idle": refuse unless --interrupt."""
+    from hermes_cli import gateway_lifecycle_guards as guards
+    monkeypatch.setattr(guards, "live_work", lambda home, profile=None: None)
+    with pytest.raises(SystemExit):
+        guards.refuse_while_sessions_live("stop", SimpleNamespace(interrupt=False), host_home="/h", profile="coder")
+    assert "did not report its live work" in capsys.readouterr().out
+    guards.refuse_while_sessions_live("stop", SimpleNamespace(interrupt=True), host_home="/h", profile="coder")