hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

fix(gateway): wake a served profile's api_server session in-process for background-process completions

Browse Source 
A served (multiplexed) profile's api_server turn binds the raw session id
as its session key, so its watch/completion event names no profile and the
wake path self-posted it unprefixed with the primary key — resuming the
session in the DEFAULT profile's store — while an event whose source did
name a route-only served profile resolved no adapter and was deferred
forever.

_self_post_api_server now proves ownership through the served profile's
own session store (the rung the Kanban notifier already applies), scans
served stores when the raw event carries no hint, runs the wake under that
profile's scope via deliver_wake(profile=...), and fails closed for a hinted
profile that does not own the session. The ownership helper moves to
gateway/wake.py so both callers share one function.
This commit is contained in:
teknium1
2026-09-18 03:47:52 -07:00
committed by Teknium
parent 5dd426b774
commit d8edd60399
4 changed files with 176 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
gateway/kanban_watchers_notifier.py
View File

@@ -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

44
gateway/run_notifications.py
View File

@@ -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):

37
gateway/wake.py
View File

@@ -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.

95
tests/gateway/test_background_process_served_apiserver_wake.py Normal file
View File

@@ -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]