fix(honcho): prune orphaned observation flags; share the local-platform set; drop test-shape defenses

A flush that rebuilds an evicted session's SDK session stores its observation
flags again, and the cap pass pruned every per-session dict but that one, so the
dict grew one entry per evicted-then-flushed session. The cap pass now prunes it
with the rest.

The peer-failure notice classified platforms with its own {"cli","tui","desktop"}
set, so an ACP session read the gateway wording ("do not suggest peerName"); it
now uses agent.coding_context.INTERACTIVE_CODING_PLATFORMS, which includes acp.

Three getattr/try-except guards existed only for test doubles (a bare __new__
provider, a SimpleNamespace config); the real types always carry the attribute.
Removed, and the two tests build real objects. An unset timeout resolves to the
client's 30s default, so the join-budget test expects 30, not the 5s floor.

Timing tests keep the >= 2s wall-clock bound the testing rules ask for.

plugins/memory/honcho/__init__.py

@@ -18,7 +18,8 @@ import time
 from typing import Any, Callable, Dict, List, Optional
 
 from agent.memory_manager import sanitize_context
-from agent.memory_provider import MemoryProvider, is_trivial_prompt, spawn_context_thread
+from agent.memory_provider import MemoryProvider, is_trivial_prompt
+from agent.coding_context import INTERACTIVE_CODING_PLATFORMS as _LOCAL_PLATFORMS
 from agent.turn_author import a2a_key
 from plugins.memory.honcho.client import HonchoClientConfig, resolve_config_path
 from plugins.memory.honcho.client import _host_block, _HostLookup
@@ -83,7 +84,6 @@ _PROMPT_HEADERS = {
     ),
 }
 
-_LOCAL_PLATFORMS = frozenset({"cli", "tui", "desktop", ""})
 
 
 _FLAG_WORDS = {"1": True, "true": True, "yes": True, "on": True,
@@ -239,9 +239,7 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
             self._config = cfg
             self._recall_mode = cfg.recall_mode
             self._recall_sync = getattr(cfg, "recall_sync", False)
-            # getattr: test doubles and older config objects have no raw/host.
-            raw = getattr(cfg, "raw", None) or {}
-            look = _HostLookup(_host_block(raw, getattr(cfg, "host", "") or ""), raw)
+            look = _HostLookup(_host_block(cfg.raw, cfg.host or ""), cfg.raw)
             self._injection_log_path = self._resolve_injection_log_path(look)
             self._session_start_components = self._resolve_session_start(look)
             logger.debug("Honcho recall_mode: %s", self._recall_mode)
@@ -428,8 +426,7 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
         """Render the prefetch context, keeping only the ``injection.sessionStart`` components when pinned.
         The summary passes usable_honcho_summary here, so a contaminated one never reaches _base_context_cache."""
         ctx = {**ctx, "summary": usable_honcho_summary(ctx.get("summary")) or ""}
-        # getattr: tests build the provider without initialize().
-        allowed = getattr(self, "_session_start_components", None)
+        allowed = self._session_start_components
         parts, suppressed = [], []
         for name, key, header in _CONTEXT_SECTIONS:
             value = ctx.get(key, "")
@@ -1034,11 +1031,8 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider):
 
     def _shutdown_join_budget(self) -> float:
         """The floor, or the configured HTTP timeout when longer, so a thread blocked in a Honcho call can finish."""
