"""Tests: cross-connection bot relay (tools/bot_relay.py + message_agent route). Connections ARE the peer set: every Desktop-connected gateway must be message_agent-reachable. These tests pin the gateway-side plumbing — roster validation, target resolution (incl. ambiguity), outbox claim atomicity, reply write validation — and the two behavior contracts the relay adds to message_agent: - a target resolving against the Desktop-synced relay roster is queued as an envelope and acknowledged like any DM (fire-and-forget, waiter spawned); - the legacy-SOUL dedupe (empty protocol section) NO LONGER strips the tool: the injection/execution gates key on managed-install, not section text. """ import json import re from pathlib import Path import pytest from tools import bot_relay from tools.bot_mode_dm import ( MESSAGE_AGENT_TOOL_NAME, ensure_message_agent_tool, message_agent_tool, ) @pytest.fixture() def root(tmp_path): return tmp_path def _rows(): return [ { "profile": "default", "handle": "hermes", "connection_id": "cloud-1", "connection_label": "Hermes Cloud", "title": "Moxie", "description": "Main cloud agent", }, { "profile": "researcher", "handle": "researcher", "connection_id": "ssh-vps", "connection_label": "VPS", }, ] # ── roster ─────────────────────────────────────────────────────────────────── def test_roster_roundtrip_and_validation(root): rows = _rows() + [ {"profile": "", "handle": "x", "connection_id": "c"}, # no profile {"profile": "bad name!", "connection_id": "c"}, # bad charset "not-a-dict", {"profile": "default", "handle": "hermes", "connection_id": "cloud-1"}, # dupe ] count = bot_relay.write_remote_roster(root, rows) assert count == 2 back = bot_relay.read_remote_roster(root) assert [r["profile"] for r in back] == ["default", "researcher"] assert back[0]["title"] == "Moxie" def test_roster_read_missing_and_corrupt(root): assert bot_relay.read_remote_roster(root) == [] base = bot_relay.relay_root(root) base.mkdir(parents=True) (base / bot_relay.ROSTER_FILE).write_text("{corrupt", encoding="utf-8") assert bot_relay.read_remote_roster(root) == [] def test_resolve_remote_target_forms(root): bot_relay.write_remote_roster(root, _rows()) roster = bot_relay.read_remote_roster(root) assert bot_relay.resolve_remote_target("researcher", roster)["connection_id"] == "ssh-vps" assert bot_relay.resolve_remote_target("@hermes", roster)["profile"] == "default" # profile name resolves too assert bot_relay.resolve_remote_target("default", roster)["connection_id"] == "cloud-1" # exact connection-qualified form assert bot_relay.resolve_remote_target("hermes@cloud-1", roster)["profile"] == "default" # profile@connection — the form Desktop's mention middleware annotates # for remote bots (#97678); the UI alias form must not be required assert bot_relay.resolve_remote_target("default@cloud-1", roster)["profile"] == "default" assert bot_relay.resolve_remote_target("hermes@nope", roster) is None assert bot_relay.resolve_remote_target("ghost", roster) is None def test_resolve_ambiguous_handle_across_connections(root): rows = _rows() + [ {"profile": "researcher", "handle": "researcher", "connection_id": "cloud-1"} ] bot_relay.write_remote_roster(root, rows) roster = bot_relay.read_remote_roster(root) assert bot_relay.resolve_remote_target("researcher", roster) == "ambiguous" match = bot_relay.resolve_remote_target("researcher@ssh-vps", roster) assert match["connection_id"] == "ssh-vps" forms = bot_relay.remote_target_forms(roster) assert "researcher@ssh-vps" in forms and "researcher@cloud-1" in forms assert "hermes" in forms # unique handle stays bare # ── outbox / replies ───────────────────────────────────────────────────────── def test_enqueue_claim_is_atomic_and_single_shot(root): bot_relay.write_remote_roster(root, _rows()) roster = bot_relay.read_remote_roster(root) target = bot_relay.resolve_remote_target("researcher", roster) env = bot_relay.enqueue_envelope( root, target=target, message="hi", sender_profile="work", sender_handle="work" ) assert re.match(r"^[0-9a-f]{32}$", env["id"]) claimed = bot_relay.claim_pending_envelopes(root) assert [e["id"] for e in claimed] == [env["id"]] assert claimed[0]["target_connection"] == "ssh-vps" assert claimed[0]["message"] == "hi" # second drain: nothing (no double delivery) assert bot_relay.claim_pending_envelopes(root) == [] def test_claim_skips_non_dict_envelope(root): """A parseable-but-non-object outbox file must not reach the Desktop consumer or crash the claim sweep; the claim stays claimed so it is not re-queued.""" bot_relay.write_remote_roster(root, _rows()) roster = bot_relay.read_remote_roster(root) target = bot_relay.resolve_remote_target("researcher", roster) env = bot_relay.enqueue_envelope( root, target=target, message="hi", sender_profile="work", sender_handle="work" ) base = bot_relay.relay_root(root) bad = base / bot_relay.OUTBOX_DIR / f"{'9' * 32}.json" bad.write_text('"not an envelope"', encoding="utf-8") claimed = bot_relay.claim_pending_envelopes(root) assert [e["id"] for e in claimed] == [env["id"]] assert not bad.exists() and (base / bot_relay.CLAIMED_DIR / bad.name).exists() def test_write_reply_validates_envelope_id(root): with pytest.raises(ValueError): bot_relay.write_reply(root, "../../etc/passwd", reply="x") path = bot_relay.write_reply(root, "a" * 32, reply="pong") data = json.loads(Path(path).read_text(encoding="utf-8")) assert data["reply"] == "pong" and not data["error"] def test_write_reply_keeps_the_first_settled_reply_for_an_envelope(root): """Idempotent by envelope id: a re-offered delivery's second outcome (or a late duplicate) must not displace the reply the waiter already read, so the answer never turns into an error.""" env_id = "a" * 32 first = bot_relay.write_reply(root, env_id, reply="the answer") second = bot_relay.write_reply(root, env_id, error="target busy", reason="target_busy") assert second == first record = json.loads(first.read_text(encoding="utf-8")) assert (record["reply"], record["error"], record["reason"]) == ("the answer", "", "") def test_write_reply_reason_passthrough_and_classification(root): # explicit reason is persisted verbatim path = bot_relay.write_reply(root, "c" * 32, error="boom", reason="delivery_timeout") data = json.loads(Path(path).read_text(encoding="utf-8")) assert data["reason"] == "delivery_timeout" and data["error"] == "boom" # no reason given → classified from error text path = bot_relay.write_reply(root, "d" * 32, error="Error code: 429 - rate limit") data = json.loads(Path(path).read_text(encoding="utf-8")) assert data["reason"] == "provider_rate_limit" # success reply carries an empty reason path = bot_relay.write_reply(root, "e" * 32, reply="ok") data = json.loads(Path(path).read_text(encoding="utf-8")) assert data["reason"] == "" and data["reply"] == "ok" def test_waiter_is_a_runner_entrypoint_the_approval_gate_lets_through(root): """The waiter is spawned through terminal_tool from the SENDER's turn. When a bot replies to a teammate from its own one-shot delivery turn, that turn runs under ``approvals.single_query_mode`` (default ``deny``), and inline interpreter code is a flagged pattern — so the reply waiter was refused exactly when bots talked to each other, and the reply never woke the sender.""" import shlex from tools.approval import detect_dangerous_command env = {"id": "b" * 32, "target_handle": "researcher", "target_connection": "ssh-vps"} cmd = bot_relay.waiter_command(root, env) parts = shlex.split(cmd) assert detect_dangerous_command(cmd)[0] is False, cmd assert parts[1].endswith("bot_mode_dm.py") and parts[2] == "--wait-reply" assert parts[3] == str(bot_relay.relay_root(root) / bot_relay.REPLIES_DIR / f"{'b' * 32}.json") assert parts[4:] == ["@researcher on ssh-vps", str(bot_relay.REPLY_WAIT_SECONDS)] def test_waiter_outlives_the_desktop_deliver_deadline(): """The Desktop posts its timeout reply when RELAY_DELIVER_TIMEOUT_MS passes. A waiter that gave up first left that reply, and any turn finishing after minute 15, in a file nobody read (#93911). relay-deliver-budget.test.ts pins the TS constants against these Python ones. The re-offer window (#111021) sits between the two: only past the Desktop's deadline — and the gateway's own worst-case deliver hold, which ends before it — is a claimed envelope's silence provably a dead Desktop rather than a slow turn, and the waiter must still be listening when the ONE re-offered delivery hits its own Desktop deadline.""" live_hold_s = bot_relay.TURN_WAIT_SECONDS_FALLBACK + bot_relay.TURN_ATTEMPT_TIMEOUT_SECONDS * bot_relay.TURN_MAX_ATTEMPTS desktop_budget_s = live_hold_s + bot_relay.DESKTOP_DELIVER_SETTLEMENT_MARGIN_SECONDS assert bot_relay.DESKTOP_DELIVER_TIMEOUT_SECONDS == desktop_budget_s assert bot_relay.REOFFER_AFTER_SECONDS > desktop_budget_s > live_hold_s assert bot_relay.REPLY_WAIT_SECONDS > bot_relay.REOFFER_AFTER_SECONDS + desktop_budget_s @pytest.mark.parametrize( ("reply_file", "expected_code", "expected_tokens"), [ ({"reply": "pong"}, 0, ["@researcher on ssh-vps", "pong"]), ({"reply": ""}, 0, ["@researcher on ssh-vps"]), ({"error": "turn failed", "reason": "provider_rate_limit"}, 1, ["@researcher on ssh-vps", "provider_rate_limit", "turn failed"]), ({"error": "turn failed"}, 1, ["@researcher on ssh-vps", "turn failed"]), (None, 1, ["@researcher on ssh-vps", "0.3s"]), ], ids=["reply", "empty-reply", "typed-error", "untyped-error", "gave-up"], ) def test_waiter_prints_the_completion_notification_the_sender_wakes_on(root, capsys, reply_file, expected_code, expected_tokens): """The waiter's stdout IS the sender's completion notification: the reply, a typed failure the sender can branch on without parsing prose (#93091), or an honest give-up that names the budget.""" from tools import bot_mode_dm env = {"id": "d" * 32, "target_handle": "researcher", "target_connection": "ssh-vps"} reply_path = bot_relay.relay_root(root) / bot_relay.REPLIES_DIR / f"{env['id']}.json" reply_path.parent.mkdir(parents=True, exist_ok=True) if reply_file is not None: reply_path.write_text(json.dumps(reply_file), encoding="utf-8") code = bot_mode_dm._delivery_main(["--wait-reply", str(reply_path), "@researcher on ssh-vps", "0.3"]) out = capsys.readouterr().out assert code == expected_code assert all(token in out for token in expected_tokens), out assert bot_mode_dm._delivery_main(["--wait-reply", str(reply_path)]) == 2 def test_roster_rejects_connection_id_outside_handle_charset(root): bad = [ {"profile": "researcher", "handle": "researcher", "connection_id": "vps'); print(1)"}, {"profile": "researcher", "handle": "researcher", "connection_id": "foo'bar"}, {"profile": "researcher", "handle": "researcher", "connection_id": "ssh vps"}, {"profile": "researcher", "handle": "researcher", "connection_id": "a" * 65}, ] assert bot_relay.write_remote_roster(root, bad) == 0 good = { "profile": "researcher", "handle": "researcher", "connection_id": "ssh-vps", } assert bot_relay.write_remote_roster(root, [good]) == 1 def test_hostile_roster_fields_ride_as_argv_data(root): """Envelope fields come from the Desktop-pushed roster — untrusted. They must never become source text: the waiter takes them as argv, so a payload shaped like Python is a label.""" import shlex import subprocess inj = "x'); open(r'/tmp/pwned','w').write('pwned'); print('x" env = {"id": "c" * 32, "target_handle": "researcher", "target_connection": inj} cmd = bot_relay.waiter_command(root, env) parts = shlex.split(cmd) assert "-c" not in parts assert parts[4] == f"@researcher on {inj}" reply_path = bot_relay.relay_root(root) / bot_relay.REPLIES_DIR / f"{env['id']}.json" reply_path.parent.mkdir(parents=True, exist_ok=True) reply_path.write_text(json.dumps({"reply": "pong"}), encoding="utf-8") proc = subprocess.run(parts, capture_output=True, text=True, timeout=30) assert proc.returncode == 0 and proc.stdout.splitlines() == [f"Reply from @researcher on {inj}:", "pong"] # ── message_agent integration: relay route + legacy-SOUL gate fix ─────────── import textwrap def _managed_home(tmp_path, *, legacy_soul=False): home = tmp_path / ".hermes" home.mkdir(exist_ok=True) d = home / "profiles" / "researcher" d.mkdir(parents=True, exist_ok=True) (d / "profile.yaml").write_text( textwrap.dedent( """\ description: teammate for tests ui_meta: hermes-bots: shape: cloud """ ), encoding="utf-8", ) if legacy_soul: (home / "SOUL.md").write_text( "# Soul\n\n## Messaging other agents\nold shellout protocol\n", encoding="utf-8", ) return home class _FakeDB: def __init__(self, home, title): self.db_path = str(home / "state.db") self._title = title def get_session_title(self, _sid): return self._title class _FakeAgent: def __init__(self, home, title="Bot Chat"): self._session_db = _FakeDB(home, title) self.session_id = "sess-1" self._session_title_hint = None self._bot_mode_protocol = True self.tools: list = [] self.valid_tool_names: set = set() @pytest.fixture(autouse=True) def _fresh_probe_cache(): from tools import bot_mode_probe bot_mode_probe._reset_cache_for_tests() yield bot_mode_probe._reset_cache_for_tests() def test_tool_injects_despite_legacy_soul_protocol(tmp_path): """A SOUL.md still carrying the plugin-appended protocol must not cost the TOOL (nor, since load-time stripping, the live section).""" from tools import bot_mode_probe home = _managed_home(tmp_path, legacy_soul=True) assert bot_mode_probe.get_bot_mode_protocol_section(home) != "" agent = _FakeAgent(home) assert ensure_message_agent_tool(agent) is True assert [t["function"]["name"] for t in agent.tools] == [MESSAGE_AGENT_TOOL_NAME] def test_relay_route_queues_envelope_and_spawns_waiter(tmp_path, monkeypatch): home = _managed_home(tmp_path) bot_relay.write_remote_roster(home, [ {"profile": "default", "handle": "hermes", "connection_id": "cloud-1", "connection_label": "Hermes Cloud", "title": "Moxie"}, ]) spawned = {} def _fake_spawn(command, label, *, task_id, agent, **_): spawned["command"] = command spawned["label"] = label return json.dumps({"status": "queued", "to": label}) monkeypatch.setattr("tools.bot_mode_dm._spawn_delivery", _fake_spawn) agent = _FakeAgent(home) out = json.loads(message_agent_tool(target="hermes", message="ping", agent=agent)) assert out.get("status") == "queued" assert "Hermes Cloud" in spawned["label"] # envelope landed in the outbox with attribution prefixed pending = bot_relay.claim_pending_envelopes(home) assert len(pending) == 1 assert pending[0]["target_connection"] == "cloud-1" assert pending[0]["target_profile"] == "default" assert pending[0]["message"].startswith("Message from 🤖 hermes (@hermes): ping") # waiter watches this envelope's reply file assert pending[0]["id"] in spawned["command"] def test_relay_route_ambiguous_target_errors_with_forms(tmp_path, monkeypatch): home = _managed_home(tmp_path) bot_relay.write_remote_roster(home, [ {"profile": "scout", "handle": "scout", "connection_id": "cloud-1"}, {"profile": "scout", "handle": "scout", "connection_id": "ssh-vps"}, ]) monkeypatch.setattr( "tools.bot_mode_dm._spawn_delivery", lambda *a, **k: json.dumps({"status": "queued"}), ) agent = _FakeAgent(home) out = json.loads(message_agent_tool(target="scout", message="hi", agent=agent)) assert "scout@cloud-1" in out.get("error", "") and "scout@ssh-vps" in out["error"] # connection-qualified form goes through out2 = json.loads(message_agent_tool(target="scout@ssh-vps", message="hi", agent=agent)) assert out2.get("status") == "queued" def test_remote_default_is_addressable_by_its_title_slug(tmp_path, monkeypatch): """A remote ``default`` is ``@hermes`` like every gateway's own, so its Bot Mode title is the only bare form that singles it out: ``message_agent`` accepts the slug, the prompt roster offers it (never bare ``@hermes``, which is the LOCAL default), and the form does not depend on roster order.""" from tools import bot_mode_probe home = _managed_home(tmp_path) rows = [ {"profile": "default", "handle": "hermes", "connection_id": "vps-1", "title": "CoS Bot"}, {"profile": "default", "handle": "hermes", "connection_id": "cloud-1", "title": "Ops Bot"}, {"profile": "cos-bot", "handle": "cos-bot", "connection_id": "cloud-1", "title": "Other"}, ] monkeypatch.setattr("tools.bot_mode_dm._spawn_delivery", lambda *a, **k: json.dumps({"status": "queued"})) agent = _FakeAgent(home) for order in (rows, rows[::-1]): bot_relay.write_remote_roster(home, order) roster = bot_relay.read_remote_roster(home) forms = dict(zip((r["connection_id"] + "/" + r["profile"] for r in roster), bot_relay.remote_target_forms(roster, bot_mode_probe.local_taken_forms(home)))) # an exact handle beats a colliding title slug; the collided title falls back to the qualified form assert forms == {"cloud-1/cos-bot": "cos-bot", "vps-1/default": "hermes@vps-1", "cloud-1/default": "ops-bot"} assert bot_relay.resolve_remote_target("cos-bot", roster)["profile"] == "cos-bot" out = json.loads(message_agent_tool(target="@ops-bot", message="ping", agent=agent)) assert out.get("status") == "queued" [env] = bot_relay.claim_pending_envelopes(home) assert (env["target_connection"], env["target_profile"]) == ("cloud-1", "default") # a bare @hermes from the local default is the two remote defaults — offered under their reply-safe forms err = json.loads(message_agent_tool(target="hermes", message="ping", agent=agent))["error"] assert "hermes@vps-1" in err and "ops-bot" in err section = bot_mode_probe.get_bot_mode_protocol_section(home) assert "`@ops-bot`" in section and "`@hermes@vps-1`" in section and "- `@hermes` —" not in section def test_protocol_section_lists_remote_teammates(tmp_path): from tools import bot_mode_probe home = _managed_home(tmp_path) bot_relay.write_remote_roster(home, [ {"profile": "default", "handle": "hermes", "connection_id": "cloud-1", "connection_label": "Hermes Cloud", "title": "Moxie"}, ]) section = bot_mode_probe.get_bot_mode_protocol_section(home, force_refresh=True) # Offered under its title slug: bare `@hermes` is THIS gateway's own default (#103731). assert "`@moxie`" in section and "Hermes Cloud" in section def test_capability_fingerprint_changes_with_relay_roster(tmp_path): from tools import bot_mode_probe home = _managed_home(tmp_path) before = bot_mode_probe.capability_fingerprint(home) bot_relay.write_remote_roster(home, [ {"profile": "default", "handle": "hermes", "connection_id": "cloud-1"}, ]) after = bot_mode_probe.capability_fingerprint(home) assert before != after # eternal Bot Chats refresh once on roster change # ── stale artifact sweep (housekeeping contract) ───────────────────────────── def test_cleanup_bot_relay_artifacts_sweeps_stale_plaintext(tmp_path, monkeypatch): import os as _os import time as _time monkeypatch.setenv("HERMES_HOME", str(tmp_path)) target = {"profile": "scout", "handle": "scout", "connection_id": "cloud-1", "connection_label": "", "title": "", "description": ""} stale_env = bot_relay.enqueue_envelope( tmp_path, target=target, message="old secret", sender_profile="default", sender_handle="hermes", ) fresh_env = bot_relay.enqueue_envelope( tmp_path, target=target, message="new secret", sender_profile="default", sender_handle="hermes", ) base = bot_relay.relay_root(tmp_path) stale_reply = bot_relay.write_reply(tmp_path, stale_env["id"], reply="done") old = _time.time() - bot_relay.STALE_AFTER_SECONDS - 1 _os.utime(base / bot_relay.OUTBOX_DIR / f"{stale_env['id']}.json", (old, old)) _os.utime(stale_reply, (old, old)) removed = bot_relay.cleanup_bot_relay_artifacts() assert removed == 2 assert not (base / bot_relay.OUTBOX_DIR / f"{stale_env['id']}.json").exists() assert not stale_reply.exists() assert (base / bot_relay.OUTBOX_DIR / f"{fresh_env['id']}.json").exists() def test_cleanup_bot_relay_artifacts_missing_dir_is_zero(tmp_path, monkeypatch): monkeypatch.setenv("HERMES_HOME", str(tmp_path / "nope")) assert bot_relay.cleanup_bot_relay_artifacts() == 0 # ── #93091 item 2: offline fail-fast + drain-time TTL ──────────────────────── import os as _os2 import time as _time2 def _target(conn="cloud-1", profile="scout", handle="scout"): return {"profile": profile, "handle": handle, "connection_id": conn, "connection_label": "", "title": "", "description": ""} def test_enqueue_fails_fast_when_row_explicitly_offline(root): bot_relay.write_remote_roster(root, [ {"profile": "scout", "handle": "scout", "connection_id": "cloud-1", "online": False}, ]) roster = bot_relay.read_remote_roster(root) assert roster[0]["online"] is False # additive field survives normalize with pytest.raises(bot_relay.EnvelopeRefusedError) as ei: bot_relay.enqueue_envelope( root, target=roster[0], message="hi", sender_profile="default", sender_handle="hermes", ) assert ei.value.reason == "runtime_offline" assert "offline" in str(ei.value) # nothing was written to the outbox outdir = bot_relay.relay_root(root) / bot_relay.OUTBOX_DIR assert not outdir.exists() or list(outdir.glob("*.json")) == [] def test_enqueue_fails_fast_when_target_absent_from_fresh_roster(root): bot_relay.write_remote_roster(root, _rows()) # fresh, no 'scout' row with pytest.raises(bot_relay.EnvelopeRefusedError) as ei: bot_relay.enqueue_envelope( root, target=_target(), message="hi", sender_profile="default", sender_handle="hermes", ) assert ei.value.reason == "runtime_offline" def test_enqueue_fails_open_when_liveness_unknown(root): # 1. no roster ever synced → unknown → enqueue env = bot_relay.enqueue_envelope( root, target=_target(), message="hi", sender_profile="default", sender_handle="hermes", ) assert (bot_relay.relay_root(root) / bot_relay.OUTBOX_DIR / f"{env['id']}.json").exists() # 2. stale roster missing the target → unknown → enqueue bot_relay.write_remote_roster(root, _rows()) roster_path = bot_relay.relay_root(root) / bot_relay.ROSTER_FILE old = _time2.time() - bot_relay.ROSTER_FRESH_SECONDS - 5 _os2.utime(roster_path, (old, old)) env2 = bot_relay.enqueue_envelope( root, target=_target(), message="hi again", sender_profile="default", sender_handle="hermes", ) assert (bot_relay.relay_root(root) / bot_relay.OUTBOX_DIR / f"{env2['id']}.json").exists() # 3. fresh roster listing the target without an online flag → enqueue bot_relay.write_remote_roster(root, _rows()) target = bot_relay.read_remote_roster(root)[1] # researcher@ssh-vps env3 = bot_relay.enqueue_envelope( root, target=target, message="hello", sender_profile="default", sender_handle="hermes", ) assert (bot_relay.relay_root(root) / bot_relay.OUTBOX_DIR / f"{env3['id']}.json").exists() def test_drain_expires_old_envelope_with_queued_expired_reply(root): env = bot_relay.enqueue_envelope( root, target=_target(), message="too late", sender_profile="default", sender_handle="hermes", ) base = bot_relay.relay_root(root) out_path = base / bot_relay.OUTBOX_DIR / f"{env['id']}.json" # backdate the envelope beyond the TTL env["created_at"] = int(_time2.time()) - bot_relay.DEFAULT_ENVELOPE_TTL_SECONDS - 10 out_path.write_text(json.dumps(env), encoding="utf-8") claimed = bot_relay.claim_pending_envelopes(root) assert claimed == [] # not delivered assert not out_path.exists() # expired outbox file removed reply = json.loads( (base / bot_relay.REPLIES_DIR / f"{env['id']}.json").read_text(encoding="utf-8") ) assert reply["reason"] == "queued_expired" assert reply["error"] assert not reply["reply"] def test_the_outbox_is_claimed_oldest_first(root): """Two DMs from one sender to one agent must arrive in the order they were sent. The Desktop delivers each target's claimed envelopes in the order this list gives them, one turn at a time, so the claim IS the delivery order — and sorting by filename ordered them by ``uuid4().hex``. The names here are forced into the reverse of the send order to pin that deterministically, which random ids reproduce half the time.""" first = bot_relay.enqueue_envelope( root, target=_target(), message="do this first", sender_profile="default", sender_handle="hermes", ) second = bot_relay.enqueue_envelope( root, target=_target(), message="then this", sender_profile="default", sender_handle="hermes", ) outbox = bot_relay.relay_root(root) / bot_relay.OUTBOX_DIR now = _time2.time() for env, name, sent_at in ((first, "f" * 32, now - 2), (second, "0" * 32, now - 1)): path = outbox / f"{name}.json" (outbox / f"{env['id']}.json").rename(path) _os2.utime(path, (sent_at, sent_at)) claimed = bot_relay.claim_pending_envelopes(root) assert [e["id"] for e in claimed] == [first["id"], second["id"]] assert [e["message"] for e in claimed] == ["do this first", "then this"] def test_drain_delivers_fresh_envelope_under_ttl(root): env = bot_relay.enqueue_envelope( root, target=_target(), message="on time", sender_profile="default", sender_handle="hermes", ) claimed = bot_relay.claim_pending_envelopes(root) assert [e["id"] for e in claimed] == [env["id"]] # no spurious expiry reply for a delivered envelope base = bot_relay.relay_root(root) assert not (base / bot_relay.REPLIES_DIR / f"{env['id']}.json").exists() def test_drain_ttl_zero_disables_expiry(root, monkeypatch): monkeypatch.setattr(bot_relay, "_envelope_ttl_seconds", lambda: 0) env = bot_relay.enqueue_envelope( root, target=_target(), message="never expires", sender_profile="default", sender_handle="hermes", ) base = bot_relay.relay_root(root) out_path = base / bot_relay.OUTBOX_DIR / f"{env['id']}.json" env["created_at"] = int(_time2.time()) - 10 * 3600 out_path.write_text(json.dumps(env), encoding="utf-8") claimed = bot_relay.claim_pending_envelopes(root) assert [e["id"] for e in claimed] == [env["id"]] def test_invalid_ttl_config_falls_back_instead_of_breaking_drain(monkeypatch): monkeypatch.setattr(bot_relay, "_bot_mode_cfg", lambda *args, **kwargs: "not-a-number") assert bot_relay._envelope_ttl_seconds() == bot_relay.DEFAULT_ENVELOPE_TTL_SECONDS def test_message_agent_surfaces_runtime_offline_refusal(tmp_path, monkeypatch): home = _managed_home(tmp_path) bot_relay.write_remote_roster(home, [ {"profile": "default", "handle": "hermes", "connection_id": "cloud-1", "connection_label": "Hermes Cloud", "online": False}, ]) monkeypatch.setattr( "tools.bot_mode_dm._spawn_delivery", lambda *a, **k: json.dumps({"status": "queued"}), ) agent = _FakeAgent(home) out = json.loads(message_agent_tool(target="hermes", message="ping", agent=agent)) assert out.get("reason") == "runtime_offline" assert "offline" in out.get("error", "") # fail-fast means no envelope was queued assert bot_relay.claim_pending_envelopes(home) == [] # ── delivery turn author (HERMES_TURN_AUTHOR on the recipient turn) ────────── def test_delivery_turn_author_from_envelope_sender_fields(): author = bot_relay.delivery_turn_author("ops", "ops-bot") assert author == {"id": "bot:ops", "name": "ops-bot", "is_bot": True} # The display name falls back to the profile; the id never comes from the handle. assert bot_relay.delivery_turn_author("ops", "") == {"id": "bot:ops", "name": "ops", "is_bot": True} assert bot_relay.delivery_turn_author("", "ops-bot") is None assert bot_relay.delivery_turn_author(None, None) is None def test_delivery_turn_author_qualifies_a_remote_sender_by_its_connection(): """A relayed DM always crosses gateways, so the sender's connection id is part of the author id, ``local`` included; the recipient's own ``ops`` is the only bare ``bot:ops``.""" remote = bot_relay.delivery_turn_author("ops", "ops-bot", "cloud-1") assert remote == {"id": "bot:cloud-1/ops", "name": "ops-bot", "is_bot": True} assert bot_relay.delivery_turn_author("ops", "ops-bot", "local") == {"id": "bot:local/ops", "name": "ops-bot", "is_bot": True} # An older Desktop that sends no connection id still yields an author. assert bot_relay.delivery_turn_author("ops", "ops-bot", "") == {"id": "bot:ops", "name": "ops-bot", "is_bot": True} def test_delivery_env_carries_only_the_given_author(monkeypatch): """The dispatcher's own HERMES_TURN_AUTHOR never reaches the child: dropped without an author, replaced with one.""" from agent.turn_author import TURN_AUTHOR_ENV monkeypatch.setenv("HERMES_RELAY_TEST_MARKER", "kept") monkeypatch.setenv(TURN_AUTHOR_ENV, json.dumps({"id": "bot:previous", "name": "previous", "is_bot": True})) monkeypatch.setenv("HERMES_SESSION_KEY", "session-A") monkeypatch.setenv("HERMES_UI_SESSION_ID", "ui-A") monkeypatch.setenv("HERMES_SESSION_ID", "session-A") monkeypatch.setenv("HERMES_SESSION_PROFILE", "profile-A") # A session-* knob, not identity: stripping it would break the child's watcher tuning. monkeypatch.setenv("HERMES_SESSION_STALL_TIMEOUT", "97") assert TURN_AUTHOR_ENV not in bot_relay.delivery_env(None) env = bot_relay.delivery_env(bot_relay.delivery_turn_author("ops", "ops")) assert json.loads(env[TURN_AUTHOR_ENV]) == {"id": "bot:ops", "name": "ops", "is_bot": True} assert env["HERMES_RELAY_TEST_MARKER"] == "kept" assert "HERMES_SESSION_KEY" not in env assert "HERMES_UI_SESSION_ID" not in env assert "HERMES_SESSION_ID" not in env assert "HERMES_SESSION_PROFILE" not in env assert env["HERMES_SESSION_STALL_TIMEOUT"] == "97" def test_delivery_env_under_multiplex_names_the_pinned_launch_home(tmp_path, monkeypatch): """A relayed DM into the launch profile spawns with the launch home and its secrets, even after a host mirrors another home into HERMES_HOME; a bound scope or home override still wins.""" from agent.secret_scope import reset_secret_scope, set_multiplex_active, set_secret_scope from hermes_constants import reset_hermes_home_override, set_hermes_home_override launch, mirrored = tmp_path / "launch", tmp_path / "mirrored" launch.mkdir() mirrored.mkdir() (launch / ".env").write_text("OPENROUTER_API_KEY=sk-launch\n", encoding="utf-8") monkeypatch.setenv("HERMES_HOME", str(launch)) monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) set_multiplex_active(True) # pins the launch home monkeypatch.setenv("HERMES_HOME", str(mirrored)) env = bot_relay.delivery_env(None, None) assert env["HERMES_HOME"] == str(launch) assert env["OPENROUTER_API_KEY"] == "sk-launch" token = set_secret_scope({"OPENROUTER_API_KEY": "sk-bound"}) try: assert bot_relay.delivery_env(None, None)["OPENROUTER_API_KEY"] == "sk-bound" finally: reset_secret_scope(token) (mirrored / ".env").write_text("OPENROUTER_API_KEY=sk-override\n", encoding="utf-8") home_token = set_hermes_home_override(str(mirrored)) try: env = bot_relay.delivery_env(None, None) assert (env["HERMES_HOME"], env["OPENROUTER_API_KEY"]) == (str(mirrored), "sk-override") finally: reset_hermes_home_override(home_token) def test_delivery_env_single_profile_host_passes_the_process_env_through(tmp_path, monkeypatch): """Without multiplex nothing is pinned or overlaid: the child sees the process env as before.""" (tmp_path / ".env").write_text("OPENROUTER_API_KEY=from-dotenv\n", encoding="utf-8") monkeypatch.setenv("HERMES_HOME", str(tmp_path)) monkeypatch.setenv("OPENROUTER_API_KEY", "from-shell") assert bot_relay.delivery_env(None, None)["OPENROUTER_API_KEY"] == "from-shell"