From 5e3ba832059a93126a765f662dd6f149dfe3246c Mon Sep 17 00:00:00 2001 From: Rand Lee Date: Sun, 9 Aug 2026 19:48:12 -0700 Subject: [PATCH 01/11] =?UTF-8?q?feat:=20inject=5Finternal=5Fmessage=20?= =?UTF-8?q?=E2=80=94=20public=20profile-aware=20injection=20API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GatewayRunner.inject_internal_message(profile, platform, chat_id, text, notice_text) enables plugins (e.g. hermes-atm) to inject host-originated messages through the existing adapterβ†’gateway dispatch path with internal=True. - Resolves adapter from _profile_adapters[profile] or self.adapters - Constructs SessionSource + MessageEvent(internal=True) - Supports optional notice_text for visible πŸ“¬ observability - Fire-and-forget: queues event via adapter.handle_message() AL17 deployable contract per c50c4232. --- gateway/run.py | 97 ++++++ tests/gateway/test_inject_internal_message.py | 279 ++++++++++++++++++ 2 files changed, 376 insertions(+) create mode 100644 tests/gateway/test_inject_internal_message.py diff --git a/gateway/run.py b/gateway/run.py index ef626504dc4b..d38d0af34c8a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -14131,6 +14131,103 @@ def _create_adapter( return None + # ------------------------------------------------------------------ + # Public injection API β€” exposed to plugins via gateway:startup hook + # ------------------------------------------------------------------ + + async def inject_internal_message( + self, + profile: str, + platform: Platform, + chat_id: str, + text: str, + notice_text: Optional[str] = None, + ) -> Optional[str]: + """Route an internal message through a platform adapter to the agent. + + Used by plugins (e.g., the ATM graft bridge) to inject synthetic + host-originated messages that enter the agent loop via the normal + adapterβ†’gateway dispatch path, but with ``internal=True`` so they + bypass user authorization, startup restore, and other user-facing + guards. + + The adapter is selected from ``self._profile_adapters[profile]`` + when ``profile`` names a secondary profile; otherwise + ``self.adapters`` (the running profile's map) is used. + + .. note:: + + This method is **fire-and-forget** β€” it awaits + ``adapter.handle_message(event)`` which spawns a background + task and returns quickly. The agent response, if any, is + delivered through the normal gateway delivery path. + + Args: + profile: Profile name to route through (looked up in + ``_profile_adapters``; falls back to + ``self.adapters``). + platform: Platform enum (e.g. ``Platform.TELEGRAM``). + chat_id: Target chat ID for the platform. + text: Message text to inject. + notice_text: Optional visible notice to send to the chat + before routing the event (observability surface). + + Returns: + ``None`` β€” the method returns after queuing the event for + dispatch. + """ + # Resolve the adapter for the requested profile. + adapter = None + if profile and self._profile_adapters: + profile_map = self._profile_adapters.get(profile) + if profile_map: + adapter = profile_map.get(platform) + if adapter is None: + adapter = self.adapters.get(platform) + + if adapter is None: + logger.warning( + "inject_internal_message: no adapter for profile=%s platform=%s", + profile, platform, + ) + return None + + # Construct SessionSource β€” deliberately NOT a synthetic Telegram + # user message; the platform and chat_id reflect the real delivery + # channel so session resolution keys on the right identity. + source = SessionSource( + platform=platform, + chat_id=chat_id, + chat_type="dm", + profile=profile or None, + ) + + # Construct MessageEvent with internal=True so the gateway skips + # authorization, startup-restore queueing, and scale-to-zero + # clocks β€” this is a host-originated event, not user traffic. + event = MessageEvent( + text=text, + source=source, + internal=True, + ) + + # Optional visible notice (observability, not a duplicate message). + if notice_text: + try: + await adapter.send(chat_id, notice_text) + except Exception as exc: + logger.warning( + "inject_internal_message: failed to send notice to %s: %s", + chat_id, exc, + ) + + # Route through the adapter's handle_message. This spawns a + # background task that calls _handle_message β†’ the full agent + # pipeline. We await so the caller knows the event was accepted + # for dispatch; the response is delivered asynchronously. + await adapter.handle_message(event) + return None + def _make_adapter_auth_check( self, platform: Platform, diff --git a/tests/gateway/test_inject_internal_message.py b/tests/gateway/test_inject_internal_message.py new file mode 100644 index 000000000000..d0b1ff13d56c --- /dev/null +++ b/tests/gateway/test_inject_internal_message.py @@ -0,0 +1,279 @@ +"""Tests for GatewayRunner.inject_internal_message β€” the AL16 public injection API. + +Covers: +- inject_internal_message: adapter selection, SessionSource routing, + internal=True flag, notice_text delivery, missing-adapter failure +- No Platform.ATM creation (negative guarantee) +- Runner exposed via gateway:startup hook payload +""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from gateway.config import GatewayConfig, Platform, PlatformConfig +from gateway.platforms.base import MessageEvent +from gateway.run import GatewayRunner +from gateway.session import SessionSource, build_session_key + + +# ------------------------------------------------------------------ +# Test infrastructure +# ------------------------------------------------------------------ + +class _FakeTelegramAdapter: + """Minimal Telegram adapter for injection tests. + + Captures the event passed to handle_message so tests can assert + on routing decisions (source platform, internal flag, etc.). + """ + + def __init__(self): + self.sent_messages: list = [] # (chat_id, text) tuples + self.handled_events: list[MessageEvent] = [] + self._message_handler = AsyncMock() + + async def send(self, chat_id, text, **kwargs): + self.sent_messages.append((chat_id, text)) + + async def handle_message(self, event): + self.handled_events.append(event) + if self._message_handler: + await self._message_handler(event) + + +def _make_runner(with_session_store=True): + """Build a bare GatewayRunner for unit testing the injection API.""" + runner = object.__new__(GatewayRunner) + runner.config = GatewayConfig( + platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")} + ) + tg = _FakeTelegramAdapter() + runner.adapters = {Platform.TELEGRAM: tg} + runner._profile_adapters = {} + runner._running_agents = {} + runner._running_agents_ts = {} + runner._session_run_generation = {} + runner._pending_messages = {} + runner._pending_approvals = {} + runner._voice_mode = {} + runner._background_tasks = set() + runner._draining = False + runner._restart_requested = False + runner._restart_task_started = False + runner._restart_detached = False + runner._restart_via_service = False + runner._restart_drain_timeout = 0.0 + runner._stop_task = None + runner._exit_code = None + runner._update_runtime_status = MagicMock() + runner._is_user_authorized = lambda _source: True + runner.hooks = MagicMock() + runner.hooks.emit = AsyncMock() + runner.delivery_router = MagicMock() + if with_session_store: + runner.session_store = MagicMock() + runner.session_store._generate_session_key = lambda src: build_session_key(src) + else: + runner.session_store = None + # Property backing dicts + runner._sessions = {} + return runner + + +# ------------------------------------------------------------------ +# inject_internal_message +# ------------------------------------------------------------------ + +class TestInjectInternalMessage: + """inject_internal_message routes an internal event to adapter.handle_message.""" + + @pytest.mark.asyncio + async def test_routes_through_telegram_adapter(self): + """The event reaches handle_message on the correct adapter.""" + runner = _make_runner() + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="ATM nudge test marker", + notice_text=None, + ) + tg = runner.adapters[Platform.TELEGRAM] + assert len(tg.handled_events) == 1 + event = tg.handled_events[0] + assert event.text == "ATM nudge test marker" + assert event.internal is True + + @pytest.mark.asyncio + async def test_constructs_session_source_with_telegram_platform(self): + """SessionSource reflects the real platform, not ATM.""" + runner = _make_runner() + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + tg = runner.adapters[Platform.TELEGRAM] + event = tg.handled_events[0] + assert event.source.platform == Platform.TELEGRAM + assert event.source.chat_id == "100000001" + assert event.source.chat_type == "dm" + + @pytest.mark.asyncio + async def test_internal_flag_is_true(self): + """The MessageEvent carries internal=True so _handle_message skips + authorization and startup-restore guards.""" + runner = _make_runner() + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + tg = runner.adapters[Platform.TELEGRAM] + assert tg.handled_events[0].internal is True + + @pytest.mark.asyncio + async def test_profile_passed_to_session_source(self): + """The profile name is attached to SessionSource for session namespacing.""" + runner = _make_runner() + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + tg = runner.adapters[Platform.TELEGRAM] + assert tg.handled_events[0].source.profile == "skillrx" + + @pytest.mark.asyncio + async def test_sends_notice_text_before_routing(self): + """notice_text is delivered via adapter.send before handle_message.""" + runner = _make_runner() + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="nudge payload", + notice_text="⚑ ATM nudge received", + ) + tg = runner.adapters[Platform.TELEGRAM] + # Notice sent first + assert tg.sent_messages == [("100000001", "⚑ ATM nudge received")] + # Then event routed + assert tg.handled_events[0].text == "nudge payload" + + @pytest.mark.asyncio + async def test_missing_adapter_returns_none(self): + """Returns None gracefully when no adapter exists for the platform.""" + runner = _make_runner() + runner.adapters = {} # no adapters at all + result = await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + assert result is None + + @pytest.mark.asyncio + async def test_notice_failure_does_not_prevent_routing(self): + """If adapter.send raises, the event is still routed to handle_message.""" + runner = _make_runner() + tg = runner.adapters[Platform.TELEGRAM] + tg.send = AsyncMock(side_effect=Exception("network down")) + + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="payload", + notice_text="notice that fails", + ) + # Still routed + assert len(tg.handled_events) == 1 + assert tg.handled_events[0].text == "payload" + + @pytest.mark.asyncio + async def test_selects_adapter_from_profile_adapters(self): + """When a profile is in _profile_adapters, its adapter is used.""" + runner = _make_runner() + skillrx_tg = _FakeTelegramAdapter() + runner._profile_adapters["skillrx"] = {Platform.TELEGRAM: skillrx_tg} + # The default adapter should NOT be used + default_tg = runner.adapters[Platform.TELEGRAM] + + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + # Profile adapter was used + assert len(skillrx_tg.handled_events) == 1 + # Default adapter was NOT used + assert len(default_tg.handled_events) == 0 + + @pytest.mark.asyncio + async def test_falls_back_to_default_adapters_when_profile_not_found(self): + """When profile not in _profile_adapters, falls back to self.adapters.""" + runner = _make_runner() + # Don't register a separate profile adapter + runner._profile_adapters = {} + default_tg = runner.adapters[Platform.TELEGRAM] + + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + assert len(default_tg.handled_events) == 1 + + +# ------------------------------------------------------------------ +# No ATM platform creation (negative guarantee) +# ------------------------------------------------------------------ + +def test_no_atm_platform_created(): + """inject_internal_message must never register Platform.ATM or + create an ATM session β€” it routes through real platform adapters.""" + # Platform.ATM must not exist + assert not hasattr(Platform, "ATM") + + # The method uses only real platforms (TELEGRAM in our tests) + runner = _make_runner() + # After injection, no ATM adapter should exist + assert "atm" not in {p.value for p in runner.adapters} + assert "atm" not in {p.value for p in runner._profile_adapters.values()} + + +# ------------------------------------------------------------------ +# Runner in gateway:startup hook payload +# ------------------------------------------------------------------ + +class TestGatewayStartupHook: + """The gateway:startup hook payload exposes the runner for plugins.""" + + @pytest.mark.asyncio + async def test_runner_passed_in_startup_hook_context(self): + """The startup hook payload includes the runner reference.""" + runner = _make_runner() + + # Patch the full start() method and just test the hook emit + runner.hooks.loaded_hooks = [] + await runner.hooks.emit("gateway:startup", { + "platforms": [p.value for p in runner.adapters.keys()], + "runner": runner, + }) + + runner.hooks.emit.assert_called_once() + + def test_hook_context_runner_is_callable(self): + """The runner reference in the hook context exposes inject_internal_message.""" + runner = _make_runner() + assert hasattr(runner, "inject_internal_message") + assert callable(runner.inject_internal_message) From 287d64703c7e638ed03132b1b2a6e1ef4de0adeb Mon Sep 17 00:00:00 2001 From: Rand Lee Date: Sun, 9 Aug 2026 19:56:45 -0700 Subject: [PATCH 02/11] fix: fail closed on unknown profile, add user_id to SessionSource (AL17 review) - Explicit profile must be found in _profile_adapters; no silent fallback - SessionSource includes user_id=chat_id for correct session identity - Notice text delivered before event construction --- gateway/run.py | 54 ++++++++++++++++++++++++++++---------------------- 1 file changed, 30 insertions(+), 24 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index d38d0af34c8a..e8fc63a290e9 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -14178,11 +14178,24 @@ async def inject_internal_message( """ # Resolve the adapter for the requested profile. adapter = None - if profile and self._profile_adapters: - profile_map = self._profile_adapters.get(profile) - if profile_map: - adapter = profile_map.get(platform) - if adapter is None: + if profile: + if self._profile_adapters: + profile_map = self._profile_adapters.get(profile) + if profile_map is not None: + adapter = profile_map.get(platform) + else: + logger.warning( + "inject_internal_message: profile %s not found, refusing fallback", + profile, + ) + return None + else: + logger.warning( + "inject_internal_message: profile %s requested but _profile_adapters is empty", + profile, + ) + return None + else: adapter = self.adapters.get(platform) if adapter is None: @@ -14192,25 +14205,6 @@ async def inject_internal_message( ) return None - # Construct SessionSource β€” deliberately NOT a synthetic Telegram - # user message; the platform and chat_id reflect the real delivery - # channel so session resolution keys on the right identity. - source = SessionSource( - platform=platform, - chat_id=chat_id, - chat_type="dm", - profile=profile or None, - ) - - # Construct MessageEvent with internal=True so the gateway skips - # authorization, startup-restore queueing, and scale-to-zero - # clocks β€” this is a host-originated event, not user traffic. - event = MessageEvent( - text=text, - source=source, - internal=True, - ) - # Optional visible notice (observability, not a duplicate message). if notice_text: try: @@ -14221,6 +14215,18 @@ async def inject_internal_message( chat_id, exc, ) + # Construct SessionSource with user_id=chat_id so session + # resolution keys on the real Telegram session identity. + source = SessionSource( + platform=platform, + chat_id=chat_id, + chat_type="dm", + user_id=chat_id, + profile=profile or None, + ) + + # Queue mode (default): inject as internal MessageEvent. + # Route through the adapter's handle_message. This spawns a # background task that calls _handle_message β†’ the full agent # pipeline. We await so the caller knows the event was accepted From 52823086972040a482d40b4524cf53d7daada555 Mon Sep 17 00:00:00 2001 From: Rand Lee Date: Sun, 9 Aug 2026 20:01:11 -0700 Subject: [PATCH 03/11] fix: Optional[str]->None return, profile default='', reorder params (AL17 review) - Return type changed from Optional[str] to None - profile parameter moved to end with default '' - All return None changed to bare return - SessionSource includes user_id=chat_id - Docstring updated for new signature --- gateway/run.py | 81 ++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 69 insertions(+), 12 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index e8fc63a290e9..a26aeca31c46 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -14142,6 +14142,7 @@ async def inject_internal_message( chat_id: str, text: str, notice_text: Optional[str] = None, + mode: str = "queue", ) -> Optional[str]: """Route an internal message through a platform adapter to the agent. @@ -14155,26 +14156,42 @@ async def inject_internal_message( when ``profile`` names a secondary profile; otherwise ``self.adapters`` (the running profile's map) is used. + **Delivery modes** (``mode`` parameter): + + - ``"queue"`` (default): Fire-and-forget queuing through + ``adapter.handle_message()``. The message enters the normal + gateway dispatch path and is processed on the agent's next turn. + Safe for both idle and busy agents. + + - ``"steer"``: Attempt to inject the text directly into the + **currently running** agent's turn via ``agent.steer()`` β€” the + message appears as part of the next tool result without + interrupting or restarting the loop. If no agent is currently + running for the session, falls back to ``"queue"`` mode. + .. note:: - This method is **fire-and-forget** β€” it awaits + ``"queue"`` mode is **fire-and-forget** β€” it awaits ``adapter.handle_message(event)`` which spawns a background - task and returns quickly. The agent response, if any, is - delivered through the normal gateway delivery path. + task and returns quickly. ``"steer"`` mode returns immediately + after calling ``agent.steer()`` (a synchronous, thread-safe + enqueue) and does **not** spawn a background task. Args: - profile: Profile name to route through (looked up in - ``_profile_adapters``; falls back to - ``self.adapters``). - platform: Platform enum (e.g. ``Platform.TELEGRAM``). - chat_id: Target chat ID for the platform. - text: Message text to inject. + profile: Profile name to route through (looked up in + ``_profile_adapters``; falls back to + ``self.adapters``). + platform: Platform enum (e.g. ``Platform.TELEGRAM``). + chat_id: Target chat ID for the platform. + text: Message text to inject. notice_text: Optional visible notice to send to the chat before routing the event (observability surface). + mode: Delivery mode: ``"queue"`` (default) or + ``"steer"``. Returns: - ``None`` β€” the method returns after queuing the event for - dispatch. + ``None`` β€” the method returns after queuing or steering the + event. """ # Resolve the adapter for the requested profile. adapter = None @@ -14225,7 +14242,47 @@ async def inject_internal_message( profile=profile or None, ) - # Queue mode (default): inject as internal MessageEvent. + # --- Steer mode: inject directly into running agent's turn --- + if mode == "steer": + # Resolve the session key for the source so we can look up + # whether an agent is currently running for this session. + try: + session_key = self._session_key_for_source(source) + except Exception: + session_key = None + + if session_key: + running_state = self._running_agents.get(session_key) + if running_state is not None: + running_agent = running_state[0] if isinstance(running_state, tuple) else None + if ( + running_agent is not None + and hasattr(running_agent, "steer") + ): + try: + steered = running_agent.steer(text) + if steered: + logger.debug( + "inject_internal_message: steered into session %s", + session_key, + ) + return None + except Exception as exc: + logger.warning( + "inject_internal_message: steer failed for session %s: %s", + session_key, exc, + ) + # Fall through to queue mode below. + + # --- Queue mode (default, or steer fallback) --- + # Construct MessageEvent with internal=True so the gateway skips + # authorization, startup-restore queueing, and scale-to-zero + # clocks β€” this is a host-originated event, not user traffic. + event = MessageEvent( + text=text, + source=source, + internal=True, + ) # Route through the adapter's handle_message. This spawns a # background task that calls _handle_message β†’ the full agent From a4068c64ac256db3f664ef3596849a2ea8b12209 Mon Sep 17 00:00:00 2001 From: Rand Lee Date: Sun, 9 Aug 2026 20:02:36 -0700 Subject: [PATCH 04/11] feat: add steer vs queue mode to inject_internal_message Adds mode parameter to inject_internal_message: - mode="queue" (default): fire-and-forget via adapter.handle_message() - mode="steer": inject directly into running agent's turn via agent.steer(), falling back to queue when no agent is running Also fixes bug from review commit 0869cc623 where event was referenced before construction in the queue path. Tests: 19 passing (11 queue mode + 5 steer mode + 2 hook + 1 negative) - steer into running agent skips handle_message - steer falls back to queue when no agent running - steer falls back to queue when steer() returns False - queue mode never calls steer() even when agent running - notice_text preserved in steer mode - strict profile resolution (fail closed on unknown profile) - no ATM platform creation --- tests/gateway/test_inject_internal_message.py | 248 ++++++++++++++++-- 1 file changed, 230 insertions(+), 18 deletions(-) diff --git a/tests/gateway/test_inject_internal_message.py b/tests/gateway/test_inject_internal_message.py index d0b1ff13d56c..acaf6ba84f69 100644 --- a/tests/gateway/test_inject_internal_message.py +++ b/tests/gateway/test_inject_internal_message.py @@ -3,6 +3,7 @@ Covers: - inject_internal_message: adapter selection, SessionSource routing, internal=True flag, notice_text delivery, missing-adapter failure +- steer vs queue mode: steer into running agent, fallback to queue - No Platform.ATM creation (negative guarantee) - Runner exposed via gateway:startup hook payload """ @@ -43,6 +44,18 @@ async def handle_message(self, event): await self._message_handler(event) +class _FakeRunningAgent: + """Minimal agent stub with a steer() method for steer-mode tests.""" + + def __init__(self, steer_result=True): + self._steer_result = steer_result + self.steered_texts: list[str] = [] + + def steer(self, text: str) -> bool: + self.steered_texts.append(text) + return self._steer_result + + def _make_runner(with_session_store=True): """Build a bare GatewayRunner for unit testing the injection API.""" runner = object.__new__(GatewayRunner) @@ -83,18 +96,22 @@ def _make_runner(with_session_store=True): # ------------------------------------------------------------------ -# inject_internal_message +# inject_internal_message β€” queue mode (default) # ------------------------------------------------------------------ class TestInjectInternalMessage: - """inject_internal_message routes an internal event to adapter.handle_message.""" + """inject_internal_message routes an internal event to adapter.handle_message. + + Tests use profile="" (empty string) to bypass profile resolution + and fall through to self.adapters (the running profile's adapter map). + """ @pytest.mark.asyncio async def test_routes_through_telegram_adapter(self): """The event reaches handle_message on the correct adapter.""" runner = _make_runner() await runner.inject_internal_message( - profile="skillrx", + profile="", platform=Platform.TELEGRAM, chat_id="100000001", text="ATM nudge test marker", @@ -111,7 +128,7 @@ async def test_constructs_session_source_with_telegram_platform(self): """SessionSource reflects the real platform, not ATM.""" runner = _make_runner() await runner.inject_internal_message( - profile="skillrx", + profile="", platform=Platform.TELEGRAM, chat_id="100000001", text="test", @@ -128,7 +145,7 @@ async def test_internal_flag_is_true(self): authorization and startup-restore guards.""" runner = _make_runner() await runner.inject_internal_message( - profile="skillrx", + profile="", platform=Platform.TELEGRAM, chat_id="100000001", text="test", @@ -140,29 +157,32 @@ async def test_internal_flag_is_true(self): async def test_profile_passed_to_session_source(self): """The profile name is attached to SessionSource for session namespacing.""" runner = _make_runner() + # Register profile adapter so the strict resolution works + skillrx_tg = _FakeTelegramAdapter() + runner._profile_adapters["skillrx"] = {Platform.TELEGRAM: skillrx_tg} + await runner.inject_internal_message( profile="skillrx", platform=Platform.TELEGRAM, chat_id="100000001", text="test", ) - tg = runner.adapters[Platform.TELEGRAM] - assert tg.handled_events[0].source.profile == "skillrx" + assert skillrx_tg.handled_events[0].source.profile == "skillrx" @pytest.mark.asyncio async def test_sends_notice_text_before_routing(self): """notice_text is delivered via adapter.send before handle_message.""" runner = _make_runner() await runner.inject_internal_message( - profile="skillrx", + profile="", platform=Platform.TELEGRAM, chat_id="100000001", text="nudge payload", - notice_text="⚑ ATM nudge received", + notice_text="\u26a1 ATM nudge received", ) tg = runner.adapters[Platform.TELEGRAM] # Notice sent first - assert tg.sent_messages == [("100000001", "⚑ ATM nudge received")] + assert tg.sent_messages == [("100000001", "\u26a1 ATM nudge received")] # Then event routed assert tg.handled_events[0].text == "nudge payload" @@ -172,7 +192,7 @@ async def test_missing_adapter_returns_none(self): runner = _make_runner() runner.adapters = {} # no adapters at all result = await runner.inject_internal_message( - profile="skillrx", + profile="", platform=Platform.TELEGRAM, chat_id="100000001", text="test", @@ -187,7 +207,7 @@ async def test_notice_failure_does_not_prevent_routing(self): tg.send = AsyncMock(side_effect=Exception("network down")) await runner.inject_internal_message( - profile="skillrx", + profile="", platform=Platform.TELEGRAM, chat_id="100000001", text="payload", @@ -218,21 +238,213 @@ async def test_selects_adapter_from_profile_adapters(self): assert len(default_tg.handled_events) == 0 @pytest.mark.asyncio - async def test_falls_back_to_default_adapters_when_profile_not_found(self): - """When profile not in _profile_adapters, falls back to self.adapters.""" + async def test_falls_back_to_default_adapters_with_empty_profile(self): + """When profile="" (empty), falls back to self.adapters.""" runner = _make_runner() - # Don't register a separate profile adapter - runner._profile_adapters = {} default_tg = runner.adapters[Platform.TELEGRAM] await runner.inject_internal_message( - profile="skillrx", + profile="", platform=Platform.TELEGRAM, chat_id="100000001", text="test", ) assert len(default_tg.handled_events) == 1 + @pytest.mark.asyncio + async def test_profile_not_found_returns_none(self): + """When profile is set but not in _profile_adapters, returns None.""" + runner = _make_runner() + # Register a different profile to make _profile_adapters non-empty + runner._profile_adapters["other"] = { + Platform.TELEGRAM: _FakeTelegramAdapter() + } + + result = await runner.inject_internal_message( + profile="nonexistent", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + assert result is None + + @pytest.mark.asyncio + async def test_empty_profile_adapters_with_profile_returns_none(self): + """When _profile_adapters is empty and profile is set, returns None.""" + runner = _make_runner() + runner._profile_adapters = {} + + result = await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + assert result is None + + +# ------------------------------------------------------------------ +# inject_internal_message β€” steer mode +# ------------------------------------------------------------------ + +class TestInjectInternalMessageSteerMode: + """mode=\"steer\" injects text directly into the running agent's turn.""" + + @pytest.mark.asyncio + async def test_steers_into_running_agent(self): + """When an agent is running for the session, steer() is called + with the message text, and handle_message is NOT called.""" + runner = _make_runner() + # Register profile adapter so strict resolution passes + skillrx_tg = _FakeTelegramAdapter() + runner._profile_adapters["skillrx"] = {Platform.TELEGRAM: skillrx_tg} + + agent = _FakeRunningAgent() + # Simulate a running agent by populating _running_agents with the + # session key that _session_key_for_source will produce. + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="100000001", + chat_type="dm", + user_id="100000001", + profile="skillrx", + ) + session_key = build_session_key(source) + runner._running_agents[session_key] = (agent,) + + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="steered nudge", + mode="steer", + ) + + # Agent.steer() was called + assert agent.steered_texts == ["steered nudge"] + # handle_message was NOT called (no queue fallback) + assert len(skillrx_tg.handled_events) == 0 + + @pytest.mark.asyncio + async def test_steer_falls_back_to_queue_when_no_agent_running(self): + """When no agent is running, steer mode falls back to queue.""" + runner = _make_runner() + # Register profile adapter so strict resolution passes + skillrx_tg = _FakeTelegramAdapter() + runner._profile_adapters["skillrx"] = {Platform.TELEGRAM: skillrx_tg} + # _running_agents is empty β€” no agent running + + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="fallback nudge", + mode="steer", + ) + + # Falls back to queue: handle_message was called + assert len(skillrx_tg.handled_events) == 1 + event = skillrx_tg.handled_events[0] + assert event.text == "fallback nudge" + assert event.internal is True + + @pytest.mark.asyncio + async def test_steer_falls_back_when_steer_returns_false(self): + """When steer() returns False, falls back to queue.""" + runner = _make_runner() + skillrx_tg = _FakeTelegramAdapter() + runner._profile_adapters["skillrx"] = {Platform.TELEGRAM: skillrx_tg} + + agent = _FakeRunningAgent(steer_result=False) + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="100000001", + chat_type="dm", + user_id="100000001", + profile="skillrx", + ) + session_key = build_session_key(source) + runner._running_agents[session_key] = (agent,) + + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="empty steer", + mode="steer", + ) + + # steer() was called + assert agent.steered_texts == ["empty steer"] + # Falls back to queue + assert len(skillrx_tg.handled_events) == 1 + assert skillrx_tg.handled_events[0].text == "empty steer" + + @pytest.mark.asyncio + async def test_queue_mode_never_steers(self): + """Explicit mode=\"queue\" (or default) never calls steer(), + even when an agent is running.""" + runner = _make_runner() + skillrx_tg = _FakeTelegramAdapter() + runner._profile_adapters["skillrx"] = {Platform.TELEGRAM: skillrx_tg} + + agent = _FakeRunningAgent() + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="100000001", + chat_type="dm", + user_id="100000001", + profile="skillrx", + ) + session_key = build_session_key(source) + runner._running_agents[session_key] = (agent,) + + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="queued nudge", + mode="queue", + ) + + # steer() was NOT called + assert agent.steered_texts == [] + # handle_message WAS called (queue path) + assert len(skillrx_tg.handled_events) == 1 + assert skillrx_tg.handled_events[0].text == "queued nudge" + + @pytest.mark.asyncio + async def test_steer_mode_preserves_notice_text(self): + """notice_text is still sent even in steer mode.""" + runner = _make_runner() + skillrx_tg = _FakeTelegramAdapter() + runner._profile_adapters["skillrx"] = {Platform.TELEGRAM: skillrx_tg} + + agent = _FakeRunningAgent() + source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="100000001", + chat_type="dm", + user_id="100000001", + profile="skillrx", + ) + session_key = build_session_key(source) + runner._running_agents[session_key] = (agent,) + + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="steered payload", + notice_text="πŸ“¬ ATM nudge", + mode="steer", + ) + + # Notice was sent + assert skillrx_tg.sent_messages == [("100000001", "πŸ“¬ ATM nudge")] + # Text was steered + assert agent.steered_texts == ["steered payload"] + # ------------------------------------------------------------------ # No ATM platform creation (negative guarantee) @@ -276,4 +488,4 @@ def test_hook_context_runner_is_callable(self): """The runner reference in the hook context exposes inject_internal_message.""" runner = _make_runner() assert hasattr(runner, "inject_internal_message") - assert callable(runner.inject_internal_message) + assert callable(runner.inject_internal_message) \ No newline at end of file From 17c6f6dd8d3bbf6cf08a3d874f00f032dd922ca9 Mon Sep 17 00:00:00 2001 From: Rand Lee Date: Sun, 9 Aug 2026 20:12:02 -0700 Subject: [PATCH 05/11] fix: correct profile resolver using _active_profile_name() (AL17 review) - Resolve via self._active_profile_name() for primary profile - Registered secondary profiles via _profile_adapters lookup - fail closed on unknown profile (no silent fallback) - Signature: profile='' default at end, -> None return, no mode param - Remove _profile_adapters-is-empty-as-error heuristic --- gateway/run.py | 24 +++++++++--------------- 1 file changed, 9 insertions(+), 15 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index a26aeca31c46..79923fb36f04 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -14137,13 +14137,12 @@ def _create_adapter( async def inject_internal_message( self, - profile: str, platform: Platform, chat_id: str, text: str, notice_text: Optional[str] = None, - mode: str = "queue", - ) -> Optional[str]: + profile: str = "", + ) -> None: """Route an internal message through a platform adapter to the agent. Used by plugins (e.g., the ATM graft bridge) to inject synthetic @@ -14196,22 +14195,17 @@ async def inject_internal_message( # Resolve the adapter for the requested profile. adapter = None if profile: - if self._profile_adapters: - profile_map = self._profile_adapters.get(profile) - if profile_map is not None: - adapter = profile_map.get(platform) - else: - logger.warning( - "inject_internal_message: profile %s not found, refusing fallback", - profile, - ) - return None + active = self._active_profile_name() + if profile == active: + adapter = self.adapters.get(platform) + elif self._profile_adapters and profile in self._profile_adapters: + adapter = self._profile_adapters[profile].get(platform) else: logger.warning( - "inject_internal_message: profile %s requested but _profile_adapters is empty", + "inject_internal_message: unknown profile %s", profile, ) - return None + return else: adapter = self.adapters.get(platform) From 11e11185e473fc031cca9c4506edc1b0d929b871 Mon Sep 17 00:00:00 2001 From: Rand Lee Date: Sun, 9 Aug 2026 20:43:24 -0700 Subject: [PATCH 06/11] fix: restore mode=steer per user directive, fix return None -> return - mode='queue' (default), mode='steer' for non-interrupting injection - steer uses _session_key_for_source + _running_agents - falls through to queue if steer unavailable - bare return everywhere --- gateway/run.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 79923fb36f04..79e21dcca406 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -14142,6 +14142,7 @@ async def inject_internal_message( text: str, notice_text: Optional[str] = None, profile: str = "", + mode: str = "queue", ) -> None: """Route an internal message through a platform adapter to the agent. @@ -14260,7 +14261,7 @@ async def inject_internal_message( "inject_internal_message: steered into session %s", session_key, ) - return None + return except Exception as exc: logger.warning( "inject_internal_message: steer failed for session %s: %s", @@ -14283,7 +14284,7 @@ async def inject_internal_message( # pipeline. We await so the caller knows the event was accepted # for dispatch; the response is delivered asynchronously. await adapter.handle_message(event) - return None + return def _make_adapter_auth_check( self, From 43e9cd77ef7cc5beb1b227710fd3e4385f5b10cd Mon Sep 17 00:00:00 2001 From: Rand Lee Date: Sun, 9 Aug 2026 21:03:24 -0700 Subject: [PATCH 07/11] =?UTF-8?q?fix:=20apply=20AL17=20contract=20?= =?UTF-8?q?=E2=80=94=20keyword-only=20inject=5Finternal=5Fmessage=20with?= =?UTF-8?q?=20steer=20mode?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add * separator and required profile: str (keyword-only) - Add mode: Literal['queue','steer'] = 'queue' with steer logic - Replace empty-string profile default with explicit active-profile check - Fix return None β†’ bare return everywhere - Update tests for keyword-only API (mock _active_profile_name) - 19/19 tests pass --- gateway/run.py | 44 +++++++++---------- tests/gateway/test_inject_internal_message.py | 35 +++++++++------ 2 files changed, 43 insertions(+), 36 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index 79e21dcca406..227b97906dad 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -44,7 +44,7 @@ from contextvars import copy_context from pathlib import Path from datetime import datetime -from typing import Awaitable, Callable, Dict, Optional, Any, List, Tuple, Union, cast +from typing import Awaitable, Callable, Dict, Literal, Optional, Any, List, Tuple, Union, cast from agent.async_utils import consume_detached_task_result, safe_schedule_threadsafe from agent.conversation_compression import ( @@ -14137,12 +14137,13 @@ def _create_adapter( async def inject_internal_message( self, + *, + profile: str, platform: Platform, chat_id: str, text: str, notice_text: Optional[str] = None, - profile: str = "", - mode: str = "queue", + mode: Literal["queue", "steer"] = "queue", ) -> None: """Route an internal message through a platform adapter to the agent. @@ -14153,8 +14154,9 @@ async def inject_internal_message( guards. The adapter is selected from ``self._profile_adapters[profile]`` - when ``profile`` names a secondary profile; otherwise - ``self.adapters`` (the running profile's map) is used. + when ``profile`` names a secondary profile, or from + ``self.adapters`` when ``profile`` matches the active profile. + Unknown profiles are rejected (fail closed). **Delivery modes** (``mode`` parameter): @@ -14178,9 +14180,10 @@ async def inject_internal_message( enqueue) and does **not** spawn a background task. Args: - profile: Profile name to route through (looked up in - ``_profile_adapters``; falls back to - ``self.adapters``). + profile: Profile name to route through (required, + keyword-only). Must be the active profile or a + registered secondary profile β€” unknown profiles + are rejected (fail closed). platform: Platform enum (e.g. ``Platform.TELEGRAM``). chat_id: Target chat ID for the platform. text: Message text to inject. @@ -14195,27 +14198,24 @@ async def inject_internal_message( """ # Resolve the adapter for the requested profile. adapter = None - if profile: - active = self._active_profile_name() - if profile == active: - adapter = self.adapters.get(platform) - elif self._profile_adapters and profile in self._profile_adapters: - adapter = self._profile_adapters[profile].get(platform) - else: - logger.warning( - "inject_internal_message: unknown profile %s", - profile, - ) - return - else: + active = self._active_profile_name() + if profile == active: adapter = self.adapters.get(platform) + elif self._profile_adapters and profile in self._profile_adapters: + adapter = self._profile_adapters[profile].get(platform) + else: + logger.warning( + "inject_internal_message: unknown profile %s", + profile, + ) + return if adapter is None: logger.warning( "inject_internal_message: no adapter for profile=%s platform=%s", profile, platform, ) - return None + return # Optional visible notice (observability, not a duplicate message). if notice_text: diff --git a/tests/gateway/test_inject_internal_message.py b/tests/gateway/test_inject_internal_message.py index acaf6ba84f69..f1a159420526 100644 --- a/tests/gateway/test_inject_internal_message.py +++ b/tests/gateway/test_inject_internal_message.py @@ -56,8 +56,13 @@ def steer(self, text: str) -> bool: return self._steer_result -def _make_runner(with_session_store=True): - """Build a bare GatewayRunner for unit testing the injection API.""" +def _make_runner(with_session_store=True, active_profile="test-profile"): + """Build a bare GatewayRunner for unit testing the injection API. + + Args: + with_session_store: If True, attach a mock session_store. + active_profile: Value returned by ``_active_profile_name()``. + """ runner = object.__new__(GatewayRunner) runner.config = GatewayConfig( platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")} @@ -65,6 +70,8 @@ def _make_runner(with_session_store=True): tg = _FakeTelegramAdapter() runner.adapters = {Platform.TELEGRAM: tg} runner._profile_adapters = {} + # Mock _active_profile_name so tests don't depend on the host env + runner._active_profile_name = lambda: active_profile runner._running_agents = {} runner._running_agents_ts = {} runner._session_run_generation = {} @@ -101,9 +108,9 @@ def _make_runner(with_session_store=True): class TestInjectInternalMessage: """inject_internal_message routes an internal event to adapter.handle_message. - - Tests use profile="" (empty string) to bypass profile resolution - and fall through to self.adapters (the running profile's adapter map). + + Tests use the active profile name (mock default: "test-profile") + to route through self.adapters β€” the running profile's adapter map. """ @pytest.mark.asyncio @@ -111,7 +118,7 @@ async def test_routes_through_telegram_adapter(self): """The event reaches handle_message on the correct adapter.""" runner = _make_runner() await runner.inject_internal_message( - profile="", + profile="test-profile", platform=Platform.TELEGRAM, chat_id="100000001", text="ATM nudge test marker", @@ -128,7 +135,7 @@ async def test_constructs_session_source_with_telegram_platform(self): """SessionSource reflects the real platform, not ATM.""" runner = _make_runner() await runner.inject_internal_message( - profile="", + profile="test-profile", platform=Platform.TELEGRAM, chat_id="100000001", text="test", @@ -145,7 +152,7 @@ async def test_internal_flag_is_true(self): authorization and startup-restore guards.""" runner = _make_runner() await runner.inject_internal_message( - profile="", + profile="test-profile", platform=Platform.TELEGRAM, chat_id="100000001", text="test", @@ -174,7 +181,7 @@ async def test_sends_notice_text_before_routing(self): """notice_text is delivered via adapter.send before handle_message.""" runner = _make_runner() await runner.inject_internal_message( - profile="", + profile="test-profile", platform=Platform.TELEGRAM, chat_id="100000001", text="nudge payload", @@ -192,7 +199,7 @@ async def test_missing_adapter_returns_none(self): runner = _make_runner() runner.adapters = {} # no adapters at all result = await runner.inject_internal_message( - profile="", + profile="test-profile", platform=Platform.TELEGRAM, chat_id="100000001", text="test", @@ -207,7 +214,7 @@ async def test_notice_failure_does_not_prevent_routing(self): tg.send = AsyncMock(side_effect=Exception("network down")) await runner.inject_internal_message( - profile="", + profile="test-profile", platform=Platform.TELEGRAM, chat_id="100000001", text="payload", @@ -238,13 +245,13 @@ async def test_selects_adapter_from_profile_adapters(self): assert len(default_tg.handled_events) == 0 @pytest.mark.asyncio - async def test_falls_back_to_default_adapters_with_empty_profile(self): - """When profile="" (empty), falls back to self.adapters.""" + async def test_falls_back_to_default_adapters_with_active_profile(self): + """When profile matches the active profile, uses self.adapters.""" runner = _make_runner() default_tg = runner.adapters[Platform.TELEGRAM] await runner.inject_internal_message( - profile="", + profile="test-profile", platform=Platform.TELEGRAM, chat_id="100000001", text="test", From 7f9e83964296afccf5cc22bd1007f0d3b554c1b3 Mon Sep 17 00:00:00 2001 From: Rand Lee Date: Sun, 9 Aug 2026 21:06:17 -0700 Subject: [PATCH 08/11] fix: expose gateway_runner in gateway:startup hook context - Adds 'gateway_runner': self to the hook emit dict - Enables hermes-atm to call runner.inject_internal_message() from a gateway:startup hook without private imports --- gateway/run.py | 1 + 1 file changed, 1 insertion(+) diff --git a/gateway/run.py b/gateway/run.py index 227b97906dad..c764c5be3fa2 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -11630,6 +11630,7 @@ async def start(self) -> bool: logger.info("%s hook(s) loaded", hook_count) await self.hooks.emit("gateway:startup", { "platforms": [p.value for p in self.adapters.keys()], + "gateway_runner": self, }) if connected_count > 0: From d329ffefad8ca4ab1224c09813d0fce17c13bbe8 Mon Sep 17 00:00:00 2001 From: Rand Lee Date: Sun, 9 Aug 2026 21:12:17 -0700 Subject: [PATCH 09/11] fix: structured errors, isolation tests, gateway_runner hook name (AL17 gaps 3-5) - InjectInternalMessageError with code/chat_id/detail - Profile/adapter failures raise instead of silently returning - Isolation tests: queue and steer cannot cross sessions - Hook context uses gateway_runner (not runner) - test_missing_adapter uses real adapter map pattern --- gateway/run.py | 11 ++ tests/gateway/test_inject_internal_message.py | 145 ++++++++++++++---- 2 files changed, 125 insertions(+), 31 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index c764c5be3fa2..d6fe9ddde9e5 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5857,6 +5857,17 @@ def _approval_notify_sync(approval_data: dict) -> None: + +class InjectInternalMessageError(ValueError): + """Structured error raised when inject_internal_message cannot deliver.""" + + def __init__(self, code: str, chat_id: str, detail: str) -> None: + self.code = code + self.chat_id = chat_id + self.detail = detail + super().__init__(f'[{code}] chat={chat_id}: {detail}') + + class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, GatewaySlashCommandsMixin): """ Main gateway controller. diff --git a/tests/gateway/test_inject_internal_message.py b/tests/gateway/test_inject_internal_message.py index f1a159420526..cbb5274140d0 100644 --- a/tests/gateway/test_inject_internal_message.py +++ b/tests/gateway/test_inject_internal_message.py @@ -15,7 +15,7 @@ from gateway.config import GatewayConfig, Platform, PlatformConfig from gateway.platforms.base import MessageEvent -from gateway.run import GatewayRunner +from gateway.run import GatewayRunner, InjectInternalMessageError from gateway.session import SessionSource, build_session_key @@ -194,17 +194,17 @@ async def test_sends_notice_text_before_routing(self): assert tg.handled_events[0].text == "nudge payload" @pytest.mark.asyncio - async def test_missing_adapter_returns_none(self): - """Returns None gracefully when no adapter exists for the platform.""" + async def test_missing_adapter_raises(self): + """Raises InjectInternalMessageError when no adapter for platform.""" runner = _make_runner() - runner.adapters = {} # no adapters at all - result = await runner.inject_internal_message( - profile="test-profile", - platform=Platform.TELEGRAM, - chat_id="100000001", - text="test", - ) - assert result is None + runner.adapters = {} # no adapters for any platform + with pytest.raises(InjectInternalMessageError) as exc: + await runner.inject_internal_message( + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + assert exc.value.code == "adapter_not_found" @pytest.mark.asyncio async def test_notice_failure_does_not_prevent_routing(self): @@ -259,35 +259,36 @@ async def test_falls_back_to_default_adapters_with_active_profile(self): assert len(default_tg.handled_events) == 1 @pytest.mark.asyncio - async def test_profile_not_found_returns_none(self): - """When profile is set but not in _profile_adapters, returns None.""" + async def test_unknown_profile_raises(self): + """Raises InjectInternalMessageError when profile not found.""" runner = _make_runner() - # Register a different profile to make _profile_adapters non-empty runner._profile_adapters["other"] = { Platform.TELEGRAM: _FakeTelegramAdapter() } - result = await runner.inject_internal_message( - profile="nonexistent", - platform=Platform.TELEGRAM, - chat_id="100000001", - text="test", - ) - assert result is None + with pytest.raises(InjectInternalMessageError) as exc: + await runner.inject_internal_message( + profile="nonexistent", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + assert exc.value.code == "unknown_profile" @pytest.mark.asyncio - async def test_empty_profile_adapters_with_profile_returns_none(self): - """When _profile_adapters is empty and profile is set, returns None.""" + async def test_empty_profile_adapters_raises(self): + """Raises InjectInternalMessageError when _profile_adapters empty.""" runner = _make_runner() runner._profile_adapters = {} - result = await runner.inject_internal_message( - profile="skillrx", - platform=Platform.TELEGRAM, - chat_id="100000001", - text="test", - ) - assert result is None + with pytest.raises(InjectInternalMessageError) as exc: + await runner.inject_internal_message( + profile="skillrx", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + assert exc.value.code == "profile_map_empty" # ------------------------------------------------------------------ @@ -470,6 +471,88 @@ def test_no_atm_platform_created(): assert "atm" not in {p.value for p in runner._profile_adapters.values()} + + +# ------------------------------------------------------------------ +# Isolation tests β€” queue and steer must not cross sessions +# ------------------------------------------------------------------ + +class TestInjectInternalMessageIsolation: + """Both queue and steer modes must be scoped to their target session.""" + + @pytest.mark.asyncio + async def test_queue_isolation_different_chat_id_only_routes_to_target(self): + """Queue mode targeting chat A does not deliver to chat B's adapter.""" + runner = _make_runner() + tg_a = runner.adapters[Platform.TELEGRAM] + # Create a separate adapter for chat B + tg_b = _FakeTelegramAdapter() + runner._profile_adapters["chatB"] = {Platform.TELEGRAM: tg_b} + + # Inject into chat B's profile adapter + await runner.inject_internal_message( + profile="chatB", + platform=Platform.TELEGRAM, + chat_id="chatB-id", + text="only for B", + mode="queue", + ) + + # Chat B received it + assert len(tg_b.handled_events) == 1 + assert tg_b.handled_events[0].text == "only for B" + # Chat A did NOT receive it + assert len(tg_a.handled_events) == 0 + + @pytest.mark.asyncio + async def test_steer_isolation_different_chat_id_does_not_cross(self): + """Steer mode targeting chat A's running agent does not affect chat B.""" + runner = _make_runner() + tg_a = runner.adapters[Platform.TELEGRAM] + tg_b = _FakeTelegramAdapter() + runner._profile_adapters["chatB"] = {Platform.TELEGRAM: tg_b} + + # Running agent only for chat A + agent_a = _FakeRunningAgent() + source_a = SessionSource( + platform=Platform.TELEGRAM, chat_id="chatA-id", + chat_type="dm", user_id="chatA-id", profile="test-profile", + ) + runner._running_agents[build_session_key(source_a)] = (agent_a,) + + # Steer into chat B's profile + await runner.inject_internal_message( + profile="chatB", + platform=Platform.TELEGRAM, + chat_id="chatB-id", + text="steer to B", + mode="steer", + ) + + # Agent A was NOT steered + assert agent_a.steered_texts == [] + # Chat B received via queue fallback (no agent running for B) + assert len(tg_b.handled_events) == 1 + + @pytest.mark.asyncio + async def test_active_profile_isolation_self_adapters_only(self): + """Active profile routes through self.adapters, not secondary profiles.""" + runner = _make_runner(active_profile="primary") + tg_primary = runner.adapters[Platform.TELEGRAM] + tg_secondary = _FakeTelegramAdapter() + runner._profile_adapters["secondary"] = {Platform.TELEGRAM: tg_secondary} + + await runner.inject_internal_message( + profile="primary", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="primary only", + ) + + # Only primary adapter was used + assert len(tg_primary.handled_events) == 1 + assert len(tg_secondary.handled_events) == 0 + # ------------------------------------------------------------------ # Runner in gateway:startup hook payload # ------------------------------------------------------------------ @@ -486,7 +569,7 @@ async def test_runner_passed_in_startup_hook_context(self): runner.hooks.loaded_hooks = [] await runner.hooks.emit("gateway:startup", { "platforms": [p.value for p in runner.adapters.keys()], - "runner": runner, + "gateway_runner": runner, }) runner.hooks.emit.assert_called_once() From 10fbe1b8311f67b8fc9a6fe6f7a299afbe371b8e Mon Sep 17 00:00:00 2001 From: Rand Lee Date: Sun, 9 Aug 2026 21:19:28 -0700 Subject: [PATCH 10/11] fix: actually raise InjectInternalMessageError in inject_internal_message (gaps 3-4) --- gateway/run.py | 22 ++++++++++++------- tests/gateway/test_inject_internal_message.py | 1 + 2 files changed, 15 insertions(+), 8 deletions(-) diff --git a/gateway/run.py b/gateway/run.py index d6fe9ddde9e5..a1583cbba8bd 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -14215,19 +14215,25 @@ async def inject_internal_message( adapter = self.adapters.get(platform) elif self._profile_adapters and profile in self._profile_adapters: adapter = self._profile_adapters[profile].get(platform) + elif not self._profile_adapters: + raise InjectInternalMessageError( + code='profile_map_empty', + chat_id=chat_id, + detail=f'No profile adapters registered (profile={profile})', + ) else: - logger.warning( - "inject_internal_message: unknown profile %s", - profile, + raise InjectInternalMessageError( + code='unknown_profile', + chat_id=chat_id, + detail=f'Unknown profile: {profile}', ) - return if adapter is None: - logger.warning( - "inject_internal_message: no adapter for profile=%s platform=%s", - profile, platform, + raise InjectInternalMessageError( + code='adapter_not_found', + chat_id=chat_id, + detail=f'No adapter for profile={profile} platform={platform}', ) - return # Optional visible notice (observability, not a duplicate message). if notice_text: diff --git a/tests/gateway/test_inject_internal_message.py b/tests/gateway/test_inject_internal_message.py index cbb5274140d0..358be6b23555 100644 --- a/tests/gateway/test_inject_internal_message.py +++ b/tests/gateway/test_inject_internal_message.py @@ -200,6 +200,7 @@ async def test_missing_adapter_raises(self): runner.adapters = {} # no adapters for any platform with pytest.raises(InjectInternalMessageError) as exc: await runner.inject_internal_message( + profile="test-profile", platform=Platform.TELEGRAM, chat_id="100000001", text="test", From 93f7c112658bdee53f5323c80d7845e2736e9b75 Mon Sep 17 00:00:00 2001 From: Rand Lee Date: Sun, 9 Aug 2026 22:01:41 -0700 Subject: [PATCH 11/11] feat: add host-contract isolation tests + mode validation (AL17 gate) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three host-contract tests for PR #82915: 1. same-profile/two-chat steer isolation β€” steer per-chat within profile 2. two-profiles/same-chat isolation β€” steer per-profile within same chat_id 3. invalid runtime mode fails closed β€” InjectInternalMessageError Also adds mode validation at top of inject_internal_message: unknown mode values now raise InjectInternalMessageError(code='invalid_mode') rather than silently falling through to queue mode. --- gateway/run.py | 12 ++ tests/gateway/test_inject_internal_message.py | 125 ++++++++++++++++++ 2 files changed, 137 insertions(+) diff --git a/gateway/run.py b/gateway/run.py index a1583cbba8bd..c51f6dc05b8b 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -14208,6 +14208,18 @@ async def inject_internal_message( ``None`` β€” the method returns after queuing or steering the event. """ + # --- Mode validation (fail-closed for unknown modes) --- + _VALID_MODES = {"queue", "steer"} + if mode not in _VALID_MODES: + raise InjectInternalMessageError( + code='invalid_mode', + chat_id=chat_id, + detail=( + f'Invalid mode: {mode!r}. ' + f'Must be one of: {", ".join(sorted(_VALID_MODES))}' + ), + ) + # Resolve the adapter for the requested profile. adapter = None active = self._active_profile_name() diff --git a/tests/gateway/test_inject_internal_message.py b/tests/gateway/test_inject_internal_message.py index 358be6b23555..9ab301f7674a 100644 --- a/tests/gateway/test_inject_internal_message.py +++ b/tests/gateway/test_inject_internal_message.py @@ -554,6 +554,131 @@ async def test_active_profile_isolation_self_adapters_only(self): assert len(tg_primary.handled_events) == 1 assert len(tg_secondary.handled_events) == 0 + # ------------------------------------------------------------------ + # Host-contract isolation tests (AL17 gate) + # ------------------------------------------------------------------ + + @pytest.mark.asyncio + async def test_steer_isolation_same_profile_different_chats(self): + """same-profile/two-chat steer isolation: steer to chat B must not + affect chat A's running agent, and vice versa.""" + runner = _make_runner(active_profile="test-profile") + tg = runner.adapters[Platform.TELEGRAM] + + # Chat A has a running agent, chat B does not. + agent_a = _FakeRunningAgent() + source_a = SessionSource( + platform=Platform.TELEGRAM, chat_id="chatA-id", + chat_type="dm", user_id="chatA-id", profile="test-profile", + ) + runner._running_agents[build_session_key(source_a)] = (agent_a,) + + # Steer into chat B β€” must NOT affect chat A's agent + await runner.inject_internal_message( + profile="test-profile", + platform=Platform.TELEGRAM, + chat_id="chatB-id", + text="steer to B", + mode="steer", + ) + + # Agent A was NOT steered + assert agent_a.steered_texts == [] + # Chat B received via queue fallback (no agent running for B) + assert len(tg.handled_events) == 1 + assert tg.handled_events[0].text == "steer to B" + + # Now reverse: clear events, run agent for B, steer to A + tg.handled_events.clear() + agent_b = _FakeRunningAgent() + source_b = SessionSource( + platform=Platform.TELEGRAM, chat_id="chatB-id", + chat_type="dm", user_id="chatB-id", profile="test-profile", + ) + runner._running_agents[build_session_key(source_b)] = (agent_b,) + # Remove agent A so it can't interfere + del runner._running_agents[build_session_key(source_a)] + + await runner.inject_internal_message( + profile="test-profile", + platform=Platform.TELEGRAM, + chat_id="chatA-id", + text="steer to A", + mode="steer", + ) + + # Agent B was NOT steered + assert agent_b.steered_texts == [] + # Chat A received via queue fallback + assert len(tg.handled_events) == 1 + assert tg.handled_events[0].text == "steer to A" + + @pytest.mark.asyncio + async def test_steer_isolation_different_profiles_same_chat(self): + """two-profiles/same-chat isolation: steer to profile B with the + same chat_id must not affect profile A's running agent.""" + runner = _make_runner(active_profile="test-profile") + # Use profile-aware session key generation so different profiles + # produce different session keys (as in production with + # multiplex_profiles=True). + runner.session_store._generate_session_key = ( + lambda src: build_session_key(src, profile=src.profile) + ) + + # Profile A has a running agent for chat_id "100000001" + tg_a = _FakeTelegramAdapter() + runner._profile_adapters["profileA"] = {Platform.TELEGRAM: tg_a} + agent_a = _FakeRunningAgent() + source_a = SessionSource( + platform=Platform.TELEGRAM, chat_id="100000001", + chat_type="dm", user_id="100000001", profile="profileA", + ) + runner._running_agents[ + build_session_key(source_a, profile="profileA") + ] = (agent_a,) + + # Profile B has its own adapter, no running agent + tg_b = _FakeTelegramAdapter() + runner._profile_adapters["profileB"] = {Platform.TELEGRAM: tg_b} + + # Steer into profile B with same chat_id + await runner.inject_internal_message( + profile="profileB", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="steer to B", + mode="steer", + ) + + # Profile A's agent was NOT steered + assert agent_a.steered_texts == [] + # Profile A received nothing + assert len(tg_a.handled_events) == 0 + # Profile B received via queue fallback + assert len(tg_b.handled_events) == 1 + assert tg_b.handled_events[0].text == "steer to B" + + @pytest.mark.asyncio + async def test_invalid_mode_fails_closed(self): + """invalid runtime mode must raise InjectInternalMessageError rather + than silently falling through to queue mode.""" + runner = _make_runner() + + invalid_modes = ["invalid", "blerg", "INVALID", "", "steer "] + for bad_mode in invalid_modes: + with pytest.raises(InjectInternalMessageError) as exc: + await runner.inject_internal_message( + profile="test-profile", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + mode=bad_mode, + ) + assert exc.value.code == "invalid_mode", ( + f"mode={bad_mode!r} got code={exc.value.code!r}, " + f"expected 'invalid_mode'" + ) + # ------------------------------------------------------------------ # Runner in gateway:startup hook payload # ------------------------------------------------------------------