feat(tools): bring Teknium's peer session messaging onto current main

agent/activity_tracking.py

@@ -80,6 +80,11 @@ def _touch_activity(
                 heartbeat_current_worker_from_env()
                 # Fold new operator notes into the running turn (OUT-OF-BAND steer).
                 inject_new_comments_from_env(self)
+        # Peer-session messages (tools/peer_messaging_tool.py) ride the same steer
+        # channel. Rate-limited inside; a no-inbox session exits on one is_dir().
+        with suppress(Exception):
+            from tools.peer_messaging_tool import inject_peer_messages
+            inject_peer_messages(self)
         if force_persist or is_terminal_compression_provenance(resolved_provenance):
             # A terminal compression stamp must land durably NOW. The heartbeat wrote "context compression
             # in progress" moments ago, so deferring this write to the next 60s window leaves the durable

tests/tools/test_peer_messaging.py

@@ -0,0 +1,85 @@
+"""Peer session messaging (tools/peer_messaging_tool.py) — invariant tests.
+
+1. End-to-end mailbox contract: a message sent to a known session via the registry
+   handler is drained by ``inject_peer_messages`` into the target agent's steer
+   channel with peer provenance declared, and the inbox file is consumed.
+2. Trust/scope contract: sending to an unknown session id fails without creating a
+   mailbox, and a session with no inbox drains nothing.
+"""
+
+import json
+import time
+
+import pytest
+
+
+@pytest.fixture
+def peer_env(tmp_path, monkeypatch):
+    """Temp HERMES_HOME with a real SessionDB containing one live target session."""
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    from hermes_state import SessionDB
+    db = SessionDB(db_path=home / "state.db")
+    db.create_session("target-session-1", source="cli")
+    db.close()
+    return home
+
+
+class _StubAgent:
+    def __init__(self, session_id):
+        self.session_id = session_id
+        self.steers = []
+
+    def steer(self, text):
+        self.steers.append(text)
+        return True
+
+
+def _dispatch(name, args, session_id):
+    from tools.registry import registry
+    return json.loads(registry.dispatch(name, args, session_id=session_id))
+
+
+def test_peer_send_delivers_via_steer_and_consumes_inbox(peer_env):
+    import tools.peer_messaging_tool as pm
+
+    result = _dispatch(
+        "peer_send",
+        {"target_session_id": "target-session-1", "message": "heads up: schema changed"},
+        session_id="sender-session",
+    )
+    assert result.get("success") is True
+
+    inbox = pm._inbox_dir("target-session-1")
+    assert [p for p in inbox.iterdir() if p.suffix == ".json"], "message file must be queued"
+
+    agent = _StubAgent("target-session-1")
+    assert pm.inject_peer_messages(agent) is True
+    delivered = "\n".join(agent.steers)
+    # Provenance: attributed to the sender session, not the user.
+    assert "sender-session" in delivered
+    assert "heads up: schema changed" in delivered
+    assert "user authority" in delivered  # explicit non-user provenance framing
+    # Consumed: nothing left to deliver, and the rate limit halts an immediate re-poll.
+    assert not [p for p in inbox.iterdir() if p.suffix == ".json"]
+    agent2 = _StubAgent("target-session-1")
+    agent2._peer_inbox_last_drain_mono = time.monotonic()
+    assert pm.inject_peer_messages(agent2) is False
+
+
+def test_unknown_target_rejected_and_no_inbox_is_a_noop(peer_env):
+    import tools.peer_messaging_tool as pm
+
+    result = _dispatch(
+        "peer_send",
+        {"target_session_id": "no-such-session", "message": "hello"},
+        session_id="sender-session",
+    )
+    assert result.get("success") is False or "error" in result
+    assert not pm._inbox_dir("no-such-session").exists(), "failed send must not create a mailbox"
+
+    # A session nobody messaged drains nothing and never steers.
+    agent = _StubAgent("never-messaged")
+    assert pm.inject_peer_messages(agent) is False
+    assert agent.steers == []

tests/tools/test_peer_messaging_delivery.py

@@ -0,0 +1,97 @@
+"""Real activity/steer/persistence contracts for the continuation of #106423."""
+
+import json
+import threading
+
+from hermes_state import SessionDB
+from model_tools import handle_function_call
+from run_agent import AIAgent
+
+
+def _recipient(db, session_id):
+    agent = object.__new__(AIAgent)
+    agent.session_id = session_id
+    agent._session_db = db
+    agent._session_db_created = True
+    agent._pending_steer = None
+    agent._pending_steer_lock = threading.Lock()
+    agent.valid_tool_names = []
+    return agent
+
+
+def _send(target, body, sender):
+    return json.loads(handle_function_call(
+        "peer_send", {"target_session_id": target, "message": body},
+        session_id=sender,
+    ))
+
+
+def test_peer_activity_delivers_user_row_with_provenance_and_persists(tmp_path, monkeypatch):
+    """The real receive path works without recipient send tools and preserves history."""
+    import tools.peer_messaging_tool as pm
+
+    home = tmp_path / "home"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.delenv("HERMES_KANBAN_TASK", raising=False)
+    db = SessionDB(db_path=home / "state.db")
+    try:
+        db.create_session("sender", source="cli")
+        db.create_session("recipient", source="cli")
+        discovered = json.loads(handle_function_call("peer_sessions", {}, session_id="sender"))
+        assert [peer["session_id"] for peer in discovered["peers"]] == ["recipient"]
+        assert _send("recipient", "The schema is ready", "sender")["success"]
+        envelope = next(pm._inbox_dir("recipient").glob("*.json"))
+        # Windows editors may add a BOM without changing the envelope's JSON.
+        envelope.write_bytes(b"\xef\xbb\xbf" + envelope.read_bytes())
+        agent = _recipient(db, "recipient")
+        messages = [
+            {"role": "assistant", "tool_calls": [{"id": "probe", "type": "function",
+             "function": {"name": "terminal", "arguments": "{}"}}]},
+            {"role": "tool", "tool_call_id": "probe", "content": "done"},
+        ]
+        original_prefix = json.dumps(messages)
+        agent._touch_activity("tool batch finished")
+        agent._apply_pending_steer_to_tool_results(messages, 1)
+        assert json.dumps(messages[:-1]) == original_prefix
+        row = messages[-1]
+        assert row["role"] == "user"
+        assert "sender" in row["content"] and "The schema is ready" in row["content"]
+        assert "peer" in row["content"].lower()
+        assert agent._flush_messages_to_session_db(messages) is True
+        stored = db.get_messages("recipient")
+        assert stored[-1]["role"] == row["role"]
+        assert stored[-1]["content"] == row["content"]
+        assert not list(pm._inbox_dir("recipient").glob("*.json"))
+        count = len(stored)
+        agent._touch_activity("next activity")
+        agent._apply_pending_steer_to_tool_results(messages, 1)
+        assert agent._flush_messages_to_session_db(messages) is True
+        assert len(db.get_messages("recipient")) == count
+    finally:
+        db.close()
+
+
+def test_recipient_store_pins_mailbox_when_ambient_profile_changes(tmp_path, monkeypatch):
+    """A callback must not drain another profile's same-ID inbox."""
+    import tools.peer_messaging_tool as pm
+
+    homes = [tmp_path / "first", tmp_path / "second"]
+    databases = []
+    try:
+        for home, body in zip(homes, ("first profile payload", "second profile payload")):
+            home.mkdir()
+            monkeypatch.setenv("HERMES_HOME", str(home))
+            db = SessionDB(db_path=home / "state.db")
+            databases.append(db)
+            db.create_session("recipient", source="cli")
+            assert _send("recipient", body, "sender")["success"]
+        agent = _recipient(databases[0], "recipient")
+        assert pm.inject_peer_messages(agent) is True
+        pending = agent._drain_pending_steer()
+        assert "first profile payload" in pending
+        assert "second profile payload" not in pending
+        assert list(pm._inbox_dir("recipient").glob("*.json"))
+    finally:
+        for db in databases:
+            db.close()