fix(honcho): make honcho_search do real cross-session message search

honcho_search routed through search_context() -> peer.context(search_query=),
which returns the peer's standing representation + card. The search_query arg
does not turn that endpoint into a search, so results were effectively
query-independent: the same representation blob regardless of the query. Factual
lookups ('what medication', 'which value did we pick') returned noise.

Rewire search_context() to call the workspace message-search endpoint
(Honcho.search) with a peer_perspective filter: RRF-ranked (hybrid semantic +
full-text) raw message excerpts spanning every session the peer was a member of,
across all authors, membership-time-scoped. This is the cross-session factual
recall primitive.

peer_perspective is chosen over the alternatives because it is the only scope
that is simultaneously (a) cross-session, (b) inclusive of assistant-authored
facts about the peer (peer-author search drops these, and they are a large part
of what you want to recall about yourself), and (c) privacy-scoped to the peer's
own sessions (plain workspace search leaks other peers' sessions).

- snippets are labeled by author so the model can tell user-stated facts from
  assistant-derived ones
- max_tokens is now an enforced budget (was accepted but meaningless)
- graceful fallback to peer-authored search if peer_perspective is unsupported
- query length clamped under the embedding input cap

Replaces 3 change-detector tests that asserted the old representation-dump
behavior with 4 that assert the message-search contract + fallback path.

plugins/memory/honcho/session.py

@@ -1110,43 +1110,94 @@ class HonchoSessionManager:
         peer: str = "user",
     ) -> str:
         """
-        Semantic search over Honcho session context.
+        Hybrid (semantic + keyword) message search over a peer's history.
 
-        Returns raw excerpts ranked by relevance to the query. No LLM
-        reasoning — cheaper and faster than dialectic_query. Good for
-        factual lookups where the model will do its own synthesis.
+        Calls Honcho's workspace message-search endpoint with a
+        ``peer_perspective`` filter, which returns RRF-ranked raw message
+        snippets drawn from every session the peer was a member of (scoped
+        by their join/leave windows). This is the cross-session factual
+        recall primitive: it finds what was actually *said* — including
+        messages authored by the assistant about the peer — rather than
+        dumping the standing representation.
+
+        No LLM reasoning is involved — cheaper and faster than
+        ``dialectic_query``. Good for factual lookups where the model will
+        do its own synthesis over the returned excerpts.
 
         Args:
-            session_key: Session to search against.
-            query: Search query for semantic matching.
-            max_tokens: Token budget for returned content.
-            peer: Peer alias or explicit peer ID to search about.
+            session_key: Session whose workspace/peer scope to search within.
+            query: Search query (hybrid semantic + full-text).
+            max_tokens: Approximate budget for returned content. Snippets are
+                accumulated until this budget (≈4 chars/token) is exhausted.
+            peer: Peer alias or explicit peer ID whose sessions to search.
 
         Returns:
-            Relevant context excerpts as a string, or empty string if none.
+            Ranked message excerpts as a formatted string, or empty string
+            if none found.
         """
         session = self._cache.get(session_key)
         if not session:
             return ""
 
-        try:
-            observer_peer_id, target = self._resolve_observer_target(session, peer)
+        # Resolve the peer whose message history we scope the search to.
+        # We deliberately use the *target* peer (the human by default) for the
+        # peer_perspective filter so the search spans every session that peer
+        # participated in, across all authors — not just messages they wrote.
+        peer_id = self._resolve_peer_id(session, peer)
 
-            ctx = self._fetch_peer_context(
-                observer_peer_id,
-                search_query=query,
-                target=target,
-            )
-            parts = []
-            if ctx["representation"]:
-                parts.append(ctx["representation"])
-            card = ctx["card"] or []
-            if card:
-                parts.append("\n".join(f"- {f}" for f in card))
-            return "\n\n".join(parts)
-        except Exception as e:
-            logger.debug("Honcho search_context failed: %s", e)
+        # Honcho caps query length for the embedding model; keep well under it.
+        q = (query or "").strip()
+        if not q:
             return ""
