mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-20 15:33:54 +00:00
1634 lines
74 KiB
Python
1634 lines
74 KiB
Python
"""Honcho memory plugin — MemoryProvider for Honcho AI-native memory.
|
||
|
||
Provides cross-session user modeling with dialectic Q&A, semantic search,
|
||
peer cards, and persistent conclusions via the Honcho SDK. Honcho provides AI-native cross-session user
|
||
modeling with dialectic Q&A, semantic search, peer cards, and conclusions.
|
||
|
||
The 4 tools (profile, search, context, conclude) are exposed through
|
||
the MemoryProvider interface.
|
||
|
||
Config: Uses the existing Honcho config chain:
|
||
1. $HERMES_HOME/honcho.json (profile-scoped)
|
||
2. ~/.honcho/config.json (legacy global)
|
||
3. Environment variables
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import re
|
||
import threading
|
||
import time
|
||
from typing import Any, Callable, Dict, List, Optional
|
||
|
||
from agent.memory_manager import sanitize_context
|
||
from agent.memory_provider import MemoryProvider
|
||
from tools.registry import tool_error
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tool schemas (moved from tools/honcho_tools.py)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
PROFILE_SCHEMA = {
|
||
"name": "honcho_profile",
|
||
"description": (
|
||
"Read or write a peer's CARD — a short, curated list of standing facts "
|
||
"about that peer (name, role, preferences, communication style, recurring "
|
||
"patterns). This is the cheapest, fastest Honcho call: no query, no LLM, "
|
||
"just the current card. Pass `card` to overwrite it; omit `card` to read. "
|
||
"An empty read returns a `hint` explaining why (observation disabled, fresh "
|
||
"peer, representation still warming up) — that is NOT an error; the card "
|
||
"accumulates over time from observed conversation. "
|
||
"Related tools: honcho_context for the fuller standing snapshot (card + "
|
||
"representation + summary + recent messages); honcho_search to find "
|
||
"specific things that were actually said; honcho_reasoning for a "
|
||
"synthesized answer to a question."
|
||
),
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"peer": {
|
||
"type": "string",
|
||
"description": "Peer to query. Built-in aliases: 'user' (default), 'ai'. Or pass any peer ID from this workspace.",
|
||
},
|
||
"card": {
|
||
"type": "array",
|
||
"items": {"type": "string"},
|
||
"description": "New peer card as a list of fact strings. Omit to read the current card.",
|
||
},
|
||
},
|
||
"required": [],
|
||
},
|
||
}
|
||
|
||
SEARCH_SCHEMA = {
|
||
"name": "honcho_search",
|
||
"description": (
|
||
"Hybrid (semantic + keyword) search over a peer's actual message "
|
||
"history across ALL past sessions they took part in — not just the "
|
||
"current one. Returns RRF-ranked raw message excerpts (what was "
|
||
"literally said, including the assistant's own messages about the "
|
||
"peer), no LLM synthesis. Cheaper and faster than honcho_reasoning. "
|
||
"Use this to recall specific past facts — 'what did I say about X', "
|
||
"'what was the regimen/decision/config we settled on' — and reason "
|
||
"over the excerpts yourself. For nuanced questions needing synthesis, "
|
||
"use honcho_reasoning instead."
|
||
),
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"query": {
|
||
"type": "string",
|
||
"description": "What to look for — a topic, keyword, name, or natural-language description of the fact you're trying to recall.",
|
||
},
|
||
"max_tokens": {
|
||
"type": "integer",
|
||
"description": "Approximate budget for returned excerpts (default 800, max 2000). Larger budgets return more/longer ranked snippets.",
|
||
},
|
||
"peer": {
|
||
"type": "string",
|
||
"description": "Whose history to search. Built-in aliases: 'user' (default), 'ai'. Or pass any peer ID from this workspace. Spans every session that peer took part in.",
|
||
},
|
||
},
|
||
"required": ["query"],
|
||
},
|
||
}
|
||
|
||
REASONING_SCHEMA = {
|
||
"name": "honcho_reasoning",
|
||
"description": (
|
||
"Ask Honcho's dialectic agent a natural-language question about a peer and "
|
||
"get back a SYNTHESIZED answer. This is the only Honcho tool that runs an "
|
||
"LLM: it agentically searches both raw messages and derived conclusions, "
|
||
"reasons over them, and writes a prose answer — so it is the slowest and "
|
||
"most expensive call (seconds + tokens). Reach for it for nuanced or "
|
||
"open-ended questions ('how does this person prefer to receive feedback?', "
|
||
"'what's their relationship to project X?') where you want Honcho to do the "
|
||
"synthesis. For a specific fact that was stated, prefer honcho_search "
|
||
"(cheap, raw excerpts, you synthesize). For standing profile facts, prefer "
|
||
"honcho_profile / honcho_context (no LLM). "
|
||
"Pass reasoning_level to control depth: minimal (fast/cheap), low (default), "
|
||
"medium, high, max (deep/expensive). Omit for the configured default."
|
||
),
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"query": {
|
||
"type": "string",
|
||
"description": "A natural language question.",
|
||
},
|
||
"reasoning_level": {
|
||
"type": "string",
|
||
"description": (
|
||
"Override the default reasoning depth. "
|
||
"Omit to use the configured default (typically low). "
|
||
"Guide:\n"
|
||
"- minimal: quick factual lookups (name, role, simple preference)\n"
|
||
"- low: straightforward questions with clear answers\n"
|
||
"- medium: multi-aspect questions requiring synthesis across observations\n"
|
||
"- high: complex behavioral patterns, contradictions, deep analysis\n"
|
||
"- max: thorough audit-level analysis, leave no stone unturned"
|
||
),
|
||
"enum": ["minimal", "low", "medium", "high", "max"],
|
||
},
|
||
"peer": {
|
||
"type": "string",
|
||
"description": "Peer to query. Built-in aliases: 'user' (default), 'ai'. Or pass any peer ID from this workspace.",
|
||
},
|
||
},
|
||
"required": ["query"],
|
||
},
|
||
}
|
||
|
||
CONTEXT_SCHEMA = {
|
||
"name": "honcho_context",
|
||
"description": (
|
||
"Retrieve the standing SNAPSHOT Honcho holds for the current session — "
|
||
"session summary, the peer's representation, the peer card, and the most "
|
||
"recent messages — in one call. No query, no LLM synthesis (cheaper than "
|
||
"honcho_reasoning). Use it to orient yourself on what Honcho currently "
|
||
"knows about this conversation and peer. This is a fixed snapshot, not a "
|
||
"search: to look up a specific past fact use honcho_search; to ask a "
|
||
"question and get a synthesized answer use honcho_reasoning; for just the "
|
||
"compact card use honcho_profile."
|
||
),
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"peer": {
|
||
"type": "string",
|
||
"description": "Peer to query. Built-in aliases: 'user' (default), 'ai'. Or pass any peer ID from this workspace.",
|
||
},
|
||
},
|
||
"required": [],
|
||
},
|
||
}
|
||
|
||
CONCLUDE_SCHEMA = {
|
||
"name": "honcho_conclude",
|
||
"description": (
|
||
"Write, delete, or list CONCLUSIONS — persistent, derived facts about a peer that "
|
||
"feeds their long-term profile (card + representation). Use this to record "
|
||
"something durable you've learned about the peer (a stable preference, a "
|
||
"correction, a standing constraint) so future sessions carry it forward. "
|
||
"You MUST pass exactly one of `conclusion` (to create), `delete_id` (to "
|
||
"delete), or `list` (to list/search); any other combination is an error. "
|
||
"A deletion ID is an opaque server-generated string: first call with `list=true` "
|
||
"and optionally `query`, then pass the returned ID as `delete_id`. "
|
||
"Deletion exists only for "
|
||
"PII removal — for merely wrong facts, write a corrected conclusion instead; "
|
||
"Honcho self-heals contradictions over time. This is a WRITE tool: to read "
|
||
"the profile use honcho_profile / honcho_context, and to search what was "
|
||
"said use honcho_search."
|
||
),
|
||
"parameters": {
|
||
"type": "object",
|
||
"properties": {
|
||
"conclusion": {
|
||
"type": "string",
|
||
"description": "A factual statement to persist. Provide this when creating a conclusion. Do not send it together with delete_id or list.",
|
||
},
|
||
"delete_id": {
|
||
"type": "string",
|
||
"description": "Conclusion ID to delete for PII removal. Provide this when deleting a conclusion. Do not send it together with conclusion or list. Get this id from a prior `list` call — never guess it.",
|
||
},
|
||
"list": {
|
||
"type": "boolean",
|
||
"description": "Set to true to list or search stored conclusions (with their ids) instead of creating or deleting one. Do not send together with conclusion or delete_id.",
|
||
},
|
||
"query": {
|
||
"type": "string",
|
||
"description": "Optional semantic search query, used only when `list` is true. Omit to list the most recent conclusions instead of searching.",
|
||
},
|
||
"peer": {
|
||
"type": "string",
|
||
"description": "The peer the conclusion is ABOUT. Built-in aliases: 'user' (default), 'ai'. Or pass any peer ID from this workspace.",
|
||
},
|
||
},
|
||
"required": [],
|
||
},
|
||
}
|
||
|
||
|
||
ALL_TOOL_SCHEMAS = [PROFILE_SCHEMA, SEARCH_SCHEMA, REASONING_SCHEMA, CONTEXT_SCHEMA, CONCLUDE_SCHEMA]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# MemoryProvider implementation
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class HonchoMemoryProvider(MemoryProvider):
|
||
"""Honcho AI-native memory with dialectic Q&A and persistent user modeling."""
|
||
|
||
def backup_paths(self) -> List[str]:
|
||
"""Honcho keeps its peer/session config under ~/.honcho when no
|
||
profile-local honcho.json exists (see client.resolve_config_path)."""
|
||
paths: List[str] = []
|
||
try:
|
||
from .client import resolve_global_config_path
|
||
global_cfg = resolve_global_config_path()
|
||
# Capture the whole ~/.honcho dir so sibling state travels with it.
|
||
paths.append(str(global_cfg.parent))
|
||
except Exception:
|
||
pass
|
||
return paths
|
||
|
||
def __init__(self, query_rewriter: Optional[Callable[[str], str]] = None):
|
||
self._manager = None # HonchoSessionManager
|
||
self._config = None # HonchoClientConfig
|
||
self._session_key = ""
|
||
self._query_rewriter = query_rewriter
|
||
self._prefetch_result = ""
|
||
self._prefetch_lock = threading.Lock()
|
||
self._prefetch_thread: Optional[threading.Thread] = None
|
||
self._sync_thread: Optional[threading.Thread] = None
|
||
|
||
# B1: recall_mode — set during initialize from config
|
||
self._recall_mode = "hybrid" # "context", "tools", or "hybrid"
|
||
|
||
# Base context cache — refreshed on context_cadence, not frozen
|
||
self._base_context_cache: Optional[str] = None
|
||
self._base_context_lock = threading.Lock()
|
||
|
||
# B5: Cost-awareness turn counting and cadence
|
||
self._turn_count = 0
|
||
self._injection_frequency = "every-turn" # or "first-turn"
|
||
self._context_cadence = 1 # minimum turns between context API calls
|
||
self._dialectic_cadence = 1 # backwards-compat fallback; wizard writes 2 on new configs
|
||
self._dialectic_depth = 1 # how many .chat() calls per dialectic cycle (1-3)
|
||
self._dialectic_depth_levels: list[str] | None = None # per-pass reasoning levels
|
||
self._reasoning_heuristic: bool = True # scale base level by query length
|
||
self._reasoning_level_cap: str = "high" # ceiling for auto-selected level
|
||
self._last_context_turn = -999
|
||
self._last_dialectic_turn = -999
|
||
|
||
# Liveness + observability state
|
||
self._prefetch_thread_started_at: float = 0.0 # monotonic ts of current thread
|
||
self._prefetch_result_fired_at: int = -999 # turn the pending result was fired at
|
||
self._dialectic_empty_streak: int = 0 # consecutive empty returns
|
||
|
||
# Port #1957: lazy session init for tools-only mode
|
||
self._session_initialized = False
|
||
self._lazy_init_kwargs: Optional[dict] = None
|
||
self._lazy_init_session_id: Optional[str] = None
|
||
self._init_thread: Optional[threading.Thread] = None
|
||
self._init_lock = threading.Lock()
|
||
self._init_error = ""
|
||
|
||
# Port #4053: cron guard — when True, plugin is fully inactive
|
||
self._cron_skipped = False
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
return "honcho"
|
||
|
||
def is_available(self) -> bool:
|
||
"""Check if Honcho is configured. No network calls."""
|
||
try:
|
||
from plugins.memory.honcho.client import HonchoClientConfig
|
||
cfg = HonchoClientConfig.from_global_config()
|
||
# Port #2645: baseUrl-only verification — api_key OR base_url suffices
|
||
return cfg.enabled and bool(cfg.api_key or cfg.base_url)
|
||
except Exception:
|
||
return False
|
||
|
||
def save_config(self, values, hermes_home):
|
||
"""Write config to $HERMES_HOME/honcho.json (Honcho SDK native format)."""
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
config_path = Path(hermes_home) / "honcho.json"
|
||
existing = {}
|
||
if config_path.exists():
|
||
try:
|
||
existing = json.loads(config_path.read_text())
|
||
except Exception:
|
||
pass
|
||
existing.update(values)
|
||
from utils import atomic_json_write
|
||
atomic_json_write(config_path, existing, mode=0o600)
|
||
|
||
def get_config_schema(self):
|
||
return [
|
||
{"key": "api_key", "description": "Honcho API key", "secret": True, "env_var": "HONCHO_API_KEY", "url": "https://app.honcho.dev"},
|
||
{"key": "baseUrl", "description": "Honcho base URL (for self-hosted)"},
|
||
]
|
||
|
||
def post_setup(self, hermes_home: str, config: dict) -> None:
|
||
"""Run the full Honcho setup wizard after provider selection."""
|
||
import types
|
||
from plugins.memory.honcho.cli import cmd_setup
|
||
cmd_setup(types.SimpleNamespace())
|
||
|
||
def initialize(self, session_id: str, **kwargs) -> None:
|
||
"""Initialize Honcho session manager.
|
||
|
||
Handles: cron guard, recall_mode, session name resolution,
|
||
peer memory mode, SOUL.md ai_peer sync, memory file migration,
|
||
and pre-warming context at init.
|
||
"""
|
||
try:
|
||
# ----- Port #4053: cron guard -----
|
||
agent_context = kwargs.get("agent_context", "")
|
||
platform = kwargs.get("platform", "cli")
|
||
if agent_context in {"cron", "flush"} or platform == "cron":
|
||
logger.debug("Honcho skipped: cron/flush context (agent_context=%s, platform=%s)",
|
||
agent_context, platform)
|
||
self._cron_skipped = True
|
||
return
|
||
|
||
from plugins.memory.honcho.client import HonchoClientConfig, get_honcho_client
|
||
from plugins.memory.honcho.session import HonchoSessionManager
|
||
|
||
cfg = HonchoClientConfig.from_global_config()
|
||
if not cfg.enabled or not (cfg.api_key or cfg.base_url):
|
||
logger.debug("Honcho not configured — plugin inactive")
|
||
return
|
||
|
||
self._config = cfg
|
||
|
||
# ----- B1: recall_mode from config -----
|
||
self._recall_mode = cfg.recall_mode # "context", "tools", or "hybrid"
|
||
logger.debug("Honcho recall_mode: %s", self._recall_mode)
|
||
|
||
# ----- B5: cost-awareness config -----
|
||
# All three cadence fields now have proper typed dataclass fields
|
||
# with host-block-first resolution (client.py).
|
||
self._injection_frequency = cfg.injection_frequency
|
||
self._context_cadence = cfg.context_cadence
|
||
self._dialectic_cadence = cfg.dialectic_cadence
|
||
self._dialectic_depth = max(1, min(cfg.dialectic_depth, 3))
|
||
self._dialectic_depth_levels = cfg.dialectic_depth_levels
|
||
self._reasoning_heuristic = cfg.reasoning_heuristic
|
||
if cfg.reasoning_level_cap in self._LEVEL_ORDER:
|
||
self._reasoning_level_cap = cfg.reasoning_level_cap
|
||
|
||
# aiPeer comes from honcho.json (host block or root) only.
|
||
# SOUL.md is persona content, not identity config.
|
||
|
||
self._lazy_init_kwargs = dict(kwargs)
|
||
self._lazy_init_session_id = session_id
|
||
self._session_key = self._resolve_session_key(cfg, session_id, **kwargs)
|
||
|
||
# Network-backed session creation can block on Honcho service or DB
|
||
# outages. Startup must fail open for context/hybrid modes, where
|
||
# Honcho is initialized only to enrich prompts. Tools-only mode has
|
||
# an explicit contract: init_on_session_start=False stays lazy until
|
||
# the first tool call, while init_on_session_start=True remains an
|
||
# eager, ready-on-return initialization path.
|
||
if self._recall_mode == "tools":
|
||
if cfg.init_on_session_start:
|
||
self._ensure_session()
|
||
return
|
||
logger.debug("Honcho tools-only mode — deferring session init until first tool call")
|
||
return
|
||
|
||
self._start_session_init_background(wait_timeout=0.1)
|
||
|
||
except ImportError:
|
||
logger.debug("honcho-ai package not installed — plugin inactive")
|
||
except Exception as e:
|
||
logger.warning("Honcho init failed: %s", e)
|
||
self._manager = None
|
||
|
||
def _resolve_session_key(self, cfg, session_id: str, **kwargs) -> str:
|
||
"""Resolve the Honcho session key without touching the network."""
|
||
session_title = kwargs.get("session_title")
|
||
gateway_session_key = kwargs.get("gateway_session_key")
|
||
return (
|
||
cfg.resolve_session_name(
|
||
session_title=session_title,
|
||
session_id=session_id,
|
||
gateway_session_key=gateway_session_key,
|
||
)
|
||
or session_id
|
||
or "hermes-default"
|
||
)
|
||
|
||
def _start_session_init_background(self, *, wait_timeout: float = 0.0) -> None:
|
||
"""Start Honcho session initialization in a daemon thread.
|
||
|
||
This keeps Hermes CLI/gateway startup responsive when Honcho is down,
|
||
slow, or its database is unhealthy. The thread may still take the SDK
|
||
timeout path, but it cannot block agent construction or first prompt
|
||
assembly. ``wait_timeout`` lets fast/mock initializations finish before
|
||
returning while still failing open for slow backends.
|
||
"""
|
||
if self._cron_skipped or self._session_initialized:
|
||
return
|
||
if not self._config or self._lazy_init_kwargs is None:
|
||
return
|
||
|
||
with self._init_lock:
|
||
if self._cron_skipped or self._session_initialized:
|
||
return
|
||
if self._init_thread and self._init_thread.is_alive():
|
||
return
|
||
if not self._config or self._lazy_init_kwargs is None:
|
||
return
|
||
|
||
cfg = self._config
|
||
init_kwargs = dict(self._lazy_init_kwargs)
|
||
init_session_id = self._lazy_init_session_id or "hermes-default"
|
||
|
||
def _run() -> None:
|
||
try:
|
||
self._do_session_init(cfg, init_session_id, **init_kwargs)
|
||
self._lazy_init_kwargs = None
|
||
self._lazy_init_session_id = None
|
||
self._init_error = ""
|
||
except Exception as e:
|
||
self._init_error = str(e)
|
||
self._manager = None
|
||
logger.warning("Honcho background session init failed: %s", e)
|
||
|
||
self._init_thread = threading.Thread(
|
||
target=_run,
|
||
daemon=True,
|
||
name="honcho-session-init",
|
||
)
|
||
self._init_thread.start()
|
||
if wait_timeout > 0:
|
||
self._init_thread.join(timeout=wait_timeout)
|
||
|
||
def _do_session_init(self, cfg, session_id: str, **kwargs) -> None:
|
||
"""Shared session initialization logic for both eager and lazy paths."""
|
||
from plugins.memory.honcho.client import get_honcho_client
|
||
from plugins.memory.honcho.session import HonchoSessionManager
|
||
|
||
client = get_honcho_client(cfg)
|
||
self._manager = HonchoSessionManager(
|
||
honcho=client,
|
||
config=cfg,
|
||
context_tokens=cfg.context_tokens,
|
||
runtime_user_peer_name=kwargs.get("user_id") or None,
|
||
runtime_user_peer_name_alt=kwargs.get("user_id_alt") or None,
|
||
)
|
||
|
||
# ----- B3: resolve_session_name -----
|
||
self._session_key = self._resolve_session_key(cfg, session_id, **kwargs)
|
||
logger.debug("Honcho session key resolved: %s", self._session_key)
|
||
|
||
# Create the remote session before running startup-only migration and
|
||
# prewarm work. Do not mark the provider ready until this method's
|
||
# synchronous setup has finished; background startup sets _manager before
|
||
# get_or_create()/migration/prewarm are complete, and lifecycle hooks must
|
||
# not treat that partially initialized state as usable.
|
||
session = self._manager.get_or_create(self._session_key)
|
||
|
||
# ----- B6: Memory file migration (one-time, for new sessions) -----
|
||
# Skip under per-session strategy: every Hermes run creates a fresh
|
||
# Honcho session by design, so uploading MEMORY.md/USER.md/SOUL.md to
|
||
# each one would flood the backend with short-lived duplicates instead
|
||
# of performing a one-time migration.
|
||
try:
|
||
if not session.messages and cfg.session_strategy != "per-session":
|
||
from hermes_constants import get_hermes_home
|
||
mem_dir = str(get_hermes_home() / "memories")
|
||
self._manager.migrate_memory_files(self._session_key, mem_dir)
|
||
logger.debug("Honcho memory file migration attempted for new session: %s", self._session_key)
|
||
elif cfg.session_strategy == "per-session":
|
||
logger.debug(
|
||
"Honcho memory file migration skipped: per-session strategy creates a fresh session per run (%s)",
|
||
self._session_key,
|
||
)
|
||
except Exception as e:
|
||
logger.debug("Honcho memory file migration skipped: %s", e)
|
||
|
||
# ----- B7: Pre-warming at init -----
|
||
# Context prewarm warms peer.context() (base layer), consumed via
|
||
# pop_context_result() in prefetch(). A generic dialectic prewarm is
|
||
# retained only for providers without latest-message rewriting; otherwise
|
||
# it would shadow the real first user message and recreate #36017.
|
||
if self._recall_mode in {"context", "hybrid"}:
|
||
try:
|
||
self._manager.prefetch_context(self._session_key)
|
||
except Exception as e:
|
||
logger.debug("Honcho context prewarm failed: %s", e)
|
||
|
||
if self._query_rewriter is None:
|
||
_prewarm_query = (
|
||
"Summarize what you know about this user. "
|
||
"Focus on preferences, current projects, and working style."
|
||
)
|
||
|
||
def _prewarm_dialectic() -> None:
|
||
try:
|
||
r = self._run_dialectic_depth(
|
||
_prewarm_query, use_query_rewrite=False
|
||
)
|
||
except Exception as exc:
|
||
logger.debug("Honcho dialectic prewarm failed: %s", exc)
|
||
self._dialectic_empty_streak += 1
|
||
return
|
||
if r and r.strip():
|
||
with self._prefetch_lock:
|
||
self._prefetch_result = r
|
||
self._prefetch_result_fired_at = 0
|
||
# Treat prewarm as turn 0 so cadence gating starts clean.
|
||
self._last_dialectic_turn = 0
|
||
self._dialectic_empty_streak = 0
|
||
else:
|
||
self._dialectic_empty_streak += 1
|
||
|
||
self._prefetch_thread_started_at = time.monotonic()
|
||
prewarm_thread = threading.Thread(
|
||
target=_prewarm_dialectic,
|
||
daemon=True,
|
||
name="honcho-prewarm-dialectic",
|
||
)
|
||
prewarm_thread.start()
|
||
self._prefetch_thread = prewarm_thread
|
||
else:
|
||
logger.debug(
|
||
"Honcho generic dialectic prewarm skipped: awaiting first user message"
|
||
)
|
||
logger.debug("Honcho pre-warm started for session: %s", self._session_key)
|
||
|
||
self._session_initialized = True
|
||
|
||
def _ensure_session(self) -> bool:
|
||
"""Lazily initialize the Honcho session (for tools-only mode).
|
||
|
||
Returns True if the manager is ready, False otherwise.
|
||
"""
|
||
if self._manager and self._session_initialized:
|
||
return True
|
||
if self._cron_skipped:
|
||
return False
|
||
if self._init_thread and self._init_thread.is_alive():
|
||
return False
|
||
if not self._config or self._lazy_init_kwargs is None:
|
||
return False
|
||
|
||
try:
|
||
self._do_session_init(
|
||
self._config,
|
||
self._lazy_init_session_id or "hermes-default",
|
||
**self._lazy_init_kwargs,
|
||
)
|
||
# Clear lazy refs
|
||
self._lazy_init_kwargs = None
|
||
self._lazy_init_session_id = None
|
||
return self._manager is not None
|
||
except Exception as e:
|
||
self._manager = None
|
||
self._session_initialized = False
|
||
logger.warning("Honcho lazy session init failed: %s", e)
|
||
return False
|
||
|
||
def _session_ready(self) -> bool:
|
||
"""Return whether a manager/session key can be used safely.
|
||
|
||
Background initialization sets ``_manager`` before the blocking
|
||
get-or-create call completes, so ``_session_initialized`` guards real
|
||
async startup. Tests and legacy direct construction may inject a ready
|
||
manager/session key without setting that flag; allow that only when no
|
||
init thread is currently in flight.
|
||
"""
|
||
if not self._manager or not self._session_key:
|
||
return False
|
||
if self._session_initialized:
|
||
return True
|
||
return not (self._init_thread and self._init_thread.is_alive())
|
||
|
||
def _format_first_turn_context(self, ctx: dict) -> str:
|
||
"""Format the prefetch context dict into a readable system prompt block."""
|
||
parts = []
|
||
|
||
# Session summary — session-scoped context, placed first for relevance
|
||
summary = ctx.get("summary", "")
|
||
if summary:
|
||
parts.append(f"## Session Summary\n{summary}")
|
||
|
||
rep = ctx.get("representation", "")
|
||
if rep:
|
||
parts.append(f"## User Representation\n{rep}")
|
||
|
||
card = ctx.get("card", "")
|
||
if card:
|
||
parts.append(f"## User Peer Card\n{card}")
|
||
|
||
ai_rep = ctx.get("ai_representation", "")
|
||
if ai_rep:
|
||
parts.append(f"## AI Self-Representation\n{ai_rep}")
|
||
|
||
ai_card = ctx.get("ai_card", "")
|
||
if ai_card:
|
||
parts.append(f"## AI Identity Card\n{ai_card}")
|
||
|
||
if not parts:
|
||
return ""
|
||
return "\n\n".join(parts)
|
||
|
||
def system_prompt_block(self) -> str:
|
||
"""Return system prompt text, adapted by recall_mode.
|
||
|
||
Returns only the mode header and tool instructions — static text
|
||
that doesn't change between turns (prompt-cache friendly).
|
||
Live context (representation, card) is injected via prefetch().
|
||
"""
|
||
if self._cron_skipped:
|
||
return ""
|
||
if not self._manager or not self._session_key:
|
||
if not self._config:
|
||
return ""
|
||
|
||
# ----- B1: adapt text based on recall_mode -----
|
||
if self._recall_mode == "context":
|
||
header = (
|
||
"# Honcho Memory\n"
|
||
"Active (context-injection mode). Relevant user context is automatically "
|
||
"injected before each turn. No memory tools are available — context is "
|
||
"managed automatically."
|
||
)
|
||
elif self._recall_mode == "tools":
|
||
header = (
|
||
"# Honcho Memory\n"
|
||
"Active (tools-only mode). Use honcho_profile for a quick factual snapshot, "
|
||
"honcho_search for raw excerpts, honcho_context for raw peer context, "
|
||
"honcho_reasoning for synthesized answers (pass reasoning_level "
|
||
"minimal/low/medium/high/max — you pick the depth per call), "
|
||
"honcho_conclude to save facts about the user. "
|
||
"No automatic context injection — you must use tools to access memory."
|
||
)
|
||
else: # hybrid
|
||
header = (
|
||
"# Honcho Memory\n"
|
||
"Active (hybrid mode). Relevant context is auto-injected AND memory tools are available. "
|
||
"Use honcho_profile for a quick factual snapshot, "
|
||
"honcho_search for raw excerpts, honcho_context for raw peer context, "
|
||
"honcho_reasoning for synthesized answers (pass reasoning_level "
|
||
"minimal/low/medium/high/max — you pick the depth per call), "
|
||
"honcho_conclude to save facts about the user."
|
||
)
|
||
|
||
return header
|
||
|
||
def prefetch(self, query: str, *, session_id: str = "") -> str:
|
||
"""Return base context (representation + card) plus dialectic supplement.
|
||
|
||
Assembles two layers:
|
||
1. Base context from peer.context() — cached, refreshed on context_cadence
|
||
2. Dialectic supplement — cached, refreshed on dialectic_cadence
|
||
|
||
B1: Returns empty when recall_mode is "tools" (no injection).
|
||
B5: Respects injection_frequency — "first-turn" returns cached/empty after turn 0.
|
||
Port #3265: Truncates to context_tokens budget.
|
||
"""
|
||
if self._cron_skipped:
|
||
return ""
|
||
|
||
# B1: tools-only mode — no auto-injection
|
||
if self._recall_mode == "tools":
|
||
return ""
|
||
|
||
if not self._session_ready():
|
||
# Session init runs in a background daemon thread (kept off the
|
||
# startup path so a slow/unreachable Honcho can't block agent
|
||
# construction). On turn 1 of a fresh process that thread may not
|
||
# have finished yet. Rather than return empty immediately — which
|
||
# denied turn-1 context entirely whenever init crossed the ~100ms
|
||
# spawn window — wait a bounded time for it to complete so the
|
||
# peer card can still be injected on the first message.
|
||
self._start_session_init_background()
|
||
_init_thread = self._init_thread
|
||
if _init_thread is not None:
|
||
_init_thread.join(timeout=self._FIRST_TURN_BASE_TIMEOUT)
|
||
if not self._session_ready():
|
||
return ""
|
||
|
||
# B5: injection_frequency — if "first-turn" and past first turn, skip
|
||
# the base context layer (static representation + card). The dialectic
|
||
# supplement has its own cadence and must still be consumed below.
|
||
_skip_base = (
|
||
self._injection_frequency == "first-turn" and self._turn_count > 1
|
||
)
|
||
|
||
# Trivial prompts ("ok", "yes", slash commands) carry no semantic
|
||
# signal, so we don't fetch base context or fire new dialectic work for
|
||
# them. But a dialectic result fired at the END of the previous turn was
|
||
# primed for THIS turn — blindly returning here used to strand it
|
||
# (consumption happens further down), after which the stale-discard
|
||
# guard silently dropped a successfully-generated answer. Instead,
|
||
# drain any ready non-stale pending result and inject just that:
|
||
# trivial turns still spend no NEW work, but they no longer throw
|
||
# away work already paid for.
|
||
if self._is_trivial_prompt(query):
|
||
ready = self._consume_pending_dialectic()
|
||
return self._truncate_to_budget(ready) if ready else ""
|
||
|
||
parts = []
|
||
|
||
# ----- Layer 1: Base context (representation + card) -----
|
||
# Skipped when injectionFrequency is "first-turn" past turn 1 — the
|
||
# representation + card are static within a session and re-injecting
|
||
# them every turn is pure token waste.
|
||
if _skip_base:
|
||
# Still drain any pending dialectic result below; just don't
|
||
# fetch or inject the base context layer.
|
||
pass
|
||
else:
|
||
# Base context is pure retrieval (no LLM): representation + peer card +
|
||
# session summary. On the FIRST fetch (cache is None, i.e. turn 1 of a
|
||
# session) we fetch it SYNCHRONOUSLY with a short bound so the peer card
|
||
# is injected immediately. Firing it async and popping in the same call
|
||
# always lost the race on turn 1 (background thread can't finish inside
|
||
# one synchronous pass), which is why new sessions saw zero context
|
||
# until turn 2. Subsequent turns consume the background-refreshed result
|
||
# primed by queue_prefetch() at the end of the previous turn.
|
||
with self._base_context_lock:
|
||
_first_base_fetch = self._base_context_cache is None
|
||
if _first_base_fetch:
|
||
self._base_context_cache = ""
|
||
self._last_context_turn = self._turn_count
|
||
base_context = self._base_context_cache
|
||
|
||
if _first_base_fetch and self._manager:
|
||
# Synchronous, bounded fetch so turn 1 gets the peer card now.
|
||
_ctx_holder: dict[str, dict] = {}
|
||
|
||
def _fetch_base() -> None:
|
||
try:
|
||
ctx = self._manager.get_prefetch_context(
|
||
self._session_key, query or None
|
||
) or {}
|
||
_ctx_holder["ctx"] = ctx
|
||
if ctx:
|
||
self._manager.set_context_result(self._session_key, ctx)
|
||
except Exception as e:
|
||
logger.debug("Honcho first-turn base context failed: %s", e)
|
||
|
||
_bt = threading.Thread(
|
||
target=_fetch_base, daemon=True, name="honcho-base-first"
|
||
)
|
||
_bt.start()
|
||
# Bound the synchronous wait by the smaller of the base timeout and
|
||
# any configured request timeout, so a deliberately tight
|
||
# config.timeout (fail-fast deployments, tests) is still honored.
|
||
_base_wait = self._FIRST_TURN_BASE_TIMEOUT
|
||
if self._config and self._config.timeout:
|
||
_base_wait = min(_base_wait, self._config.timeout)
|
||
_bt.join(timeout=_base_wait)
|
||
_ctx = _ctx_holder.get("ctx")
|
||
if _ctx:
|
||
self._manager.pop_context_result(self._session_key)
|
||
formatted = self._format_first_turn_context(_ctx)
|
||
if formatted:
|
||
with self._base_context_lock:
|
||
self._base_context_cache = formatted
|
||
base_context = formatted
|
||
elif _bt.is_alive():
|
||
logger.debug(
|
||
"Honcho first-turn base context still running after %.1fs — "
|
||
"will surface on next turn", self._FIRST_TURN_BASE_TIMEOUT,
|
||
)
|
||
|
||
# Check if a background context prefetch (primed last turn) has a
|
||
# fresher result. On turn 1 this is normally empty; turn 2+ consumes
|
||
# the result queued by queue_prefetch().
|
||
if not _first_base_fetch and self._manager:
|
||
fresh_ctx = self._manager.pop_context_result(self._session_key)
|
||
if fresh_ctx:
|
||
formatted = self._format_first_turn_context(fresh_ctx)
|
||
if formatted:
|
||
with self._base_context_lock:
|
||
self._base_context_cache = formatted
|
||
base_context = formatted
|
||
|
||
if base_context:
|
||
parts.append(base_context)
|
||
|
||
# ----- Layer 2: Dialectic supplement -----
|
||
# On the very first turn, no queue_prefetch() has run yet so the
|
||
# dialectic result is empty. Run with a bounded timeout so a slow
|
||
# Honcho connection doesn't block the first response indefinitely.
|
||
# On timeout we let the thread keep running and write its result into
|
||
# _prefetch_result under the lock, so the next turn picks it up.
|
||
#
|
||
# Skip if the session-start prewarm already filled _prefetch_result —
|
||
# firing another .chat() would be duplicate work.
|
||
with self._prefetch_lock:
|
||
_prewarm_landed = bool(self._prefetch_result)
|
||
if _prewarm_landed and self._last_dialectic_turn == -999:
|
||
self._last_dialectic_turn = self._turn_count
|
||
|
||
_did_first_turn_wait = False
|
||
if self._last_dialectic_turn == -999 and query:
|
||
_did_first_turn_wait = True
|
||
# A dialectic prewarm thread is fired at session init with an
|
||
# equivalent query. If it's still running, do NOT fire a second
|
||
# .chat() here — that was duplicate work that also blocked the
|
||
# first response. Briefly wait for the in-flight prewarm to land;
|
||
# if it doesn't, let it surface on a later turn via _prefetch_result.
|
||
# First-turn dialectic wait. Decoupled from a LARGE config.timeout
|
||
# (a 60s host timeout must not block the first response for 60s),
|
||
# but still bounded by a TIGHT config.timeout when one is set
|
||
# (fail-fast deployments / tests). See _FIRST_TURN_DIALECTIC_CAP.
|
||
_dia_wait = self._FIRST_TURN_DIALECTIC_CAP
|
||
if self._config and self._config.timeout:
|
||
_dia_wait = min(_dia_wait, self._config.timeout)
|
||
if self._thread_is_live():
|
||
_live = self._prefetch_thread
|
||
if _live is not None:
|
||
_live.join(timeout=_dia_wait)
|
||
else:
|
||
_first_turn_timeout = _dia_wait
|
||
_fired_at = self._turn_count
|
||
|
||
def _run_first_turn() -> None:
|
||
try:
|
||
r = self._run_dialectic_depth(query)
|
||
except Exception as exc:
|
||
logger.debug("Honcho first-turn dialectic failed: %s", exc)
|
||
self._dialectic_empty_streak += 1
|
||
return
|
||
if r and r.strip():
|
||
with self._prefetch_lock:
|
||
self._prefetch_result = r
|
||
self._prefetch_result_fired_at = _fired_at
|
||
# Advance cadence only on a non-empty result so the next
|
||
# turn retries when the call returned nothing.
|
||
self._last_dialectic_turn = _fired_at
|
||
self._dialectic_empty_streak = 0
|
||
else:
|
||
self._dialectic_empty_streak += 1
|
||
|
||
self._prefetch_thread_started_at = time.monotonic()
|
||
first_turn_thread = threading.Thread(
|
||
target=_run_first_turn, daemon=True, name="honcho-prefetch-first"
|
||
)
|
||
first_turn_thread.start()
|
||
self._prefetch_thread = first_turn_thread
|
||
self._prefetch_thread.join(timeout=_first_turn_timeout)
|
||
if self._prefetch_thread and self._prefetch_thread.is_alive():
|
||
logger.debug(
|
||
"Honcho first-turn dialectic still running after %.1fs — "
|
||
"will surface on next turn",
|
||
self._FIRST_TURN_DIALECTIC_CAP,
|
||
)
|
||
|
||
# Turn 2+ consumes the result primed by queue_prefetch() last turn —
|
||
# give a near-finished thread a brief moment. On turn 1 we already
|
||
# waited above (_did_first_turn_wait); waiting again here just
|
||
# re-blocks on the same slow in-flight dialectic with no chance of it
|
||
# landing, so skip the redundant join.
|
||
if (not _did_first_turn_wait and self._prefetch_thread
|
||
and self._prefetch_thread.is_alive()):
|
||
self._prefetch_thread.join(timeout=3.0)
|
||
|
||
# Drain the pending result (applies the stale-discard guard). Shared
|
||
# with the trivial-prompt path above via _consume_pending_dialectic so
|
||
# both routes pop + age-check identically.
|
||
dialectic_result = self._consume_pending_dialectic()
|
||
|
||
if dialectic_result and dialectic_result.strip():
|
||
parts.append(dialectic_result)
|
||
|
||
if not parts:
|
||
return ""
|
||
|
||
result = "\n\n".join(parts)
|
||
|
||
# ----- Port #3265: token budget enforcement -----
|
||
result = self._truncate_to_budget(result)
|
||
|
||
return result
|
||
|
||
def _consume_pending_dialectic(self) -> str:
|
||
"""Pop any pending dialectic result, applying the stale-discard guard.
|
||
|
||
Drains _prefetch_result under the lock and returns it unless it is
|
||
stale (fired more than cadence × multiplier turns ago), in which case
|
||
it is dropped and "" is returned. Does NOT wait on an in-flight thread
|
||
— callers that want to give a near-finished thread a moment must join
|
||
before calling. Idempotent: a second call returns "" until a new
|
||
result is primed.
|
||
"""
|
||
with self._prefetch_lock:
|
||
dialectic_result = self._prefetch_result
|
||
fired_at = self._prefetch_result_fired_at
|
||
self._prefetch_result = ""
|
||
self._prefetch_result_fired_at = -999
|
||
|
||
# Discard stale pending results: if the fire happened more than
|
||
# cadence × multiplier turns ago (e.g. a run of trivial-prompt turns
|
||
# passed without consumption), the content likely no longer tracks
|
||
# the current conversational pivot.
|
||
stale_limit = self._dialectic_cadence * self._STALE_RESULT_MULTIPLIER
|
||
if dialectic_result and fired_at >= 0 and (self._turn_count - fired_at) > stale_limit:
|
||
logger.debug(
|
||
"Honcho pending dialectic discarded as stale: fired_at=%d, "
|
||
"turn=%d, limit=%d", fired_at, self._turn_count, stale_limit,
|
||
)
|
||
return ""
|
||
return dialectic_result if (dialectic_result and dialectic_result.strip()) else ""
|
||
|
||
def _truncate_to_budget(self, text: str) -> str:
|
||
"""Truncate text to fit within context_tokens budget if set."""
|
||
if not self._config or not self._config.context_tokens:
|
||
return text
|
||
budget_chars = self._config.context_tokens * 4 # conservative char estimate
|
||
if len(text) <= budget_chars:
|
||
return text
|
||
# Truncate at word boundary
|
||
truncated = text[:budget_chars]
|
||
last_space = truncated.rfind(" ")
|
||
if last_space > budget_chars * 0.8:
|
||
truncated = truncated[:last_space]
|
||
return truncated + " …"
|
||
|
||
def queue_prefetch(self, query: str, *, session_id: str = "") -> None:
|
||
"""Fire background prefetch threads for the upcoming turn.
|
||
|
||
B5: Checks cadence independently for dialectic and context refresh.
|
||
Context refresh updates the base layer (representation + card).
|
||
Dialectic fires the LLM reasoning supplement.
|
||
"""
|
||
if self._cron_skipped:
|
||
return
|
||
# B1: tools-only mode — no prefetch
|
||
if self._recall_mode == "tools":
|
||
return
|
||
|
||
if not self._session_ready() or not query:
|
||
self._start_session_init_background()
|
||
return
|
||
|
||
# Trivial prompts don't warrant either a context refresh or a dialectic call.
|
||
if self._is_trivial_prompt(query):
|
||
return
|
||
|
||
# ----- Context refresh (base layer) — independent cadence -----
|
||
if self._context_cadence <= 1 or (self._turn_count - self._last_context_turn) >= self._context_cadence:
|
||
self._last_context_turn = self._turn_count
|
||
try:
|
||
self._manager.prefetch_context(self._session_key, query)
|
||
except Exception as e:
|
||
logger.debug("Honcho context prefetch failed: %s", e)
|
||
|
||
# ----- Dialectic prefetch (supplement layer) -----
|
||
# Thread-alive guard with stale-thread recovery: a hung Honcho call
|
||
# older than timeout × multiplier is treated as dead so it can't
|
||
# block subsequent fires.
|
||
if self._thread_is_live():
|
||
logger.debug("Honcho dialectic prefetch skipped: prior thread still running")
|
||
return
|
||
|
||
# Cadence gate, widened by the empty-streak backoff so a persistently
|
||
# silent backend doesn't retry every turn forever.
|
||
effective = self._effective_cadence()
|
||
if (self._turn_count - self._last_dialectic_turn) < effective:
|
||
logger.debug(
|
||
"Honcho dialectic prefetch skipped: effective cadence %d "
|
||
"(base %d, empty streak %d), turns since last: %d",
|
||
effective, self._dialectic_cadence, self._dialectic_empty_streak,
|
||
self._turn_count - self._last_dialectic_turn,
|
||
)
|
||
return
|
||
|
||
# Cadence advances only on a non-empty result so empty returns
|
||
# (transient API error, sparse representation) retry next turn.
|
||
_fired_at = self._turn_count
|
||
|
||
def _run():
|
||
try:
|
||
result = self._run_dialectic_depth(query)
|
||
except Exception as e:
|
||
logger.debug("Honcho prefetch failed: %s", e)
|
||
self._dialectic_empty_streak += 1
|
||
return
|
||
if result and result.strip():
|
||
with self._prefetch_lock:
|
||
self._prefetch_result = result
|
||
self._prefetch_result_fired_at = _fired_at
|
||
self._last_dialectic_turn = _fired_at
|
||
self._dialectic_empty_streak = 0
|
||
else:
|
||
self._dialectic_empty_streak += 1
|
||
|
||
self._prefetch_thread_started_at = time.monotonic()
|
||
prefetch_thread = threading.Thread(
|
||
target=_run, daemon=True, name="honcho-prefetch"
|
||
)
|
||
prefetch_thread.start()
|
||
self._prefetch_thread = prefetch_thread
|
||
|
||
# ----- Dialectic depth: multi-pass .chat() with cold/warm prompts -----
|
||
|
||
# Proportional reasoning levels per depth/pass when dialecticDepthLevels
|
||
# is not configured. The base level is dialecticReasoningLevel.
|
||
# Index: (depth, pass) → level relative to base.
|
||
_PROPORTIONAL_LEVELS: dict[tuple[int, int], str] = {
|
||
# depth 1: single pass at base level
|
||
(1, 0): "base",
|
||
# depth 2: pass 0 lighter, pass 1 at base
|
||
(2, 0): "minimal",
|
||
(2, 1): "base",
|
||
# depth 3: pass 0 lighter, pass 1 at base, pass 2 one above minimal
|
||
(3, 0): "minimal",
|
||
(3, 1): "base",
|
||
(3, 2): "low",
|
||
}
|
||
|
||
_LEVEL_ORDER = ("minimal", "low", "medium", "high", "max")
|
||
|
||
# Char-count thresholds for the query-length reasoning heuristic.
|
||
_HEURISTIC_LENGTH_MEDIUM = 120
|
||
_HEURISTIC_LENGTH_HIGH = 400
|
||
|
||
# Liveness constants. A thread older than timeout × multiplier is treated
|
||
# as dead so a hung Honcho call can't block future retries indefinitely.
|
||
_STALE_THREAD_MULTIPLIER = 2.0
|
||
# Pending result whose fire-turn is older than cadence × multiplier is
|
||
# discarded on read so we don't inject context for a stale conversational
|
||
# pivot after a gap of trivial-prompt turns.
|
||
_STALE_RESULT_MULTIPLIER = 2
|
||
# Cap on the empty-streak backoff so a persistently silent backend
|
||
# eventually settles on a ceiling instead of unbounded widening.
|
||
_BACKOFF_MAX = 8
|
||
# First-turn base-context fetch bound. Base context (representation +
|
||
# peer card) is pure retrieval with no LLM call (~400-600ms locally), so
|
||
# turn 1 can afford to wait for it synchronously and inject the peer card
|
||
# immediately instead of returning an empty block that only fills from
|
||
# turn 2. A short bound keeps a slow/unreachable backend from blocking the
|
||
# first response.
|
||
_FIRST_TURN_BASE_TIMEOUT = 3.0
|
||
# First-turn dialectic grace. The dialectic is the slow LLM path
|
||
# (multi-pass .chat(), often 20s+ at depth 3) and must stay decoupled from
|
||
# config.timeout: once the host-block timeout is honored (e.g. 60s),
|
||
# reusing it here would block the first response for the full request
|
||
# timeout. Turn 1's value is the base context (peer card), fetched
|
||
# synchronously above; the dialectic only needs an opportunistic grace to
|
||
# catch a near-finished prewarm. If it doesn't land in this window it
|
||
# surfaces on turn 2 via _prefetch_result — no benefit to blocking longer.
|
||
_FIRST_TURN_DIALECTIC_CAP = 2.0
|
||
|
||
def _thread_is_live(self) -> bool:
|
||
"""Thread-alive guard that treats threads older than the stale
|
||
threshold as dead, so a hung Honcho request can't block new fires."""
|
||
if not self._prefetch_thread or not self._prefetch_thread.is_alive():
|
||
return False
|
||
timeout = (self._config.timeout if self._config and self._config.timeout else 8.0)
|
||
age = time.monotonic() - self._prefetch_thread_started_at
|
||
if age > timeout * self._STALE_THREAD_MULTIPLIER:
|
||
logger.debug(
|
||
"Honcho prefetch thread age %.1fs exceeds stale threshold "
|
||
"%.1fs — treating as dead", age, timeout * self._STALE_THREAD_MULTIPLIER,
|
||
)
|
||
return False
|
||
return True
|
||
|
||
def _effective_cadence(self) -> int:
|
||
"""Cadence plus empty-streak backoff, capped at _BACKOFF_MAX × base."""
|
||
if self._dialectic_empty_streak <= 0:
|
||
return self._dialectic_cadence
|
||
widened = self._dialectic_cadence + self._dialectic_empty_streak
|
||
ceiling = self._dialectic_cadence * self._BACKOFF_MAX
|
||
return min(widened, ceiling)
|
||
|
||
def liveness_snapshot(self) -> dict:
|
||
"""In-process snapshot of dialectic liveness state for diagnostics.
|
||
|
||
Returns current turn, last successful dialectic turn, pending-result
|
||
fire turn, empty streak, effective cadence, and thread status.
|
||
"""
|
||
thread_age = None
|
||
if self._prefetch_thread and self._prefetch_thread.is_alive():
|
||
thread_age = time.monotonic() - self._prefetch_thread_started_at
|
||
return {
|
||
"turn_count": self._turn_count,
|
||
"last_dialectic_turn": self._last_dialectic_turn,
|
||
"pending_result_fired_at": self._prefetch_result_fired_at,
|
||
"empty_streak": self._dialectic_empty_streak,
|
||
"effective_cadence": self._effective_cadence(),
|
||
"thread_alive": thread_age is not None,
|
||
"thread_age_seconds": thread_age,
|
||
}
|
||
|
||
def _apply_reasoning_heuristic(self, base: str, query: str) -> str:
|
||
"""Scale `base` up by query length, clamped at reasoning_level_cap.
|
||
|
||
Char-count heuristic: +1 at >=120 chars, +2 at >=400.
|
||
"""
|
||
if not self._reasoning_heuristic or not query:
|
||
return base
|
||
if base not in self._LEVEL_ORDER:
|
||
return base
|
||
n = len(query)
|
||
if n < self._HEURISTIC_LENGTH_MEDIUM:
|
||
bump = 0
|
||
elif n < self._HEURISTIC_LENGTH_HIGH:
|
||
bump = 1
|
||
else:
|
||
bump = 2
|
||
base_idx = self._LEVEL_ORDER.index(base)
|
||
cap_idx = self._LEVEL_ORDER.index(self._reasoning_level_cap)
|
||
return self._LEVEL_ORDER[min(base_idx + bump, cap_idx)]
|
||
|
||
def _resolve_pass_level(self, pass_idx: int, query: str = "") -> str:
|
||
"""Resolve reasoning level for a given pass index.
|
||
|
||
Precedence:
|
||
1. dialecticDepthLevels (explicit per-pass) — wins absolutely
|
||
2. _PROPORTIONAL_LEVELS table (depth>1 lighter-early passes)
|
||
3. Base level = dialecticReasoningLevel, optionally scaled by the
|
||
reasoning heuristic when the mapping falls through to 'base'
|
||
"""
|
||
if self._dialectic_depth_levels and pass_idx < len(self._dialectic_depth_levels):
|
||
return self._dialectic_depth_levels[pass_idx]
|
||
|
||
base = (self._config.dialectic_reasoning_level if self._config else "low")
|
||
mapping = self._PROPORTIONAL_LEVELS.get((self._dialectic_depth, pass_idx))
|
||
if mapping is None or mapping == "base":
|
||
return self._apply_reasoning_heuristic(base, query)
|
||
return mapping
|
||
|
||
def _build_dialectic_prompt(self, pass_idx: int, prior_results: list[str], is_cold: bool) -> str:
|
||
"""Build the prompt for a given dialectic pass.
|
||
|
||
Pass 0: cold start (general user query) or warm (session-scoped).
|
||
Pass 1: self-audit / targeted synthesis against gaps from pass 0.
|
||
Pass 2: reconciliation / contradiction check across prior passes.
|
||
"""
|
||
if pass_idx == 0:
|
||
if is_cold:
|
||
return (
|
||
"Who is this person? What are their preferences, goals, "
|
||
"and working style? Focus on facts that would help an AI "
|
||
"assistant be immediately useful."
|
||
)
|
||
return (
|
||
"Given what's been discussed in this session so far, what "
|
||
"context about this user is most relevant to the current "
|
||
"conversation? Prioritize active context over biographical facts."
|
||
)
|
||
elif pass_idx == 1:
|
||
prior = prior_results[-1] if prior_results else ""
|
||
return (
|
||
f"Given this initial assessment:\n\n{prior}\n\n"
|
||
"What gaps remain in your understanding that would help "
|
||
"going forward? Synthesize what you actually know about "
|
||
"the user's current state and immediate needs, grounded "
|
||
"in evidence from recent sessions."
|
||
)
|
||
else:
|
||
# pass 2: reconciliation
|
||
return (
|
||
f"Prior passes produced:\n\n"
|
||
f"Pass 1:\n{prior_results[0] if len(prior_results) > 0 else '(empty)'}\n\n"
|
||
f"Pass 2:\n{prior_results[1] if len(prior_results) > 1 else '(empty)'}\n\n"
|
||
"Do these assessments cohere? Reconcile any contradictions "
|
||
"and produce a final, concise synthesis of what matters most "
|
||
"for the current conversation."
|
||
)
|
||
|
||
@staticmethod
|
||
def _signal_sufficient(result: str) -> bool:
|
||
"""Check if a dialectic pass returned enough signal to skip further passes.
|
||
|
||
Heuristic: a response longer than 100 chars with some structure
|
||
(section headers, bullets, or an ordered list) is considered sufficient.
|
||
"""
|
||
if not result or len(result.strip()) < 100:
|
||
return False
|
||
# Structured output with sections/bullets is strong signal
|
||
if "\n" in result and (
|
||
"##" in result
|
||
or "•" in result
|
||
or re.search(r"^[*-] ", result, re.MULTILINE)
|
||
or re.search(r"^\s*\d+\. ", result, re.MULTILINE)
|
||
):
|
||
return True
|
||
# Long enough even without structure
|
||
return len(result.strip()) > 300
|
||
|
||
def _run_dialectic_depth(self, query: str, *, use_query_rewrite: bool = True) -> str:
|
||
"""Execute up to dialecticDepth .chat() calls with conditional bail-out.
|
||
|
||
Cold start (no base context): general user-oriented query.
|
||
Warm session (base context exists): session-scoped query.
|
||
Each pass is conditional — bails early if prior pass returned strong signal.
|
||
Returns the best (usually last) result.
|
||
"""
|
||
if not self._manager or not self._session_key:
|
||
return ""
|
||
|
||
is_cold = not self._base_context_cache
|
||
results: list[str] = []
|
||
rewritten_query = ""
|
||
if use_query_rewrite and self._query_rewriter:
|
||
try:
|
||
rewritten_query = self._query_rewriter(query).strip()
|
||
except Exception as exc:
|
||
logger.debug("Honcho query rewriter failed: %s", exc)
|
||
|
||
for i in range(self._dialectic_depth):
|
||
# Only non-empty prior results are usable context for dependent
|
||
# passes. A pass can return "" (e.g. a reasoning model that spends
|
||
# its whole token budget thinking and emits no content). Feeding
|
||
# that blank forward produced prompts like "Given this initial
|
||
# assessment:\n\n\n\nWhat gaps remain..." with an empty body — the
|
||
# empty-spot symptom seen in Honcho request logs.
|
||
prior_results = [r for r in results if r and r.strip()]
|
||
if i == 0:
|
||
prompt = rewritten_query or self._build_dialectic_prompt(
|
||
0, prior_results, is_cold
|
||
)
|
||
else:
|
||
# Skip further passes if prior pass delivered strong signal
|
||
if prior_results and self._signal_sufficient(prior_results[-1]):
|
||
logger.debug("Honcho dialectic depth %d: pass %d skipped, prior signal sufficient",
|
||
self._dialectic_depth, i)
|
||
break
|
||
if not prior_results:
|
||
# Every prior pass returned empty — a dependent prompt would
|
||
# carry a blank assessment. Re-issue the pass-0 base prompt
|
||
# instead so this pass starts fresh rather than referencing
|
||
# nothing.
|
||
logger.debug("Honcho dialectic depth %d: pass %d has no non-empty prior — "
|
||
"falling back to base prompt", self._dialectic_depth, i)
|
||
prompt = rewritten_query or self._build_dialectic_prompt(
|
||
0, prior_results, is_cold
|
||
)
|
||
else:
|
||
prompt = self._build_dialectic_prompt(i, prior_results, is_cold)
|
||
|
||
level = self._resolve_pass_level(i, query=query)
|
||
logger.debug("Honcho dialectic depth %d: pass %d, level=%s, cold=%s",
|
||
self._dialectic_depth, i, level, is_cold)
|
||
|
||
result = self._manager.dialectic_query(
|
||
self._session_key, prompt,
|
||
reasoning_level=level,
|
||
peer="user",
|
||
)
|
||
results.append(result or "")
|
||
|
||
# Return the last non-empty result (deepest pass that ran)
|
||
for r in reversed(results):
|
||
if r and r.strip():
|
||
return r
|
||
return ""
|
||
|
||
# Prompts that carry no semantic signal — trivial acknowledgements, slash
|
||
# commands, empty input. Skipping injection here saves tokens and prevents
|
||
# stale user-model context from derailing one-word replies.
|
||
_TRIVIAL_PROMPT_RE = re.compile(
|
||
r'^(yes|no|ok|okay|sure|thanks|thank you|y|n|yep|nope|yeah|nah|'
|
||
r'continue|go ahead|do it|proceed|got it|cool|nice|great|done|next|lgtm|k)$',
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
@classmethod
|
||
def _is_trivial_prompt(cls, text: str) -> bool:
|
||
"""Return True if the prompt is too trivial to warrant context injection."""
|
||
if not text:
|
||
return True
|
||
stripped = text.strip()
|
||
if not stripped:
|
||
return True
|
||
if stripped.startswith("/"):
|
||
return True
|
||
if cls._TRIVIAL_PROMPT_RE.match(stripped):
|
||
return True
|
||
return False
|
||
|
||
def on_turn_start(self, turn_number: int, message: str, **kwargs) -> None:
|
||
"""Track turn count for cadence and injection_frequency logic."""
|
||
self._turn_count = turn_number
|
||
|
||
@staticmethod
|
||
def _chunk_message(content: str, limit: int) -> list[str]:
|
||
"""Split content into chunks that fit within the Honcho message limit.
|
||
|
||
Splits at paragraph boundaries when possible, falling back to
|
||
sentence boundaries, then word boundaries. Each continuation
|
||
chunk is prefixed with "[continued] " so Honcho's representation
|
||
engine can reconstruct the full message.
|
||
"""
|
||
if len(content) <= limit:
|
||
return [content]
|
||
|
||
prefix = "[continued] "
|
||
prefix_len = len(prefix)
|
||
chunks = []
|
||
remaining = content
|
||
first = True
|
||
while remaining:
|
||
effective = limit if first else limit - prefix_len
|
||
if len(remaining) <= effective:
|
||
chunks.append(remaining if first else prefix + remaining)
|
||
break
|
||
|
||
segment = remaining[:effective]
|
||
|
||
# Try paragraph break, then sentence, then word
|
||
cut = segment.rfind("\n\n")
|
||
if cut < effective * 0.3:
|
||
cut = segment.rfind(". ")
|
||
if cut >= 0:
|
||
cut += 2 # include the period and space
|
||
if cut < effective * 0.3:
|
||
cut = segment.rfind(" ")
|
||
if cut < effective * 0.3:
|
||
cut = effective # hard cut
|
||
|
||
chunk = remaining[:cut].rstrip()
|
||
remaining = remaining[cut:].lstrip()
|
||
if not first:
|
||
chunk = prefix + chunk
|
||
chunks.append(chunk)
|
||
first = False
|
||
|
||
return chunks
|
||
|
||
def _empty_profile_hint(self, peer: str) -> Dict[str, Any]:
|
||
"""Build a diagnostic hint when honcho_profile returns an empty card.
|
||
|
||
A literal "No profile facts available yet." tells the model nothing
|
||
about WHY. The model then often surfaces it to the user as a cryptic
|
||
error. This hint enumerates the likely causes so the model can
|
||
explain the situation (or retry with a different peer).
|
||
|
||
Ordered by likelihood for a typical deployment:
|
||
1. Observation is disabled for this peer
|
||
2. Card hasn't accumulated yet (fresh peer, not enough dialectic
|
||
cycles — dialectic cadence runs every N turns)
|
||
3. Self-hosted Honcho backend doesn't support peer cards
|
||
(honcho-ai server < 3.x)
|
||
"""
|
||
cfg = self._config
|
||
reasons: List[str] = []
|
||
|
||
if cfg is not None:
|
||
if peer == "user":
|
||
observe_me = bool(getattr(cfg, "user_observe_me", True))
|
||
observe_others = bool(getattr(cfg, "user_observe_others", True))
|
||
else:
|
||
observe_me = bool(getattr(cfg, "ai_observe_me", True))
|
||
observe_others = bool(getattr(cfg, "ai_observe_others", True))
|
||
if not (observe_me or observe_others):
|
||
reasons.append(
|
||
f"observation is disabled for peer '{peer}' "
|
||
f"(user_observe_me/ai_observe_me in config)"
|
||
)
|
||
|
||
cadence = getattr(self, "_dialectic_cadence", 1)
|
||
turn = getattr(self, "_turn_count", 0)
|
||
if turn < max(2, cadence):
|
||
reasons.append(
|
||
f"this session has only {turn} turn(s); peer cards accumulate "
|
||
f"as the dialectic layer reasons over conversation history "
|
||
f"(cadence every {cadence} turn(s))"
|
||
)
|
||
|
||
if not reasons:
|
||
reasons.append(
|
||
"peer card has no facts yet — Honcho's dialectic layer builds "
|
||
"this over time from observed turns; self-hosted Honcho < 3.x "
|
||
"does not support peer cards at all"
|
||
)
|
||
|
||
return {
|
||
"result": "No profile facts available yet.",
|
||
"hint": (
|
||
"This is not an error. "
|
||
+ "; ".join(reasons)
|
||
+ ". Try honcho_reasoning for a synthesized answer, or "
|
||
"honcho_search to query raw conversation excerpts."
|
||
),
|
||
}
|
||
|
||
def sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None:
|
||
"""Record the conversation turn in Honcho (non-blocking).
|
||
|
||
Messages exceeding the Honcho API limit (default 25k chars) are
|
||
split into multiple messages with continuation markers.
|
||
"""
|
||
if self._cron_skipped:
|
||
return
|
||
if self._recall_mode == "tools" and not self._session_ready():
|
||
return
|
||
if not self._session_ready():
|
||
self._start_session_init_background()
|
||
return
|
||
|
||
msg_limit = self._config.message_max_chars if self._config else 25000
|
||
clean_user_content = sanitize_context(user_content or "").strip()
|
||
clean_assistant_content = sanitize_context(assistant_content or "").strip()
|
||
|
||
def _sync():
|
||
try:
|
||
session = self._manager.get_or_create(self._session_key)
|
||
for chunk in self._chunk_message(clean_user_content, msg_limit):
|
||
session.add_message("user", chunk)
|
||
for chunk in self._chunk_message(clean_assistant_content, msg_limit):
|
||
session.add_message("assistant", chunk)
|
||
self._manager._flush_session(session)
|
||
except Exception as e:
|
||
logger.debug("Honcho sync_turn failed: %s", e)
|
||
|
||
if self._sync_thread and self._sync_thread.is_alive():
|
||
self._sync_thread.join(timeout=5.0)
|
||
self._sync_thread = threading.Thread(
|
||
target=_sync, daemon=True, name="honcho-sync"
|
||
)
|
||
self._sync_thread.start()
|
||
|
||
def on_memory_write(
|
||
self,
|
||
action: str,
|
||
target: str,
|
||
content: str,
|
||
metadata: Optional[Dict[str, Any]] = None,
|
||
) -> None:
|
||
"""Mirror built-in user profile writes as Honcho conclusions.
|
||
|
||
``metadata`` is accepted for compatibility with the write-origin
|
||
work landed in main (commit 6a957a74); it's not yet threaded into
|
||
the Honcho conclusion payload. Left as a follow-up so this PR
|
||
stays focused on the 7-PR consolidation and its review follow-ups.
|
||
"""
|
||
if action != "add" or target != "user" or not content:
|
||
return
|
||
if self._cron_skipped:
|
||
return
|
||
if self._recall_mode == "tools" and not self._session_ready():
|
||
return
|
||
if not self._session_ready():
|
||
self._start_session_init_background()
|
||
return
|
||
|
||
def _write():
|
||
try:
|
||
self._manager.create_conclusion(self._session_key, content)
|
||
except Exception as e:
|
||
logger.debug("Honcho memory mirror failed: %s", e)
|
||
|
||
t = threading.Thread(target=_write, daemon=True, name="honcho-memwrite")
|
||
t.start()
|
||
|
||
def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
|
||
"""Flush all pending messages to Honcho on session end."""
|
||
if self._cron_skipped:
|
||
return
|
||
if not self._manager:
|
||
return
|
||
if not self._session_initialized and self._init_thread and self._init_thread.is_alive():
|
||
return
|
||
# Wait for pending sync
|
||
if self._sync_thread and self._sync_thread.is_alive():
|
||
self._sync_thread.join(timeout=10.0)
|
||
try:
|
||
self._manager.flush_all()
|
||
except Exception as e:
|
||
logger.debug("Honcho session-end flush failed: %s", e)
|
||
|
||
def get_tool_schemas(self) -> List[Dict[str, Any]]:
|
||
"""Return tool schemas, respecting recall_mode.
|
||
|
||
B1: context-only mode hides all tools.
|
||
"""
|
||
if self._cron_skipped:
|
||
return []
|
||
if self._recall_mode == "context":
|
||
return []
|
||
return list(ALL_TOOL_SCHEMAS)
|
||
|
||
def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str:
|
||
"""Handle a Honcho tool call, with lazy session init for tools-only mode."""
|
||
if self._cron_skipped:
|
||
return tool_error("Honcho is not active (cron context).")
|
||
|
||
# Port #1957: ensure session is initialized for tools-only mode
|
||
if not self._session_initialized:
|
||
if self._init_thread and self._init_thread.is_alive():
|
||
return tool_error("Honcho session is still initializing; try again shortly.")
|
||
if not self._ensure_session():
|
||
return tool_error("Honcho session could not be initialized.")
|
||
|
||
if not self._manager or not self._session_key:
|
||
return tool_error("Honcho is not active for this session.")
|
||
|
||
try:
|
||
if tool_name == "honcho_profile":
|
||
peer = args.get("peer", "user")
|
||
card_update = args.get("card")
|
||
if card_update:
|
||
result = self._manager.set_peer_card(self._session_key, card_update, peer=peer)
|
||
if result is None:
|
||
return tool_error("Failed to update peer card.")
|
||
return json.dumps({"result": f"Peer card updated ({len(result)} facts).", "card": result})
|
||
card = self._manager.get_peer_card(self._session_key, peer=peer)
|
||
if not card:
|
||
return json.dumps(self._empty_profile_hint(peer))
|
||
return json.dumps({"result": card})
|
||
|
||
elif tool_name == "honcho_search":
|
||
query = args.get("query", "")
|
||
if not query:
|
||
return tool_error("Missing required parameter: query")
|
||
max_tokens = min(int(args.get("max_tokens", 800)), 2000)
|
||
peer = args.get("peer", "user")
|
||
result = self._manager.search_context(
|
||
self._session_key, query, max_tokens=max_tokens, peer=peer
|
||
)
|
||
if not result:
|
||
return json.dumps({"result": "No relevant context found."})
|
||
return json.dumps({"result": result})
|
||
|
||
elif tool_name == "honcho_reasoning":
|
||
query = args.get("query", "")
|
||
if not query:
|
||
return tool_error("Missing required parameter: query")
|
||
peer = args.get("peer", "user")
|
||
reasoning_level = args.get("reasoning_level")
|
||
result = self._manager.dialectic_query(
|
||
self._session_key, query,
|
||
reasoning_level=reasoning_level,
|
||
peer=peer,
|
||
)
|
||
# Update cadence tracker so auto-injection respects the gap after an explicit call
|
||
self._last_dialectic_turn = self._turn_count
|
||
return json.dumps({"result": result or "No result from Honcho."})
|
||
|
||
elif tool_name == "honcho_context":
|
||
peer = args.get("peer", "user")
|
||
ctx = self._manager.get_session_context(self._session_key, peer=peer)
|
||
if not ctx:
|
||
return json.dumps({"result": "No context available yet."})
|
||
parts = []
|
||
if ctx.get("summary"):
|
||
parts.append(f"## Summary\n{ctx['summary']}")
|
||
if ctx.get("representation"):
|
||
parts.append(f"## Representation\n{ctx['representation']}")
|
||
if ctx.get("card"):
|
||
parts.append(f"## Card\n{ctx['card']}")
|
||
if ctx.get("recent_messages"):
|
||
msgs = ctx["recent_messages"]
|
||
msg_str = "\n".join(
|
||
f" [{m['role']}] {m['content'][:200]}"
|
||
for m in msgs[-5:] # last 5 for brevity
|
||
)
|
||
parts.append(f"## Recent messages\n{msg_str}")
|
||
return json.dumps({"result": "\n\n".join(parts) or "No context available."})
|
||
|
||
elif tool_name == "honcho_conclude":
|
||
delete_id = (args.get("delete_id") or "").strip()
|
||
conclusion = args.get("conclusion", "").strip()
|
||
list_mode = bool(args.get("list"))
|
||
peer = args.get("peer", "user")
|
||
|
||
has_delete_id = bool(delete_id)
|
||
has_conclusion = bool(conclusion)
|
||
if sum([has_delete_id, has_conclusion, list_mode]) != 1:
|
||
return tool_error("Exactly one of conclusion, delete_id, or list must be provided.")
|
||
|
||
if list_mode:
|
||
query = (args.get("query") or "").strip() or None
|
||
conclusions = self._manager.list_conclusions(self._session_key, query=query, peer=peer)
|
||
return json.dumps({"conclusions": conclusions})
|
||
if has_delete_id:
|
||
ok = self._manager.delete_conclusion(self._session_key, delete_id, peer=peer)
|
||
if ok:
|
||
return json.dumps({"result": f"Conclusion {delete_id} deleted."})
|
||
return tool_error(f"Failed to delete conclusion {delete_id}.")
|
||
ok = self._manager.create_conclusion(self._session_key, conclusion, peer=peer)
|
||
if ok:
|
||
return json.dumps({"result": f"Conclusion saved for {peer}: {conclusion}"})
|
||
return tool_error("Failed to save conclusion.")
|
||
|
||
return tool_error(f"Unknown tool: {tool_name}")
|
||
|
||
except Exception as e:
|
||
logger.error("Honcho tool %s failed: %s", tool_name, e)
|
||
return tool_error(f"Honcho {tool_name} failed: {e}")
|
||
|
||
def shutdown(self) -> None:
|
||
for t in (self._prefetch_thread, self._sync_thread):
|
||
if t and t.is_alive():
|
||
t.join(timeout=5.0)
|
||
# Flush any remaining messages
|
||
if self._manager and not (self._init_thread and self._init_thread.is_alive() and not self._session_initialized):
|
||
try:
|
||
self._manager.flush_all()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Plugin entry point
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def register(ctx) -> None:
|
||
"""Register Honcho as a memory provider plugin."""
|
||
from plugins.memory.honcho.query_rewrite import rewrite_dialectic_query
|
||
|
||
ctx.register_memory_provider(
|
||
HonchoMemoryProvider(query_rewriter=rewrite_dialectic_query)
|
||
)
|