"""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.

Five tools (profile, search, reasoning, 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).\n"
                    "reasoning_level parameter guide:\n"
                    "- minimal: use ONLY for a single quick factual lookup (e.g. "
                    "'what is the user's name'). Honcho hard-caps this tier's output "
                    "at 250 tokens combined with the model's own hidden reasoning "
                    "tokens — a multi-part answer can get cut off mid-thought before "
                    "it even reaches the final-answer phase, especially on models "
                    "with reasoning/thinking enabled.\n"
                    "- low/medium/high/max: use for anything requiring a synthesized, "
                    "multi-fact, or summary-style answer (e.g. 'summarize known facts "
                    "about this peer', 'what are their communication preferences'). "
                    "These tiers have no output-token cap of their own (fall back to "
                    "Honcho's 8192-token global default), so they don't have "
                    "minimal's cutoff failure mode.\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\n"
                    "Default to at least 'low' unless the query is genuinely a single "
                    "fact lookup."
                ),
                "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

        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()

        # Recall cadence and liveness state.
        self._turn_count = 0
        self._query_rewrite_enabled = False
        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

        # Tools-only mode may defer session initialization until a tool call.
        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 = ""

        # Cron and flush contexts disable the plugin entirely.
        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()
            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(encoding="utf-8"))
            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 guards, recall configuration, session resolution,
        memory migration, and optional dialectic prewarming.
        """
        try:
            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

            self._recall_mode = cfg.recall_mode  # "context", "tools", or "hybrid"
            logger.debug("Honcho recall_mode: %s", self._recall_mode)

            self._injection_frequency = cfg.injection_frequency
            self._context_cadence = cfg.context_cadence
            self._dialectic_cadence = cfg.dialectic_cadence
            self._query_rewrite_enabled = cfg.query_rewrite
            self._FIRST_TURN_BASE_TIMEOUT = cfg.first_turn_base_wait
            self._FIRST_TURN_DIALECTIC_CAP = cfg.first_turn_dialectic_wait
            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,
        )

        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)

        # 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)

        # Query-aware base retrieval starts with the first substantive message.
        # Generic dialectic prewarm is incompatible with latest-message rewriting.
        if self._recall_mode in {"context", "hybrid"}:
            if self._query_rewriter is None or not self._query_rewrite_enabled:
                _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
                        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
                logger.debug("Honcho dialectic prewarm started for session: %s", self._session_key)
            else:
                logger.debug(
                    "Honcho generic dialectic prewarm skipped: awaiting first user message"
                )

        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 ""

        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

        Returns empty in tools-only mode and respects the configured injection
        frequency and context budget.
        """
        if self._cron_skipped:
            return ""

        # Tools-only mode has no automatic injection.
        if self._recall_mode == "tools":
            return ""

        first_turn_base_deadline = None
        if self._turn_count <= 1:
            base_wait = self._FIRST_TURN_BASE_TIMEOUT
            request_timeout = getattr(self._config, "timeout", None)
            if request_timeout is not None:
                base_wait = min(base_wait, max(0.0, request_timeout))
            first_turn_base_deadline = time.monotonic() + max(0.0, base_wait)

        if not self._session_ready():
            # Only turn 1 may wait for session init; later turns fail open.
            self._start_session_init_background()
            if first_turn_base_deadline is not None:
                _init_thread = self._init_thread
                if _init_thread is not None:
                    _init_thread.join(
                        timeout=max(0.0, first_turn_base_deadline - time.monotonic())
                    )
            if not self._session_ready():
                return ""

        # First-turn mode suppresses only the base layer; dialectic is independent.
        _skip_base = (
            self._injection_frequency == "first-turn" and self._turn_count > 1
        )

        # Trivial turns start no work, but may consume a ready pending result.
        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) -----
        if not _skip_base:
            # The first base fetch gets the remaining turn-1 budget. Later
            # refreshes are consumed asynchronously.
            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:
                _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()
                _base_wait = (
                    max(0.0, first_turn_base_deadline - time.monotonic())
                    if first_turn_base_deadline is not None
                    else 0.0
                )
                _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", _base_wait,
                    )

            # Later turns consume the refresh queued by the previous turn.
            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 -----
        # Turn 1 may briefly wait for dialectic; unfinished work remains async.
        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

        if self._last_dialectic_turn == -999 and query:
            # Reuse an in-flight prewarm; otherwise start one dialectic. A short
            # request timeout may tighten, but never expand, the turn-1 wait.
            _dia_wait = self._FIRST_TURN_DIALECTIC_CAP
            request_timeout = getattr(self._config, "timeout", None)
            if request_timeout is not None:
                _dia_wait = min(_dia_wait, max(0.0, request_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
                        # Empty results do not consume the cadence window.
                        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",
                    _dia_wait,
                )

        # Consume only results that are already ready; later turns never wait.
        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)

        result = self._truncate_to_budget(result)

        return result

    def _consume_pending_dialectic(self) -> str:
        """Pop any pending dialectic result, applying the stale-discard guard.

        Returns an empty string when no result is ready or the pending result
        is older than the configured cadence window.
        """
        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

        # Drop results that no longer track 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.

        Context and dialectic refreshes have independent cadence controls.
        """
        if self._cron_skipped:
            return
        # Tools-only mode has no automatic 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

        # First-turn-only base context never needs a later refresh.
        context_due = (
            self._context_cadence <= 1
            or (self._turn_count - self._last_context_turn) >= self._context_cadence
        )
        if self._injection_frequency != "first-turn" and context_due:
            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
    # Total turn-1 budget shared by session init and base retrieval.
    _FIRST_TURN_BASE_TIMEOUT = 3.0
    # Independent turn-1 grace for the slower dialectic path.
    _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_rewrite_enabled 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):
            # Dependent prompts require a non-empty prior result.
            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:
                    # Retry the independent base prompt after empty passes.
                    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.

        Context-only mode exposes no Honcho 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).")

        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") or "").strip()
                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") or "").strip()
                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,
                    # Explicit reasoning bypasses the automatic-injection cap.
                    apply_injection_cap=False,
                )
                # 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.")

                query = (args.get("query") or "").strip()
                if query and not list_mode:
                    return tool_error("query is only valid when list is true.")

                if list_mode:
                    conclusions = self._manager.list_conclusions(
                        self._session_key, query=query or None, 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.query_rewrite import rewrite_memory_query

    ctx.register_memory_provider(
        HonchoMemoryProvider(query_rewriter=rewrite_memory_query)
    )