-        try:
-            from plugins.memory.honcho.client_cache import _resolve_timeout_from_sources
-            return max(self._SHUTDOWN_JOIN_FLOOR, _resolve_timeout_from_sources(self._config))
-        except Exception:
-            return self._SHUTDOWN_JOIN_FLOOR
+        from plugins.memory.honcho.client_cache import _resolve_timeout_from_sources
+        return max(self._SHUTDOWN_JOIN_FLOOR, _resolve_timeout_from_sources(self._config))
 
     def shutdown(self) -> None:
         """Join the write threads, flush and stop the manager, then join every other thread this provider or its

plugins/memory/honcho/session.py

@@ -288,6 +288,8 @@ class HonchoSessionManager(SessionAuthMixin, SessionPeersMixin, SessionContextMi
                 break
             if session_id not in live_ids:
                 del self._sessions_cache[session_id]
+        for session_id in [sid for sid in self._session_observation if sid not in live_ids]:
+            del self._session_observation[session_id]
         live_peers = {p for s in self._cache.values() for p in (s.user_peer_id, s.assistant_peer_id)}
         for peer_id in list(self._peers_cache):
             if len(self._peers_cache) <= _PEERS_CACHE_MAX_SIZE:

tests/honcho_plugin/test_async_memory.py

@@ -410,7 +410,7 @@ class TestStopAsyncWriterDrain:
             mgr.shutdown(timeout=0)
 
         assert uploads == []
-        assert time.monotonic() - started < 1.0
+        assert time.monotonic() - started < 2.0
         assert mgr._async_queue.empty()
         assert session.messages[0].get("_synced") is None
         assert caplog.text.count("still unsynced") == 1
@@ -478,7 +478,7 @@ class TestAsyncWriterRetry:
         started = time.monotonic()
         mgr.stop_async_writer(timeout=5)
 
-        assert time.monotonic() - started < 1.5
+        assert time.monotonic() - started < 2.0
         assert len(calls) == 1
 
     def test_drops_after_two_failures(self, make_manager):

tests/honcho_plugin/test_shutdown.py

@@ -49,7 +49,7 @@ class TestThreadRegistry:
             assert join_plugin_threads((theirs, None), timeout=0.05) == ["other"]
             release.set()
             assert join_plugin_threads((mine,), timeout=2) == []
-            assert time.monotonic() - started < 1.5
+            assert time.monotonic() - started < 2.0
         finally:
             release.set()
             own.join(timeout=1)
@@ -119,7 +119,7 @@ class TestProviderShutdown:
             started = time.monotonic()
             with caplog.at_level(logging.WARNING, logger="plugins.memory.honcho"):
                 provider.shutdown()
-            assert 0.15 <= time.monotonic() - started < 1.0
+            assert 0.15 <= time.monotonic() - started < 2.0
             assert "honcho-session-init" in caplog.text
             assert "timed out after 0.2s" in caplog.text
         finally:
@@ -176,7 +176,7 @@ class TestProviderShutdown:
             started = time.monotonic()
             with caplog.at_level(logging.WARNING, logger="plugins.memory.honcho"):
                 provider.shutdown()
-            assert time.monotonic() - started < 1.0
+            assert time.monotonic() - started < 2.0
             assert "1 message(s) in 1 session(s) still unsynced" in caplog.text
             assert session.messages[0].get("_synced") is None
         finally:
@@ -188,10 +188,11 @@ class TestProviderShutdown:
         async_manager.fake_client._http.close.assert_not_called()
 
 
-@pytest.mark.parametrize("timeout, budget", [(60.0, 60.0), (2.0, 5.0), (None, 5.0)])
+@pytest.mark.parametrize("timeout, budget", [(60.0, 60.0), (2.0, 5.0), (None, 30.0)])
 def test_shutdown_join_budget_is_the_floor_or_the_longer_timeout(timeout, budget):
+    """Unset timeout resolves to the client's 30s default, so the join waits as long as a blocked call can."""
     provider = HonchoMemoryProvider()
-    provider._config = HonchoClientConfig(api_key="k", timeout=timeout) if timeout else SimpleNamespace()
+    provider._config = HonchoClientConfig(api_key="k", timeout=timeout)
     assert provider._shutdown_join_budget() == budget
 
 

tests/test_honcho_session_cache_bounds.py

@@ -282,6 +282,21 @@ def test_get_or_create_stores_observation_flags_with_the_entry_and_eviction_drop
     assert session.honcho_session_id not in mgr._session_observation
 
 
+def test_cap_enforcement_drops_observation_flags_a_post_eviction_flush_stored():
+    """A flush that rebuilds an evicted session's SDK session stores its flags again; the next cap pass
+    must prune that orphan like every other per-session entry, or the dict grows one entry per evicted-then-
+    flushed session."""
+    mgr = _manager()
+    live = _session(key="live")
+    mgr._cache = {"live": live}
+    mgr._session_observation = {live.honcho_session_id: {"ai_observe_others": True}, "hs-gone": {"ai_observe_others": False}}
+
+    with mgr._cache_lock:
+        mgr._enforce_cache_caps_locked()
+
+    assert set(mgr._session_observation) == {live.honcho_session_id}
+
+
 def test_flush_does_not_resurrect_an_evicted_session():
     mgr = _manager()
     session = _session(key="gone")

tests/test_honcho_startup_fail_open.py

@@ -13,6 +13,9 @@ from plugins.memory.honcho import HonchoMemoryProvider
 
 
 class _FakeHonchoConfig(SimpleNamespace):
+    raw: dict = {}
+    host: str = "hermes"
+
     def resolve_session_name(self, **kwargs):
         return "test-session"
 

tests/test_honcho_summary_sanitize.py

@@ -122,7 +122,7 @@ def test_session_context_omits_planning_only_summary() -> None:
 def test_format_first_turn_omits_contaminated_summary() -> None:
     from plugins.memory.honcho import HonchoMemoryProvider
 
-    plugin = HonchoMemoryProvider.__new__(HonchoMemoryProvider)
+    plugin = HonchoMemoryProvider()
     out = plugin._format_first_turn_context(
         {
             "summary": PLANNING_ONLY,