diff --git a/agent/model_metadata.py b/agent/model_metadata.py index c3a0fc4b414a..7655bb2a9b87 100644 --- a/agent/model_metadata.py +++ b/agent/model_metadata.py @@ -2347,6 +2347,51 @@ def _query_anthropic_context_length(model: str, base_url: str, api_key: str) -> "gpt-5": 272_000, } +# Codex OAuth advertises 272K via /backend-api/codex/models for these +# families, but the backend actually ACCEPTS more (verified live Aug 16 2026 +# against chatgpt.com/backend-api/codex/responses: ~371K input tokens +# completed OK for gpt-5.6-sol/terra/luna and gpt-5.4; ~382K+ rejected with +# ``context_length_exceeded``; gpt-5.5 rejected 360K, so its 272K +# advertisement is real and it is NOT listed). 350K keeps ~22K margin under +# the observed ~372K enforcement. +# +# Applied ONLY when the resolved value (live probe or fallback table) is +# exactly the known-stale 272,000 advertisement — if OpenAI moves the +# advertised number in either direction (the gpt-5.6 family shifted +# 272K → 372K → 272K during July 2026), the catalog is trusted again and +# this table is inert. ``gpt-5.6`` is a FAMILY PREFIX (sol/terra/luna and +# dated snapshots; ``-pro`` slugs are not routable on Codex OAuth — the +# backend 400s them — so over-matching there is moot). ``gpt-5.4`` is EXACT: +# gpt-5.4-mini was probed and genuinely enforces 272K (rejected 360K), so +# prefix-matching the 5.4 family would over-report for mini. +_CODEX_OAUTH_VERIFIED_ABOVE_ADVERTISED_PREFIXES: Dict[str, int] = { + "gpt-5.6": 350_000, # sol / terra / luna — all three verified live +} +_CODEX_OAUTH_VERIFIED_ABOVE_ADVERTISED_EXACT: Dict[str, int] = { + "gpt-5.4": 350_000, # verified live; gpt-5.4-mini rejected 360K — excluded +} + +# The advertised value the verified-above table is allowed to override. +_CODEX_OAUTH_STALE_ADVERTISED_CTX = 272_000 + + +def _verified_codex_ctx_for_slug(model_bare: str) -> Optional[int]: + """Return the live-verified Codex cap for a slug, or ``None``. + + Exact slugs first, then family prefixes (````, ``-``, + ``.``) so dated snapshots of a verified family inherit the bump. + """ + slug = (model_bare or "").strip().lower() + if not slug: + return None + exact = _CODEX_OAUTH_VERIFIED_ABOVE_ADVERTISED_EXACT.get(slug) + if exact is not None: + return exact + for key, ctx in _CODEX_OAUTH_VERIFIED_ABOVE_ADVERTISED_PREFIXES.items(): + if slug == key or slug.startswith(key + "-") or slug.startswith(key + "."): + return ctx + return None + _codex_oauth_context_cache: Dict[str, Tuple[Dict[str, int], float]] = {} _CODEX_OAUTH_CONTEXT_CACHE_TTL = 3600 # 1 hour @@ -2474,16 +2519,33 @@ def _resolve_codex_oauth_context_length_with_source( if not model_bare: return None, "" + def _apply_verified_bump(ctx: int, source: str) -> Tuple[int, str]: + """Lift a known-stale 272K advertisement to the live-verified cap. + + Only fires when the resolved value is EXACTLY the stale 272,000 + advertisement for a slug we have probed above it (see + ``_verified_codex_ctx_for_slug``). Any other advertised value — + higher or lower — is trusted as a real server-side change. + """ + bumped = _verified_codex_ctx_for_slug(model_bare) + if bumped is not None and ctx == _CODEX_OAUTH_STALE_ADVERTISED_CTX: + logger.debug( + "Codex OAuth context for %s: advertised %d raised to " + "live-verified %d", model_bare, ctx, bumped, + ) + return bumped, source + return ctx, source + if access_token: live, fresh_probe = _fetch_codex_oauth_context_lengths_with_source(access_token) live_source = "live" if fresh_probe else "memory" if model_bare in live: - return live[model_bare], live_source + return _apply_verified_bump(live[model_bare], live_source) # Case-insensitive match in case casing drifts model_lower = model_bare.lower() for slug, ctx in live.items(): if slug.lower() == model_lower: - return ctx, live_source + return _apply_verified_bump(ctx, live_source) # Fallback: longest-key-first substring match over hardcoded defaults. model_lower = model_bare.lower() @@ -2491,7 +2553,7 @@ def _resolve_codex_oauth_context_length_with_source( _CODEX_OAUTH_CONTEXT_FALLBACK.items(), key=lambda x: len(x[0]), reverse=True ): if slug in model_lower: - return ctx, "fallback" + return _apply_verified_bump(ctx, "fallback") return None, "" diff --git a/docs/atm/FORK-MAINTENANCE.md b/docs/atm/FORK-MAINTENANCE.md new file mode 100644 index 000000000000..840cb6ec5faf --- /dev/null +++ b/docs/atm/FORK-MAINTENANCE.md @@ -0,0 +1,72 @@ +# Fork Maintenance — how randlee/hermes-agent tracks the upstream firehose + +Upstream (`NousResearch/hermes-agent`) lands 200–700 commits/day. This fork exists to +carry ONE thing on top of it: the ATM injection patch stack (see +`PATCH-REQUIREMENTS.md` in this directory — read it first; it is the contract). + +## Repo roles + +| Thing | Role | +|---|---| +| `main` | upstream/main + the ATM patch stack, advanced only by reviewed PR | +| `atm/stack` branch | the patch stack (3 code commits + this docs commit), rebased onto upstream daily; force-push allowed HERE, never on main | +| `sync/candidate-YYYYMMDD` branches | daily PR candidates produced by the cron | +| `runtime-*` tags | sources of built runtimes (see `~/.hermes/RUNTIME-PLAN.md` on the gateway host) | +| `~/Documents/forks/hermes-agent` (host) | integration WORKSPACE only — nothing executes from it (enforced by runtime-audit) | +| `~/.hermes/runtime//` (host) | immutable runtime installs; gateways run `runtime/current` | + +## Daily pipeline (2-level cron) + +**Level 1 — mechanical (no agent judgment):** `fork-sync-v2.sh` in a scratch clone: +1. fetch upstream; branch `sync/candidate-YYYYMMDD` from `upstream/main` +2. rebase `atm/stack` onto it (`git rebase`); a clean rebase proceeds, ANY conflict + → level 2 +3. fresh venv: `uv sync --frozen --no-dev --extra messaging`; run the seam contract + tests (`tests/gateway/test_inject_internal_message.py`, 26 expected) + hooks tests +4. green → pre-resolve the merge into `main`: because main and each candidate + carry different rebased copies of the stack, a raw candidate→main PR always + conflicts. The script builds the merge commit itself with the sanctioned + resolution — **candidate tree wins** (first established by loki in PR #7) — + verifies tree-hash equality with the candidate, pushes both branches, and + opens the PR from the pre-resolved merge branch (`sync/candidate-*-merge`). + The PR therefore arrives conflict-free; reviewers judge the candidate via + `git diff upstream/main..sync/candidate-*` (must be exactly the stack) +5. review chain: **contessa** (local qwen, free — does the context-intensive + work) reviews the diff-vs-upstream and test output — the diff must be exactly + the known patch stack, nothing more; then **alpha-prime** (qwen 3.7) approves + and merges routine PRs and signs off smoke tests, then advances the stack pointer. + AUTH NOTE: all agents share the `randlee` account, which also authors the PRs — + formal `gh pr review --approve` is therefore impossible (GitHub forbids + self-approval). The sanctioned path (established by loki, PR #7): post the + review verdict as a PR comment, then merge via owner bypass + (`gh pr merge --merge --admin`; enforce_admins is off). The 1-approval branch + protection stays as a guard against accidental non-admin pushes, not as a + working review gate + (`git push origin +:atm/stack`). **Loki** (frontier, expensive) is + NOT in the routine path — non-trivial PRs, reviewer disagreement, or anything + unexpected → level 2. + +**Level 2 — escalation (agent judgment):** triggered by rebase conflict, test +failure, or reviewer rejection. The escalation agent is **loki** (hermes-agent-atm +maintainer, frontier model; workspace `hendrix/loki/`, reachable via +`atm send loki`). Loki receives `PATCH-REQUIREMENTS.md` +and follows its "How to update the patch" procedure. Its output is an updated +`atm/stack` + a PR — never a direct push to main, never a force-push of main, never +a branch-protection change. If the contract can't be met, it stops and reports to +Rand with analysis. + +## Promotion (deliberate, not automatic) + +Merged main ≠ deployed. To deploy: tag, then on the gateway host +`make-runtime.sh --repo --ref --name --hermes-atm +--atm-graft `, canary one profile, flip `runtime/current`, rolling restart. +Rollback = flip the symlink back. Full procedure: `~/.hermes/RUNTIME-PLAN.md`. + +## History / lessons already learned + +- Merge-based daily syncs (the pre-2026-08-16 pipeline) accumulated conflict debt + and once ended with an agent force-pushing main and loosening branch protection. + Rebase-the-stack + PR + protected main is the replacement. Do not regress to it. +- The patch's only recurring conflict is the `gateway/run.py` import block (trivial). +- Goal state is patch size ZERO: if upstream ever ships a public injection API, + adapt hermes-atm to it and retire this stack. diff --git a/docs/atm/PATCH-REQUIREMENTS.md b/docs/atm/PATCH-REQUIREMENTS.md new file mode 100644 index 000000000000..b9b9fb9854e2 --- /dev/null +++ b/docs/atm/PATCH-REQUIREMENTS.md @@ -0,0 +1,98 @@ +# ATM Patch Requirements — the contract an updated patch MUST satisfy + +This document is the knowledge base for the escalation path of the fork-maintenance +pipeline: when the mechanical rebase of the ATM patch stack onto upstream fails (or +tests fail after it), an agent is given this document and expected to produce an +updated patch. It states WHAT the patch must provide and WHY, so the patch can be +re-derived even if upstream refactors everything it currently touches. + +## What the patch is + +`NousResearch/hermes-agent` has no public way for an external process to inject a +message into a running gateway session. The ATM patch adds exactly one seam: + +**`GatewayRunner.inject_internal_message(...)`** — a public, keyword-only async API +that the `hermes-atm` package (pip; source in atm-core `crates/hermes-atm`) calls to +deliver agent-team-mail nudges into a profile's chat session. Everything else in the +stack exists to support or test that seam. + +Current stack shape (3 commits + this doc commit): +1. `feat: expose public gateway injection seam` — the API + hook exposure + tests +2. `fix(gateway): notify visible internal-message notices` — notice delivery hardening +3. `test(gateway): cover soft visible-notice failure` + +## The contract (every item is load-bearing) + +1. **Public API surface** — `GatewayRunner` must expose: + ```python + async def inject_internal_message( + *, profile: str, platform: Platform, chat_id: str, text: str, + notice_text: Optional[str] = None, + mode: Literal["queue", "steer"] = "queue", + ) -> None + ``` + plus `InjectInternalMessageError(code, chat_id, detail)`. hermes-atm validates at + install time that `inject_internal_message` is callable on the runner and calls it + with exactly these keywords. Changing names/signature breaks every deployed + hermes-atm wheel — do not. + +2. **Hook exposure** — the `gateway:startup` hook context dict must contain + `"gateway_runner": self`. This is how hermes-atm's installed hook obtains the + runner without private imports. (Upstream may rename the emit site; the key in the + context dict must survive.) + +3. **Profile resolution, fail-closed** — resolve the target profile via the runner's + profile-adapter map (or the active profile). Unknown profile / empty adapter map + MUST raise `InjectInternalMessageError` — never fall through to a default chat. + Misrouted injection = message delivered to the wrong Telegram chat. Fail closed. + +4. **Modes** — `"queue"`: enqueue via the platform adapter's normal inbound-message + path (fire-and-forget). `"steer"`: if the profile's agent is mid-turn and exposes + a steer capability, inject into the running turn; otherwise fall back to queue. + Queue is the default and the only mode hermes-atm currently uses in production. + +5. **Visible notice is soft-fail** — when `notice_text` is provided, send it via + `adapter.send(chat_id, notice_text, metadata={"notify": True})` BEFORE routing the + event; check the result's `success` and log a warning (with the result error) on + failure; catch exceptions and log. A notice failure must NEVER prevent the main + event from routing. (This is commits 2–3 of the stack.) + +6. **Tests are the contract's enforcement** — + `tests/gateway/test_inject_internal_message.py` (26 tests: queue, steer, + isolation, error cases, notice soft-fail). The updated patch must keep all 26 + green. Only change a test when upstream semantics genuinely force it, and then + re-verify hermes-atm compatibility (its hook + runtime must still work — see + "Definition of done"). + +## Known conflict hotspots + +- `gateway/run.py` import block (upstream churns `datetime`/`typing` imports; ours + adds `Literal`, `Optional`). Union the imports — this is the most common conflict + and is always trivially resolvable. +- The `inject_internal_message` method body sits in `GatewayRunner` (a huge class + upstream refactors freely). If upstream moves adapter access or the event-routing + entry point (`adapter.handle_message` today), re-wire the seam to the new + internals while keeping the public surface identical. +- The `gateway:startup` emit site (search: `hooks.emit("gateway:startup"`). + +## How to update the patch (escalation procedure) + +1. Work in a scratch clone. NEVER in `~/Documents/forks/hermes-agent` directly, + never force-push `main`, never touch branch protection. +2. `git fetch upstream main`; start from `upstream/main`; attempt + `git rebase upstream/main` of the `atm/stack` branch. +3. Resolve conflicts per the contract above — the contract, not the old diff, is the + spec. If upstream now provides an equivalent public injection API, prefer + adapting hermes-atm to it and SHRINKING the patch (goal state: patch size zero). +4. Verify (all with Python 3.11, deps frozen from the repo's own `uv.lock`): + `uv sync --frozen --no-dev --extra messaging` then + `python -m pytest tests/gateway/test_inject_internal_message.py` → 26 passed, + plus `tests/gateway/test_hooks.py` and any test file the conflict touched. +5. Definition of done: tests green AND a live check that + `hermes_atm.HermesAtmRuntime.from_gateway_runner` accepts the runner (import + `hermes_atm`, construct against a stub runner exposing the API — the seam tests + include this shape) — then push the branch and open a PR to `main`; the review + pipeline (contessa → qwen reviewer) takes it from there. +6. If the contract itself cannot be satisfied (upstream removed a capability the + seam needs), STOP and escalate to Rand with a written analysis — do not ship a + behavioral compromise. diff --git a/gateway/run.py b/gateway/run.py index a4fcde2d499d..383a88b8028e 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -45,7 +45,7 @@ from contextvars import copy_context from pathlib import Path from datetime import datetime, timedelta, timezone -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 ( @@ -6408,6 +6408,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. @@ -12716,6 +12727,7 @@ async def _connect_one_startup(p, p_cfg, adp): 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: @@ -15407,6 +15419,188 @@ 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, + mode: Literal["queue", "steer"] = "queue", + ) -> None: + """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, or from + ``self.adapters`` when ``profile`` matches the active profile. + Unknown profiles are rejected (fail closed). + + **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:: + + ``"queue"`` mode is **fire-and-forget** — it awaits + ``adapter.handle_message(event)`` which spawns a background + 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 (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. + 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 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() + 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) + elif not self._profile_adapters: + raise InjectInternalMessageError( + code='profile_map_empty', + chat_id=chat_id, + detail=f'No profile adapters registered (profile={profile})', + ) + else: + raise InjectInternalMessageError( + code='unknown_profile', + chat_id=chat_id, + detail=f'Unknown profile: {profile}', + ) + + if adapter is None: + raise InjectInternalMessageError( + code='adapter_not_found', + chat_id=chat_id, + detail=f'No adapter for profile={profile} platform={platform}', + ) + + # Optional visible notice (observability, not a duplicate message). + if notice_text: + try: + notice_result = await adapter.send( + chat_id, + notice_text, + metadata={"notify": True}, + ) + if not getattr(notice_result, "success", False): + logger.warning( + "inject_internal_message: visible notice was not delivered: %s", + getattr(notice_result, "error", "unknown adapter failure"), + ) + except Exception as exc: + logger.warning( + "inject_internal_message: visible notice send raised: %s", + 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, + ) + + # --- 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 + 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 + # pipeline. We await so the caller knows the event was accepted + # for dispatch; the response is delivered asynchronously. + await adapter.handle_message(event) + return + def _make_adapter_auth_check( self, platform: Platform, diff --git a/tests/agent/test_model_metadata.py b/tests/agent/test_model_metadata.py index 3ab8dc05094a..6c026c34538e 100644 --- a/tests/agent/test_model_metadata.py +++ b/tests/agent/test_model_metadata.py @@ -401,12 +401,12 @@ def test_live_catalogue_cache_is_scoped_to_access_token(self): first_response = MagicMock() first_response.status_code = 200 first_response.json.return_value = { - "models": [{"slug": "gpt-5.6-terra", "context_window": 272_000}] + "models": [{"slug": "gpt-5.5", "context_window": 272_000}] } second_response = MagicMock() second_response.status_code = 200 second_response.json.return_value = { - "models": [{"slug": "gpt-5.6-terra", "context_window": 372_000}] + "models": [{"slug": "gpt-5.5", "context_window": 372_000}] } with patch( @@ -414,19 +414,19 @@ def test_live_catalogue_cache_is_scoped_to_access_token(self): side_effect=[first_response, second_response], ) as mock_get, patch("agent.model_metadata.save_context_length") as mock_save: first = get_model_context_length( - "gpt-5.6-terra", + "gpt-5.5", base_url="https://chatgpt.com/backend-api/codex", api_key="token-account-a", provider="openai-codex", ) first_again = get_model_context_length( - "gpt-5.6-terra", + "gpt-5.5", base_url="https://chatgpt.com/backend-api/codex", api_key="token-account-a", provider="openai-codex", ) second = get_model_context_length( - "gpt-5.6-terra", + "gpt-5.5", base_url="https://chatgpt.com/backend-api/codex", api_key="token-account-b", provider="openai-codex", @@ -478,7 +478,7 @@ def test_live_codex_context_replaces_stale_cache_in_both_directions( monkeypatch.setattr(mm, "_get_context_cache_path", lambda: cache_file) base_url = "https://chatgpt.com/backend-api/codex" - stale_key = f"gpt-5.6-terra@{base_url}" + stale_key = f"gpt-5.5@{base_url}" other_key = "other-model@https://api.openai.com/v1/" import yaml as _yaml cache_file.write_text(_yaml.dump({"context_lengths": { @@ -489,14 +489,14 @@ def test_live_codex_context_replaces_stale_cache_in_both_directions( fake_response = MagicMock() fake_response.status_code = 200 fake_response.json.return_value = { - "models": [{"slug": "gpt-5.6-terra", "context_window": live_context}] + "models": [{"slug": "gpt-5.5", "context_window": live_context}] } # Exercise real persistence here: this test verifies that a live value # replaces the stale on-disk entry. Failure-path tests below mock the # writer because they assert that fallback values are not persisted. with patch("agent.model_metadata.requests.get", return_value=fake_response) as mock_get: ctx = mm.get_model_context_length( - model="gpt-5.6-terra", + model="gpt-5.5", base_url=base_url, api_key="fake-token", provider="openai-codex", @@ -510,6 +510,103 @@ def test_live_codex_context_replaces_stale_cache_in_both_directions( assert remaining.get(stale_key) == live_context assert remaining.get(other_key) == 128_000 + @pytest.mark.parametrize( + "slug", + [ + "gpt-5.6-sol", + "gpt-5.6-terra", + "gpt-5.6-luna", + "gpt-5.6-sol-2026-07-09", # dated snapshot via gpt-5.6 family prefix + "gpt-5.4", + ], + ) + def test_stale_272k_advertisement_bumped_to_live_verified_350k(self, slug): + """Codex advertises 272K for these slugs but the backend accepts ~372K + (verified live Aug 2026); the resolver lifts exactly-272K to 350K.""" + from agent.model_metadata import get_model_context_length + + fake_response = MagicMock() + fake_response.status_code = 200 + fake_response.json.return_value = { + "models": [{"slug": slug, "context_window": 272_000}] + } + with patch("agent.model_metadata.requests.get", return_value=fake_response), \ + patch("agent.model_metadata.get_cached_context_length", return_value=None), \ + patch("agent.model_metadata.save_context_length"): + ctx = get_model_context_length( + model=slug, + base_url="https://chatgpt.com/backend-api/codex", + api_key="fake-token", + provider="openai-codex", + ) + assert ctx == 350_000 + + def test_non_272k_advertisement_is_trusted_verbatim(self): + """Any advertised value other than the known-stale 272,000 — higher or + lower — is a real server-side change and must NOT be overridden.""" + from agent.model_metadata import get_model_context_length + + for advertised in (372_000, 200_000, 1_050_000): + fake_response = MagicMock() + fake_response.status_code = 200 + fake_response.json.return_value = { + "models": [{"slug": "gpt-5.6-sol", "context_window": advertised}] + } + import agent.model_metadata as mm + mm._codex_oauth_context_cache = {} + with patch("agent.model_metadata.requests.get", return_value=fake_response), \ + patch("agent.model_metadata.get_cached_context_length", return_value=None), \ + patch("agent.model_metadata.save_context_length"): + ctx = get_model_context_length( + model="gpt-5.6-sol", + base_url="https://chatgpt.com/backend-api/codex", + api_key="fake-token", + provider="openai-codex", + ) + assert ctx == advertised, f"advertised {advertised} must be trusted" + + @pytest.mark.parametrize("slug", ["gpt-5.5", "gpt-5.4-mini"]) + def test_slugs_that_enforce_272k_keep_advertised_value(self, slug): + """gpt-5.5 and gpt-5.4-mini both rejected 360K in the live probe — + their 272K advertisement is real enforcement, so no bump applies + (gpt-5.4 is an exact-match entry precisely to exclude -mini).""" + from agent.model_metadata import get_model_context_length + + fake_response = MagicMock() + fake_response.status_code = 200 + fake_response.json.return_value = { + "models": [{"slug": slug, "context_window": 272_000}] + } + with patch("agent.model_metadata.requests.get", return_value=fake_response), \ + patch("agent.model_metadata.get_cached_context_length", return_value=None), \ + patch("agent.model_metadata.save_context_length"): + ctx = get_model_context_length( + model=slug, + base_url="https://chatgpt.com/backend-api/codex", + api_key="fake-token", + provider="openai-codex", + ) + assert ctx == 272_000 + + def test_fallback_table_resolution_also_bumped(self): + """When the live probe fails, the 272K fallback-table value for a + verified slug is bumped the same way (same enforcement applies).""" + from agent.model_metadata import get_model_context_length + + fake_response = MagicMock() + fake_response.status_code = 401 + fake_response.json.return_value = {} + with patch("agent.model_metadata.requests.get", return_value=fake_response), \ + patch("agent.model_metadata.get_cached_context_length", return_value=None), \ + patch("agent.model_metadata.save_context_length"): + ctx = get_model_context_length( + model="gpt-5.6-sol", + base_url="https://chatgpt.com/backend-api/codex", + api_key="expired-token", + provider="openai-codex", + ) + assert ctx == 350_000 + diff --git a/tests/gateway/test_inject_internal_message.py b/tests/gateway/test_inject_internal_message.py new file mode 100644 index 000000000000..2cfa18775327 --- /dev/null +++ b/tests/gateway/test_inject_internal_message.py @@ -0,0 +1,731 @@ +"""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 +- steer vs queue mode: steer into running agent, fallback to queue +- 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, InjectInternalMessageError +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.send_kwargs: list[dict] = [] + self.handled_events: list[MessageEvent] = [] + self._message_handler = AsyncMock() + + async def send(self, chat_id, text, **kwargs): + self.sent_messages.append((chat_id, text)) + self.send_kwargs.append(kwargs) + return MagicMock(success=True, error=None) + + async def handle_message(self, event): + self.handled_events.append(event) + if self._message_handler: + 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, 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="***")} + ) + 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 = {} + 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 — queue mode (default) +# ------------------------------------------------------------------ + +class TestInjectInternalMessage: + """inject_internal_message routes an internal event to adapter.handle_message. + + Tests use the active profile name (mock default: "test-profile") + to route through 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="test-profile", + 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="test-profile", + 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="test-profile", + 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() + # 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", + ) + 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="test-profile", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="nudge payload", + notice_text="\u26a1 ATM nudge received", + ) + tg = runner.adapters[Platform.TELEGRAM] + # Notice sent first + assert tg.sent_messages == [("100000001", "\u26a1 ATM nudge received")] + assert tg.send_kwargs == [{"metadata": {"notify": True}}] + # Then event routed + assert tg.handled_events[0].text == "nudge payload" + + @pytest.mark.asyncio + async def test_missing_adapter_raises(self): + """Raises InjectInternalMessageError when no adapter for platform.""" + runner = _make_runner() + 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", + ) + assert exc.value.code == "adapter_not_found" + + @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="test-profile", + 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_reported_notice_failure_does_not_prevent_routing(self, caplog): + """A failed SendResult is observable but cannot suppress the XML event.""" + runner = _make_runner() + tg = runner.adapters[Platform.TELEGRAM] + tg.send = AsyncMock(return_value=MagicMock(success=False, error="network down")) + + with caplog.at_level("WARNING"): + await runner.inject_internal_message( + profile="test-profile", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="payload", + notice_text="notice that fails", + ) + + assert len(tg.handled_events) == 1 + assert tg.handled_events[0].text == "payload" + assert "visible notice was not delivered: network down" in caplog.text + + @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_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="test-profile", + platform=Platform.TELEGRAM, + chat_id="100000001", + text="test", + ) + assert len(default_tg.handled_events) == 1 + + @pytest.mark.asyncio + async def test_unknown_profile_raises(self): + """Raises InjectInternalMessageError when profile not found.""" + runner = _make_runner() + runner._profile_adapters["other"] = { + Platform.TELEGRAM: _FakeTelegramAdapter() + } + + 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_raises(self): + """Raises InjectInternalMessageError when _profile_adapters empty.""" + runner = _make_runner() + runner._profile_adapters = {} + + 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" + + +# ------------------------------------------------------------------ +# 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) +# ------------------------------------------------------------------ + +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()} + + + + +# ------------------------------------------------------------------ +# 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 + + # ------------------------------------------------------------------ + # 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 +# ------------------------------------------------------------------ + +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()], + "gateway_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) diff --git a/ui-tui/packages/hermes-ink/src/ink/parse-keypress.test.ts b/ui-tui/packages/hermes-ink/src/ink/parse-keypress.test.ts index f8585521cb50..3989192b9fa9 100644 --- a/ui-tui/packages/hermes-ink/src/ink/parse-keypress.test.ts +++ b/ui-tui/packages/hermes-ink/src/ink/parse-keypress.test.ts @@ -3,6 +3,15 @@ import { describe, expect, it } from 'vitest' import { INITIAL_STATE, parseMultipleKeypresses } from './parse-keypress.js' import { PASTE_END, PASTE_START } from './termio/csi.js' +describe('legacy modified return parsing', () => { + it.each(['\r', '\n'])('parses ESC+%j as one Alt+Enter keypress', lineEnding => { + const sequence = `\x1b${lineEnding}` + const [keys] = parseMultipleKeypresses(INITIAL_STATE, sequence) + + expect(keys).toEqual([expect.objectContaining({ name: 'return', ctrl: false, meta: true, shift: false, sequence })]) + }) +}) + describe('parseMultipleKeypresses bracketed paste recovery', () => { it('emits empty bracketed pastes when the terminal sends both markers', () => { const [keys, state] = parseMultipleKeypresses(INITIAL_STATE, PASTE_START + PASTE_END) diff --git a/ui-tui/packages/hermes-ink/src/ink/parse-keypress.ts b/ui-tui/packages/hermes-ink/src/ink/parse-keypress.ts index 07e31c6f5395..421aa4cca9cd 100644 --- a/ui-tui/packages/hermes-ink/src/ink/parse-keypress.ts +++ b/ui-tui/packages/hermes-ink/src/ink/parse-keypress.ts @@ -292,7 +292,7 @@ export function parseMultipleKeypresses( const inputString = isFlush ? '' : inputToString(input) // Get or create tokenizer - const tokenizer = prevState._tokenizer ?? createTokenizer({ x10Mouse: true }) + const tokenizer = prevState._tokenizer ?? createTokenizer({ x10Mouse: true, legacyAltEnter: true }) // Tokenize the input const tokens = isFlush ? tokenizer.flush() : tokenizer.feed(inputString) @@ -796,9 +796,10 @@ function parseKeypress(s: string = ''): ParsedKey { return createNavKey(s, 'mouse', false) } - if (s === '\r' || s === '\n') { + if (s === '\r' || s === '\n' || s === '\x1b\r' || s === '\x1b\n') { key.raw = undefined key.name = 'return' + key.meta = s.startsWith('\x1b') } else if (s === '\t') { key.name = 'tab' } else if (s === '\b' || s === '\x1b\b') { diff --git a/ui-tui/packages/hermes-ink/src/ink/termio/parser.test.ts b/ui-tui/packages/hermes-ink/src/ink/termio/parser.test.ts new file mode 100644 index 000000000000..e50118963417 --- /dev/null +++ b/ui-tui/packages/hermes-ink/src/ink/termio/parser.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from 'vitest' + +import { Parser } from './parser.js' + +const renderedText = (actions: ReturnType): string => + actions + .filter(action => action.type === 'text') + .flatMap(action => action.graphemes) + .map(grapheme => grapheme.value) + .join('') + +describe('output parser line endings after ESC', () => { + it.each(['\r', '\n'])('preserves %j received in the same chunk as ESC', lineEnding => { + const actions = new Parser().feed(`before\x1b${lineEnding}after`) + + expect(renderedText(actions).replaceAll('\x1b', '')).toBe(`before${lineEnding}after`) + }) + + it.each(['\r', '\n'])('preserves %j received in the chunk after ESC', lineEnding => { + const parser = new Parser() + const actions = [...parser.feed('before\x1b'), ...parser.feed(`${lineEnding}after`)] + + expect(renderedText(actions).replaceAll('\x1b', '')).toBe(`before${lineEnding}after`) + }) +}) diff --git a/ui-tui/packages/hermes-ink/src/ink/termio/tokenize.test.ts b/ui-tui/packages/hermes-ink/src/ink/termio/tokenize.test.ts index b3cf2cb5e8bc..c514aaaf8e03 100644 --- a/ui-tui/packages/hermes-ink/src/ink/termio/tokenize.test.ts +++ b/ui-tui/packages/hermes-ink/src/ink/termio/tokenize.test.ts @@ -3,6 +3,31 @@ import { describe, expect, it } from 'vitest' import { createTokenizer, type Token } from './tokenize.js' describe('tokenizer escape-sequence boundaries', () => { + it.each(['\r', '\n'])('keeps ESC+%j together when received in one feed', lineEnding => { + const t = createTokenizer({ legacyAltEnter: true }) + const sequence = `\x1b${lineEnding}` + + expect(t.feed(sequence)).toEqual([{ type: 'sequence', value: sequence }]) + expect(t.buffer()).toBe('') + }) + + it.each(['\r', '\n'])('reassembles ESC+%j split across two feeds', lineEnding => { + const t = createTokenizer({ legacyAltEnter: true }) + const sequence = `\x1b${lineEnding}` + + expect(t.feed('\x1b')).toEqual([]) + expect(t.feed(lineEnding)).toEqual([{ type: 'sequence', value: sequence }]) + expect(t.buffer()).toBe('') + }) + + it.each(['\r', '\n'])('keeps Escape distinct when it is flushed before %j', lineEnding => { + const t = createTokenizer({ legacyAltEnter: true }) + + expect(t.feed('\x1b')).toEqual([]) + expect(t.flush()).toEqual([{ type: 'sequence', value: '\x1b' }]) + expect(t.feed(lineEnding)).toEqual([{ type: 'text', value: lineEnding }]) + }) + it('reassembles a CSI mouse sequence split across two feeds', () => { const t = createTokenizer({ x10Mouse: true }) diff --git a/ui-tui/packages/hermes-ink/src/ink/termio/tokenize.ts b/ui-tui/packages/hermes-ink/src/ink/termio/tokenize.ts index 03f99cf2f4a3..f7a09ac06b01 100644 --- a/ui-tui/packages/hermes-ink/src/ink/termio/tokenize.ts +++ b/ui-tui/packages/hermes-ink/src/ink/termio/tokenize.ts @@ -31,6 +31,11 @@ type TokenizerOptions = { * output streams, and enabling this there swallows display text. Default false. */ x10Mouse?: boolean + /** + * Treat ESC followed by CR or LF as one legacy Alt+Enter key sequence. + * Only enable for keyboard input; output streams must preserve line endings. + */ + legacyAltEnter?: boolean } /** @@ -53,13 +58,14 @@ export function createTokenizer(options?: TokenizerOptions): Tokenizer { // buffer it kept last time (the continuation never arrived), we drop it. let lastFlushedBuffer = '' const x10Mouse = options?.x10Mouse ?? false + const legacyAltEnter = options?.legacyAltEnter ?? false return { feed(input: string): Token[] { // Real bytes arrived — any kept partial is no longer stale. lastFlushedBuffer = '' - const result = tokenize(input, currentState, currentBuffer, false, x10Mouse) + const result = tokenize(input, currentState, currentBuffer, false, x10Mouse, legacyAltEnter) currentState = result.state.state currentBuffer = result.state.buffer @@ -68,7 +74,7 @@ export function createTokenizer(options?: TokenizerOptions): Tokenizer { }, flush(): Token[] { - const result = tokenize('', currentState, currentBuffer, true, x10Mouse) + const result = tokenize('', currentState, currentBuffer, true, x10Mouse, legacyAltEnter) currentState = result.state.state currentBuffer = result.state.buffer @@ -109,7 +115,8 @@ function tokenize( initialState: State, initialBuffer: string, flush: boolean, - x10Mouse: boolean + x10Mouse: boolean, + legacyAltEnter: boolean ): { tokens: Token[]; state: InternalState } { const tokens: Token[] = [] @@ -177,6 +184,13 @@ function tokenize( // 'O' - SS3 result.state = 'ss3' i++ + } else if (legacyAltEnter && (code === C0.CR || code === C0.LF)) { + // Legacy terminals encode Alt+Enter as ESC followed by CR or LF. + // Keep both bytes in one token so the key parser can preserve Alt. + // A standalone Escape is emitted by flush() before a later Enter; + // without that timing boundary the legacy encoding is ambiguous. + i++ + emitSequence(data.slice(seqStart, i)) } else if (isCSIIntermediate(code)) { // Intermediate byte (e.g., ESC ( for charset) - continue buffering result.state = 'escapeIntermediate'