diff --git a/gateway/kanban_watchers_notifier.py b/gateway/kanban_watchers_notifier.py index 512257e707..8a72fa8a49 100644 --- a/gateway/kanban_watchers_notifier.py +++ b/gateway/kanban_watchers_notifier.py @@ -18,6 +18,7 @@ from typing import Any, Callable, Optional from agent.i18n import t from gateway.kanban_watchers_common import _list_boards, _to_thread_process_service, logger +from gateway.wake import session_owned_by_profile def _kbc(): @@ -132,43 +133,6 @@ def _platform_names(mapping: Any) -> set[str]: return {getattr(platform, "value", str(platform)).lower() for platform in mapping} -def _session_owned_by_profile(config: Any, profile: Optional[str], session_id: Any) -> bool: - """True when a stateless (``api_server``) subscription's raw session id is canonically owned by - served *profile*'s own session store. - - A shared-listener mirror platform has no chat/thread/guild anchor a ``profile_routes`` entry - could match, so the session store itself is the ownership proof: the row must exist in that - profile's ``state.db`` under its own home and carry that profile's stamp (a NULL legacy stamp - belongs to the store's own profile — the same rule the dashboard's session routes apply). An - unserved profile, a missing row, a row stamped for another profile, or an unreadable store all - fail closed. - """ - if not session_id or not profile: - return False - profile = str(profile) - try: - from gateway.run import _multiplex_profile_homes - home = dict(_multiplex_profile_homes(config)).get(profile) - if home is None: - return False - from hermes_state import SessionDB - db = SessionDB(Path(home) / "state.db", read_only=True) - except Exception as exc: - logger.debug("kanban notifier: session ownership check unavailable for %s/%s: %s", - profile, session_id, exc) - return False - try: - row = db.get_session(str(session_id)) - except Exception as exc: - logger.debug("kanban notifier: session ownership lookup failed for %s/%s: %s", - profile, session_id, exc) - return False - finally: - with contextlib.suppress(Exception): - db.close() - return bool(row) and (row.get("profile_name") or profile) == profile - - def _adapter_for_subscription(runner: Any, platform: Any, sub: dict, owner_profile: Optional[str]) -> Any: """Resolve a durable route without turning a missing secondary bot into primary authority.""" adapter = runner._authorization_adapter(platform, owner_profile) @@ -214,7 +178,7 @@ def _adapter_for_subscription(runner: Any, platform: Any, sub: dict, owner_profi # session that lives in the served profile's state.db, never the platform. The default profile # keeps the historical fallthrough below (no store read). if profile != primary_profile and getattr(platform, "value", platform) == "api_server" \ - and _session_owned_by_profile(config, profile, chat): + and session_owned_by_profile(config, profile, chat): return primary return primary if profile == primary_profile else None diff --git a/gateway/run_notifications.py b/gateway/run_notifications.py index c2101ec351..835326f4a5 100644 --- a/gateway/run_notifications.py +++ b/gateway/run_notifications.py @@ -1064,6 +1064,7 @@ class GatewayNotificationsMixin: self-post them as a new role=user prompt. Other watch events wake the session via self-post. """ from gateway.wake import deliver_wake, persist_delegation_delivery + scope = contextlib.nullcontext() if evt.get("type") == "async_delegation": info = "Async delegation completion — persisting delivery row for api_server session %s (no wake turn)" fail = "Async delegation delivery persist failed for session %s: %s" @@ -1072,16 +1073,51 @@ class GatewayNotificationsMixin: info = "Watch pattern notification — waking api_server session %s via self-post" fail = "Watch notification self-post wake failed for session %s: %s" from agent.notification_presentation import diagnostic_process_event - deliver = lambda: deliver_wake(adapter, text=synth_text, session_id=raw_sid, + try: + served = await asyncio.to_thread(self._served_api_server_wake_profile, evt, raw_sid) + except LookupError as e: + logger.warning(fail, raw_sid, e) + return False + if served: + # The wake runs in the OWNING profile's scope, in-process (see ``deliver_wake``): + # the raw event carries no profile, so a completion scope was never installed. + from gateway.run import _async_profile_runtime_scope + source = SessionSource(platform=Platform.API_SERVER, chat_id=raw_sid, profile=served) + scope = _async_profile_runtime_scope(self._resolve_profile_home_for_source(source)) + deliver = lambda: deliver_wake(adapter, text=synth_text, session_id=raw_sid, profile=served, notification_category="diagnostic" if diagnostic_process_event(evt) else "result") # noqa: E731 try: logger.info(info, raw_sid) - await deliver() + async with scope: + await deliver() return True except Exception as e: logger.warning(fail, raw_sid, e) return False + def _served_api_server_wake_profile(self, evt: dict, raw_sid: str) -> Optional[str]: + """The served (non-primary) profile whose own session store holds *raw_sid*, else ``None`` + (the default profile's HTTP self-post). Blocking: reads served ``state.db`` files. + + A served profile's ``api_server`` turn binds the RAW session id as its session key, so its + completion event names no profile: the only ownership proof is the served profile's own + store, exactly the rung the Kanban notifier applies. An event whose source DOES name a + served profile (structured key / persisted origin) must be owned by that profile or it + raises ``LookupError`` — never a wake in the default profile's store. + """ + if not getattr(self.config, "multiplex_profiles", False): + return None + from gateway.run import _multiplex_profile_homes + from gateway.wake import session_owned_by_profile + primary = getattr(self, "_primary_profile_name", None) or "default" + hinted = str(getattr(self._build_process_event_source(evt), "profile", None) or "").strip() + if hinted and hinted != primary: + if session_owned_by_profile(self.config, hinted, raw_sid): + return hinted + raise LookupError(f"session is not in served profile {hinted!r}'s own store") + return next((name for name, _home in _multiplex_profile_homes(self.config) + if name != primary and session_owned_by_profile(self.config, name, raw_sid)), None) + def _resolve_injection_adapter(self, platform_name: str, source=None): """Adapter for a synthetic-event platform: alias-aware transport resolver first (one Platform.RELAY adapter fronts N logical platforms; native wins), literal ``p.value`` scan as @@ -1134,6 +1170,10 @@ class GatewayNotificationsMixin: return None platform_name = source.platform.value if hasattr(source.platform, "value") else str(source.platform) adapter = self._resolve_injection_adapter(platform_name, source) + if not adapter and platform_name == Platform.API_SERVER.value and getattr(source, "profile", None): + # A route-only served profile owns no adapter map; the shared listener wakes exactly the + # session that profile's store owns (proven in ``_self_post_api_server``), fail-closed. + adapter = self.adapters.get(Platform.API_SERVER) if not adapter: return False if not adapter_supports_push(adapter): diff --git a/gateway/wake.py b/gateway/wake.py index 726a3ea2ac..3416ff5332 100644 --- a/gateway/wake.py +++ b/gateway/wake.py @@ -39,6 +39,43 @@ class WakeNotAccepted(RuntimeError): """No adapter admission: retry without treating a healthy chat as dead.""" +def session_owned_by_profile(config: Any, profile: Optional[str], session_id: Any) -> bool: + """True when a stateless (``api_server``) destination's raw session id is canonically owned by + served *profile*'s own session store. + + A shared-listener mirror platform has no chat/thread/guild anchor a ``profile_routes`` entry + could match, so the session store itself is the ownership proof: the row must exist in that + profile's ``state.db`` under its own home and carry that profile's stamp (a NULL legacy stamp + belongs to the store's own profile — the same rule the dashboard's session routes apply). An + unserved profile, a missing row, a row stamped for another profile, or an unreadable store all + fail closed. Shared by the Kanban notifier and the background-process wake path. + """ + import contextlib + from pathlib import Path + if not session_id or not profile: + return False + profile = str(profile) + try: + from gateway.run import _multiplex_profile_homes + home = dict(_multiplex_profile_homes(config)).get(profile) + if home is None: + return False + from hermes_state import SessionDB + db = SessionDB(Path(home) / "state.db", read_only=True) + except Exception as exc: + logger.debug("wake: session ownership check unavailable for %s/%s: %s", profile, session_id, exc) + return False + try: + row = db.get_session(str(session_id)) + except Exception as exc: + logger.debug("wake: session ownership lookup failed for %s/%s: %s", profile, session_id, exc) + return False + finally: + with contextlib.suppress(Exception): + db.close() + return bool(row) and (row.get("profile_name") or profile) == profile + + async def admit_internal_event(adapter: Any, event: Any) -> None: """Require a concrete adapter admission, not merely a handler returning None. diff --git a/tests/gateway/test_background_process_served_apiserver_wake.py b/tests/gateway/test_background_process_served_apiserver_wake.py new file mode 100644 index 0000000000..4d24a67340 --- /dev/null +++ b/tests/gateway/test_background_process_served_apiserver_wake.py @@ -0,0 +1,95 @@ +"""A served profile's background-process completion wakes its ``api_server`` session in-process. + +A served (multiplexed) profile's ``api_server`` turn binds the RAW session id as its session key, so +the completion / watch event it leaves behind names no profile at all. The wake path therefore +self-posted it to the unprefixed shared listener with the PRIMARY key — resuming the session in the +DEFAULT profile's store — and an event whose source did name a route-only served profile resolved +no adapter and was deferred forever. + +Invariant (mirrors the Kanban one): the served profile whose own session store holds that exact +session is woken in-process, in that profile's scope, through the one shared adapter, without a +secondary credential; the default profile keeps its HTTP self-post; a served profile that does not +own the session never gets a wake in anyone's store. +""" + +import asyncio +from types import SimpleNamespace + +from gateway.config import Platform +from gateway.run import GatewayRunner +from tests.gateway.test_kanban_notifier_served_apiserver_wake import ( + RecordingApiServerAdapter, _FakeHttpSession, _own_session, served, # noqa: F401 (fixture) +) + +SESSION = "20260918_090000_aa11bb" # a served profile's api_server session (raw id) + + +def _make_runner(*, adapter): + runner = GatewayRunner.__new__(GatewayRunner) + runner.adapters = {Platform.API_SERVER: adapter} + runner._profile_adapters = {"builder": {}, "atlas": {}} + runner._profile_failed_platforms = {} + runner._primary_profile_name = "default" + runner.config = SimpleNamespace(multiplex_profiles=True, profile_routes=()) + return runner + + +def _completion_event(session_id, session_key=None): + """Exactly what ``_bind_api_server_session`` leaves on a process event: the raw id, no profile.""" + return {"type": "completion", "session_id": "proc_1", "command": "make build", "exit_code": 0, + "session_key": session_key if session_key is not None else session_id, + "platform": "api_server", "chat_id": session_id, "user_id": "", "user_name": "", "thread_id": ""} + + +def _wake(runner, evt): + return asyncio.run(runner._inject_watch_notification("[SYSTEM: make build exited 0]", evt)) + + +def test_served_profile_completion_wakes_in_process_only_for_the_session_it_owns(served, monkeypatch): + import aiohttp + + _own_session(served.builder, SESSION, "builder") + _FakeHttpSession.calls = [] + monkeypatch.setattr(aiohttp, "ClientSession", _FakeHttpSession) + adapter = RecordingApiServerAdapter() + adapter._api_key, adapter._host, adapter._port, adapter._model_name = "k" * 20, "127.0.0.1", 8642, "hermes" + + # Raw event (no profile anywhere): the owning store is the proof. + assert _wake(_make_runner(adapter=adapter), _completion_event(SESSION)) is True + assert [t["session_id"] for t in adapter.turns] == [SESSION] + assert adapter.homes == [str(served.builder)] and adapter.profiles == ["builder"] + assert _FakeHttpSession.calls == [] # no HTTP self-post, so no secondary API_SERVER_KEY + + # Event whose source names the route-only served profile: same in-process wake, not deferred. + adapter.turns.clear() + assert _wake(_make_runner(adapter=adapter), _completion_event(SESSION, f"agent:builder:api_server:dm:{SESSION}")) is True + assert [t["session_id"] for t in adapter.turns] == [SESSION] and adapter.profiles[-1] == "builder" + + # A served profile that does not own the session: no wake in anyone's store (retryable False). + adapter.turns.clear() + _own_session(served.atlas, "atlas-owned", "atlas") + assert _wake(_make_runner(adapter=adapter), _completion_event("atlas-owned", "agent:builder:api_server:dm:atlas-owned")) is False + assert adapter.turns == [] and _FakeHttpSession.calls == [] + + # Control: the default profile's own session keeps the HTTP self-post. + _own_session(served.root, "default-owned", "default") + assert _wake(_make_runner(adapter=adapter), _completion_event("default-owned")) is True + assert adapter.turns == [] + assert [c["headers"]["X-Hermes-Session-Id"] for c in _FakeHttpSession.calls] == ["default-owned"] + assert _FakeHttpSession.calls[0]["url"].endswith("/v1/chat/completions") + + +def test_single_profile_gateway_keeps_the_http_self_post(served, monkeypatch): + """Standalone (non-multiplex) gateway: no store scan, the historical HTTP self-post.""" + import aiohttp + + _own_session(served.builder, SESSION, "builder") + _FakeHttpSession.calls = [] + monkeypatch.setattr(aiohttp, "ClientSession", _FakeHttpSession) + adapter = RecordingApiServerAdapter() + adapter._api_key, adapter._host, adapter._port, adapter._model_name = "k" * 20, "127.0.0.1", 8642, "hermes" + runner = _make_runner(adapter=adapter) + runner.config = SimpleNamespace(multiplex_profiles=False, profile_routes=()) + assert _wake(runner, _completion_event(SESSION)) is True + assert adapter.turns == [] + assert [c["headers"]["X-Hermes-Session-Id"] for c in _FakeHttpSession.calls] == [SESSION]