diff --git a/gateway/run_active_work_verb.py b/gateway/run_active_work_verb.py index 790857bd61..350e1d57f4 100644 --- a/gateway/run_active_work_verb.py +++ b/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 diff --git a/hermes_cli/gateway_lifecycle_guards.py b/hermes_cli/gateway_lifecycle_guards.py index 4c6fe0b7eb..ac3680daec 100644 --- a/hermes_cli/gateway_lifecycle_guards.py +++ b/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: diff --git a/tests/hermes_cli/test_gateway_lifecycle_guards.py b/tests/hermes_cli/test_gateway_lifecycle_guards.py index 0e0a24cd22..53ca0a99a0 100644 --- a/tests/hermes_cli/test_gateway_lifecycle_guards.py +++ b/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")