feat(honcho): declare identity_signature and drop the gateway's honcho keys

The gateway agent cache read honcho.json itself through a honcho-named
block in gateway/run.py and gateway/run_agent_cache.py. Every other memory
provider had no way to bust the cache when its identity mapping changed.

HonchoMemoryProvider.identity_signature() now returns the same values under
provider-neutral keys: user_identity, agent_identity, pin_user_identity,
runtime_identity_prefix, user_identity_aliases, session_prefixing. The
gateway files them under memory.<key> through the MemoryProvider hook. The
hook reads config only, memoizes on the file's mtime and size, and returns
an empty dict when the file cannot be read.

The honcho-specific extractor, its memo and its key tuple are gone from the
gateway. The pinPeerName cache-busting test now asserts on
memory.pin_user_identity.

gateway/run.py

@@ -4238,11 +4238,6 @@ class GatewayRunner(
         ("memory", "provider"), ("checkpoints", "enabled"), ("checkpoints", "max_snapshots"),
         ("checkpoints", "max_total_size_mb"), ("checkpoints", "max_file_size_mb"))
 
-    _HONCHO_CACHE_BUSTING_KEYS = (
-        "honcho.peer_name", "honcho.ai_peer", "honcho.pin_peer_name", "honcho.runtime_peer_prefix",
-        "honcho.user_peer_aliases")
-    _HONCHO_CACHE_BUSTING_MEMO: dict[tuple[str, int | None], dict[str, Any]] = {}
-
     @staticmethod
     def _init_cached_agent_for_turn(agent: Any, interrupt_depth: int) -> None:
         """Reset per-turn state on a cached agent before a new turn starts.

gateway/run_agent_cache.py

@@ -42,34 +42,6 @@ def _tuple_agent(entry: Any) -> Any:
 class GatewayAgentCacheMixin:
     """Agent cache, session model overrides, turn leases, run generations and conversation-scope reset for GatewayRunner."""
 
-    @classmethod
-    def _extract_honcho_cache_busting_config(cls) -> dict[str, Any]:
-        """Extract Honcho identity keys, memoized by honcho.json mtime; all-None when unavailable."""
-        try:
-            from plugins.memory.honcho.client import HonchoClientConfig, resolve_config_path
-            path = resolve_config_path()
-            try:
-                mtime_ns = path.stat().st_mtime_ns
-            except OSError:
-                mtime_ns = None
-            memo_key = (str(path), mtime_ns)
-            cached = cls._HONCHO_CACHE_BUSTING_MEMO.get(memo_key)
-            if cached is not None:
-                return dict(cached)
-            hcfg = HonchoClientConfig.from_global_config(config_path=path)
-            aliases = hcfg.user_peer_aliases or {}
-            values = {
-                "honcho.peer_name": hcfg.peer_name,
-                "honcho.ai_peer": hcfg.ai_peer,
-                "honcho.pin_peer_name": bool(hcfg.pin_peer_name),
-                "honcho.runtime_peer_prefix": hcfg.runtime_peer_prefix or "",
-                "honcho.user_peer_aliases": sorted(aliases.items()) if isinstance(aliases, dict) else [],
-            }
-            cls._HONCHO_CACHE_BUSTING_MEMO = {memo_key: values}
-            return dict(values)
-        except Exception:
-            return dict.fromkeys(cls._HONCHO_CACHE_BUSTING_KEYS)
-
     @classmethod
     def _extract_cache_busting_config(cls, user_config: dict | None) -> dict:
         """Values that must bust the cached agent, as a flat dict keyed by 'section.key'. Missing keys /
@@ -89,12 +61,7 @@ class GatewayAgentCacheMixin:
             out["tools.registry_generation"] = getattr(registry, "_generation", None)
         except Exception:
             out["tools.registry_generation"] = None