+        if len(q) > 4000:
+            q = q[:4000]
+
+        # Convert the token budget into an approximate result count + char cap.
+        # ~4 chars/token; assume a useful snippet is a few hundred chars.
+        char_budget = max(200, int(max_tokens) * 4)
+        limit = max(3, min(20, char_budget // 300))
+
+        try:
+            messages = self.honcho.search(
+                q,
+                filters={"peer_perspective": peer_id},
+                limit=limit,
+            )
+        except Exception as e:
+            logger.debug("Honcho message search failed (peer_perspective=%s): %s", peer_id, e)
+            # Fall back to peer-authored search if the perspective filter is
+            # unsupported by the running Honcho version.
+            try:
+                peer_obj = self._get_or_create_peer(peer_id)
+                messages = peer_obj.search(q, limit=limit)
+            except Exception as e2:
+                logger.debug("Honcho peer search fallback also failed: %s", e2)
+                return ""
+
+        if not messages:
+            return ""
+
+        # Format ranked snippets, honoring the char budget. Label the author
+        # so the model can tell user-stated facts from assistant-derived ones.
+        assistant_id = session.assistant_peer_id
+        lines: list[str] = []
+        used = 0
+        for m in messages:
+            content = (getattr(m, "content", "") or "").strip()
+            if not content:
+                continue
+            author = getattr(m, "peer_id", "") or "unknown"
+            who = "assistant" if author == assistant_id else author
+            sess = getattr(m, "session_id", "") or ""
+            # Trim individual snippets so one long message can't eat the budget.
+            snippet = content[:1200]
+            entry = f"[{who}{f' · {sess}' if sess else ''}] {snippet}"
+            if used + len(entry) > char_budget and lines:
+                break
+            lines.append(entry)
+            used += len(entry)
+
+        return "\n\n".join(lines)
 
     def create_conclusion(self, session_key: str, content: str, peer: str = "user") -> bool:
         """Write a conclusion about a target peer back to Honcho.

tests/honcho_plugin/test_session.py

@@ -247,55 +247,70 @@ class TestPeerLookupHelpers:
         assert result == ["Role: user"]
         assistant_peer.set_card.assert_called_once_with(["Role: user"], target=session.user_peer_id)
 
-    def test_search_context_uses_assistant_perspective_with_target(self):
+    def test_search_context_uses_peer_perspective_message_search(self):
+        """honcho_search must do cross-session message search scoped to the
+        target peer via the peer_perspective filter — not dump the
+        representation. Regression guard for the representation-dump bug."""
         mgr, session = self._make_cached_manager()
-        assistant_peer = MagicMock()
-        assistant_peer.context.return_value = SimpleNamespace(
-            representation="Robert runs neuralancer",
-            peer_card=["Location: Melbourne"],
-        )
-        mgr._get_or_create_peer = MagicMock(return_value=assistant_peer)
+        honcho_client = MagicMock()
+        honcho_client.search.return_value = [
+            SimpleNamespace(content="Robert runs neuralancer", peer_id="hermes", session_id="s-old", id="m1"),
+            SimpleNamespace(content="I founded neuralancer in 2019", peer_id="robert", session_id="s-old", id="m2"),
+        ]
+        mgr._honcho = honcho_client
 
         result = mgr.search_context(session.key, "neuralancer")
 
+        # Returns the actual message content, ranked.
         assert "Robert runs neuralancer" in result
-        assert "- Location: Melbourne" in result
-        assistant_peer.context.assert_called_once_with(
-            target=session.user_peer_id,
-            search_query="neuralancer",
-        )
+        assert "neuralancer in 2019" in result
+        # Scoped to the target (user) peer's sessions, all authors.
+        honcho_client.search.assert_called_once()
+        _args, kwargs = honcho_client.search.call_args
+        assert kwargs["filters"] == {"peer_perspective": session.user_peer_id}
+        # Assistant-authored messages are labeled so the model can tell
+        # user-stated facts from assistant-derived ones.
+        assert "[assistant" in result
 
-    def test_search_context_unified_mode_uses_user_self_context(self):
+    def test_search_context_explicit_ai_peer_searches_ai_perspective(self):
         mgr, session = self._make_cached_manager()
-        mgr._ai_observe_others = False
-        user_peer = MagicMock()
-        user_peer.context.return_value = SimpleNamespace(
-            representation="Unified self context",
-            peer_card=["Name: Robert"],
-        )
-        mgr._get_or_create_peer = MagicMock(return_value=user_peer)
-
-        result = mgr.search_context(session.key, "self")
-
-        assert "Unified self context" in result
-        user_peer.context.assert_called_once_with(search_query="self")
-
-    def test_search_context_accepts_explicit_ai_peer_id(self):
-        mgr, session = self._make_cached_manager()
-        ai_peer = MagicMock()
-        ai_peer.context.return_value = SimpleNamespace(
-            representation="Assistant self context",
-            peer_card=["Role: Assistant"],
-        )
-        mgr._get_or_create_peer = MagicMock(return_value=ai_peer)
+        honcho_client = MagicMock()
+        honcho_client.search.return_value = [
+            SimpleNamespace(content="Assistant note", peer_id="hermes", session_id="s1", id="m1"),
+        ]
+        mgr._honcho = honcho_client
 
         result = mgr.search_context(session.key, "assistant", peer=session.assistant_peer_id)
 
-        assert "Assistant self context" in result
-        ai_peer.context.assert_called_once_with(
-            target=session.assistant_peer_id,
-            search_query="assistant",
-        )
+        assert "Assistant note" in result
+        _args, kwargs = honcho_client.search.call_args
+        assert kwargs["filters"] == {"peer_perspective": session.assistant_peer_id}
+
+    def test_search_context_empty_query_returns_empty(self):
+        mgr, session = self._make_cached_manager()
+        honcho_client = MagicMock()
+        mgr._honcho = honcho_client
+
+        assert mgr.search_context(session.key, "   ") == ""
+        honcho_client.search.assert_not_called()
+
+    def test_search_context_falls_back_to_peer_search_on_filter_error(self):
+        """If the workspace search with peer_perspective raises (older Honcho),
+        fall back to peer-authored search rather than returning nothing."""
+        mgr, session = self._make_cached_manager()
+        honcho_client = MagicMock()
+        honcho_client.search.side_effect = RuntimeError("peer_perspective unsupported")
+        mgr._honcho = honcho_client
+        peer_obj = MagicMock()
+        peer_obj.search.return_value = [
+            SimpleNamespace(content="fallback hit", peer_id="robert", session_id="s1", id="m1"),
+        ]
+        mgr._get_or_create_peer = MagicMock(return_value=peer_obj)
+
+        result = mgr.search_context(session.key, "anything")
+
+        assert "fallback hit" in result
+        peer_obj.search.assert_called_once()
 
     def test_get_prefetch_context_fetches_user_and_ai_from_peer_api(self):
         mgr, session = self._make_cached_manager()