-        # Honcho identity-mapping keys live in honcho.json, not user_config; only read that file
-        # when Honcho is the active memory provider.
-        provider = cfg_get(cfg, "memory", "provider")
-        honcho = isinstance(provider, str) and provider.lower() == "honcho"
-        out.update(cls._extract_honcho_cache_busting_config() if honcho else dict.fromkeys(cls._HONCHO_CACHE_BUSTING_KEYS))
-        for key, value in cls._memory_provider_identity_signature(provider).items():
+        for key, value in cls._memory_provider_identity_signature(cfg_get(cfg, "memory", "provider")).items():
             out[f"memory.{key}"] = value
         return out
 

plugins/memory/honcho/__init__.py

@@ -17,7 +17,7 @@ from typing import Any, Callable, Dict, List, Optional
 
 from agent.memory_manager import sanitize_context
 from agent.memory_provider import MemoryProvider, is_trivial_prompt
-from plugins.memory.honcho.client import spawn_context_thread
+from plugins.memory.honcho.client import HonchoClientConfig, resolve_config_path, spawn_context_thread
 from plugins.memory.honcho.dialectic import DialecticMixin
 from plugins.memory.honcho.tool_schemas import ALL_TOOL_SCHEMAS
 from tools.registry import tool_error
@@ -122,6 +122,8 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
         self._turn_count = 0
         # Author of the turn in flight, refreshed by on_turn_start.
         self._turn_author: dict[str, Any] = {}
+        # (config path, mtime_ns, size) -> identity_signature() values.
+        self._identity_signature_memo: dict[tuple, dict[str, Any]] = {}
         self._query_rewrite_enabled = False
         self._injection_frequency = "every-turn"  # or "first-turn"
         self._context_cadence = 1   # minimum turns between context API calls
@@ -552,6 +554,35 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
     # Shared with the core prefetch gate so the two classifiers can never drift apart.
     _is_trivial_prompt = staticmethod(is_trivial_prompt)
 
+    def identity_signature(self) -> Dict[str, Any]:
+        """Identity-mapping values from honcho.json that must bust a cached gateway agent when they
+        change. Reads config only, never the network, and memoizes on the file's mtime and size so the
+        per-message call stays a stat. ``{}`` when the config cannot be read."""
+        try:
+            path = resolve_config_path()
+            try:
+                stat = path.stat()
+                memo_key = (str(path), stat.st_mtime_ns, stat.st_size)
+            except OSError:
+                memo_key = (str(path), None, None)
+            cached = self._identity_signature_memo.get(memo_key)
+            if cached is not None:
+                return dict(cached)
+            cfg = HonchoClientConfig.from_global_config(config_path=path)
+            aliases = cfg.user_peer_aliases if isinstance(cfg.user_peer_aliases, dict) else {}
+            values = {
+                "user_identity": cfg.peer_name,
+                "agent_identity": cfg.ai_peer,
+                "pin_user_identity": bool(cfg.pin_peer_name),
+                "runtime_identity_prefix": cfg.runtime_peer_prefix or "",
+                "user_identity_aliases": sorted(aliases.items()),
+                "session_prefixing": [bool(cfg.session_peer_prefix)],
+            }
+            self._identity_signature_memo = {memo_key: values}
+            return dict(values)
+        except Exception:
+            return {}
+
     def on_turn_start(self, turn_number: int, message: str, **kwargs) -> None:
         """Track turn count for cadence, and record who wrote this turn: a shared session carries
         several participants, and the peer resolved at session init only names whoever opened it."""

tests/honcho_plugin/test_identity_signature.py

@@ -0,0 +1,86 @@
+"""``HonchoMemoryProvider.identity_signature()``: the identity-mapping values the gateway folds
+into its agent-cache key, read from honcho.json without touching the network."""
+
+import json
+
+import pytest
+
+from plugins.memory.honcho import HonchoMemoryProvider
+
+
+@pytest.fixture
+def honcho_json(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    path = tmp_path / "honcho.json"
+
+    def _write(**values):
+        path.write_text(json.dumps({"apiKey": "k", **values}))
+        return path
+
+    return _write
+
+
+def test_signature_uses_neutral_keys(honcho_json):
+    honcho_json(peerName="eri", aiPeer="hermes", pinUserPeer=True, runtimePeerPrefix="tg_",
+                userPeerAliases={"222": "bob", "111": "alice"}, sessionPeerPrefix=True)
+
+    sig = HonchoMemoryProvider().identity_signature()
+
+    assert sig == {
+        "user_identity": "eri",
+        "agent_identity": "hermes",
+        "pin_user_identity": True,
+        "runtime_identity_prefix": "tg_",
+        "user_identity_aliases": [("111", "alice"), ("222", "bob")],
+        "session_prefixing": [True],
+    }
+    assert not any(k.startswith("honcho") for k in sig)
+
+
+def test_signature_defaults_without_identity_keys(honcho_json):
+    honcho_json()
+
+    sig = HonchoMemoryProvider().identity_signature()
+
+    assert sig["user_identity"] is None
+    assert sig["pin_user_identity"] is False
+    assert sig["runtime_identity_prefix"] == ""
+    assert sig["user_identity_aliases"] == []
+    assert sig["session_prefixing"] == [False]
+
+
+def test_signature_tracks_edits_to_the_file(honcho_json):
+    provider = HonchoMemoryProvider()
+    honcho_json(peerName="eri", pinUserPeer=True)
+    assert provider.identity_signature()["pin_user_identity"] is True
+
+    honcho_json(peerName="eri", pinUserPeer=False)
+    assert provider.identity_signature()["pin_user_identity"] is False
+
+
+def test_signature_is_memoized_on_an_unchanged_file(honcho_json, monkeypatch):
+    honcho_json(peerName="eri")
+    provider = HonchoMemoryProvider()
+    first = provider.identity_signature()
+
+    from plugins.memory.honcho import client as client_module
+    monkeypatch.setattr(client_module.HonchoClientConfig, "from_global_config",
+                        classmethod(lambda cls, **kw: pytest.fail("config re-read on an unchanged file")))
+    assert provider.identity_signature() == first
+
+
+def test_signature_never_touches_the_network(honcho_json, network_attempts):
+    honcho_json(peerName="eri")
+    HonchoMemoryProvider().identity_signature()
+    assert network_attempts == []
+
+
+def test_signature_is_empty_when_config_cannot_be_read(honcho_json, monkeypatch):
+    honcho_json(peerName="eri")
+    from plugins.memory.honcho import client as client_module
+
+    def _boom(cls, **kw):
+        raise RuntimeError("unreadable")
+
+    monkeypatch.setattr(client_module.HonchoClientConfig, "from_global_config", classmethod(_boom))
+    assert HonchoMemoryProvider().identity_signature() == {}

tests/honcho_plugin/test_pin_peer_name.py

@@ -515,11 +515,14 @@ class TestPinTransition:
         )
 
     def test_cache_busting_signature_reflects_pin_peer_name(self, tmp_path, monkeypatch):
-        """Gateway agent cache must bust when honcho.json's pinPeerName flips."""
+        """Gateway agent cache must bust when honcho.json's pinPeerName flips. The gateway reads the
+        flag through ``identity_signature()`` and files it under ``memory.*``."""
         from gateway.run import GatewayRunner
+        from gateway.run_agent_cache import GatewayAgentCacheMixin
 
         cfg_path = tmp_path / "honcho.json"
         monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        monkeypatch.setattr(GatewayAgentCacheMixin, "_MEMORY_IDENTITY_PROVIDER_MEMO", {})
 
         cfg_path.write_text(json.dumps({"apiKey": "k", "peerName": "Igor", "pinPeerName": True}))
         sig_pinned = GatewayRunner._extract_cache_busting_config({"memory": {"provider": "honcho"}})
@@ -527,7 +530,9 @@ class TestPinTransition:
         cfg_path.write_text(json.dumps({"apiKey": "k", "peerName": "Igor", "pinPeerName": False}))
         sig_unpinned = GatewayRunner._extract_cache_busting_config({"memory": {"provider": "honcho"}})
 
-        assert sig_pinned["honcho.pin_peer_name"] != sig_unpinned["honcho.pin_peer_name"]
+        assert sig_pinned["memory.pin_user_identity"] is True
+        assert sig_unpinned["memory.pin_user_identity"] is False
+        assert not any(k.startswith("honcho.") for k in sig_pinned)
 
 
 class TestProfilePeerUniqueness: