plugin_init.py

"""tracedecay Hermes plugin registration."""
import json
import hashlib
import logging
import os
import re
import shutil
import threading
import time
from pathlib import Path

from . import schemas, tools

logger = logging.getLogger(__name__)

# Canonical profile-home resolver (hermes_constants), with the legacy
# hermes_cli.config location as fallback; both guarded so the plugin still
# imports outside a Hermes install.
try:
    from hermes_constants import get_hermes_home as _hermes_get_hermes_home
except Exception:
    try:
        from hermes_cli.config import get_hermes_home as _hermes_get_hermes_home
    except Exception:
        _hermes_get_hermes_home = None

# Canonical config read/write path for provider save_config(); guarded for
# use outside Hermes (raw-YAML fallback below).
try:
    from hermes_cli import config as _hermes_cli_config
except Exception:
    _hermes_cli_config = None

try:
    from agent.memory_provider import MemoryProvider
except Exception:
    class MemoryProvider:
        pass

try:
    from agent.context_engine import ContextEngine
except Exception:
    class ContextEngine:
        pass

# Hermes' centralized auxiliary LLM facade is the MODULE-LEVEL
# agent.auxiliary_client.call_llm(task=..., messages=..., ...) — AIAgent
# instances carry no ``auxiliary_client`` attribute and no host call site
# hands the plugin an agent object. Guarded so the plugin still degrades
# gracefully (deterministic fallback summaries) outside a hermes install.
try:
    from agent import auxiliary_client as _hermes_auxiliary_client
except Exception:
    _hermes_auxiliary_client = None

def _resolve_auxiliary_client(agent=None):
    """Best auxiliary LLM client: an agent-attached one, else hermes' module-level facade."""
    client = getattr(agent, "auxiliary_client", None)
    if client is not None and callable(getattr(client, "call_llm", None)):
        return client
    if _hermes_auxiliary_client is not None and callable(
        getattr(_hermes_auxiliary_client, "call_llm", None)
    ):
        return _hermes_auxiliary_client
    return None

MEMORY_FACT_ACTIONS = {
    "fact_add": "add",
    "fact_search": "search",
    "fact_probe": "probe",
    "fact_related": "related",
    "fact_reason": "reason",
    "fact_contradict": "contradict",
    "fact_update": "update",
    "fact_remove": "remove",
    "fact_list": "list",
}

MEMORY_ACTION_DESCRIPTIONS = {
    "fact_add": (
        "Add a holographic memory fact. The result includes a write-time diff "
        "report (diff/closest_fact_id/similarity/reason): 'near_duplicate' "
        "means a very similar fact already exists (consider updating it "
        "instead), 'possible_conflict' means a negation/state-change cue "
        "suggests supersession (confirm which fact is current), and "
        "'rejected_secret_like' means the content looked like a credential "
        "and was NOT stored. Calibrate trust instead of defaulting high: "
        "reserve >=0.85 for verified/durable facts, use ~0.7 for ordinary "
        "observations and ~0.5 when unsure - aim for a spread across facts."
    ),
    "fact_search": (
        "Search holographic memory facts by query. Recall memory FIRST "
        "before reaching for external or web search - prior sessions often "
        "already answered the question."
    ),
    "fact_probe": "Find facts connected to one entity.",
    "fact_related": "List entities related to one entity.",
    "fact_reason": "Reason over facts that connect multiple entities.",
    "fact_contradict": "Scan memory facts for likely contradictions.",
    "fact_update": "Update an existing holographic memory fact.",
    "fact_remove": "Remove a holographic memory fact.",
    "fact_list": "List holographic memory facts.",
}

MEMORY_TOOL_MAP = {"fact_store": {"tracedecay_name": "tracedecay_fact_store"}}
for _hermes_name, _action in MEMORY_FACT_ACTIONS.items():
    MEMORY_TOOL_MAP[_hermes_name] = {
        "tracedecay_name": "tracedecay_fact_store",
        "fixed_args": {"action": _action},
    }
MEMORY_TOOL_MAP["fact_feedback"] = {"tracedecay_name": "tracedecay_fact_feedback"}
MEMORY_TOOL_MAP["memory_status"] = {"tracedecay_name": "tracedecay_memory_status"}

LCM_TOOL_ALIASES = {
    "lcm_grep": "tracedecay_lcm_grep",
    "lcm_load_session": "tracedecay_lcm_load_session",
    "lcm_describe": "tracedecay_lcm_describe",
    "lcm_expand": "tracedecay_lcm_expand",
    "lcm_expand_query": "tracedecay_lcm_expand_query",
    "lcm_status": "tracedecay_lcm_status",
    "lcm_doctor": "tracedecay_lcm_doctor",
}
LCM_DIRECT_TOOL_NAMES = frozenset(LCM_TOOL_ALIASES.values())
LCM_DIRECT_TO_NATIVE = {tracedecay_name: native_name for native_name, tracedecay_name in LCM_TOOL_ALIASES.items()}

LCM_NATIVE_SCHEMAS = [
    {
        "name": "lcm_grep",
        "description": (
            "Search the plugin-local LCM database for past conversation content. "
            "Default scope is the active session and returns raw messages and summary nodes."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query."},
                "limit": {"type": "integer", "description": "Max results to return.", "default": 10},
                "sort": {
                    "type": "string",
                    "enum": ["recency", "relevance", "hybrid"],
                    "description": "How to order matches.",
                    "default": "recency",
                },
                "session_scope": {
                    "type": "string",
                    "enum": ["current", "all", "session"],
                    "description": "Search scope across the local LCM database.",
                    "default": "current",
                },
                "session_id": {"type": "string", "description": "Session id when session_scope='session'."},
                "source": {"type": "string", "description": "Optional source/platform filter."},
                "role": {
                    "type": "string",
                    "enum": ["system", "user", "assistant", "tool", "unknown"],
                    "description": "Optional raw-message role filter.",
                },
                "time_from": {
                    "anyOf": [{"type": "number"}, {"type": "string"}],
                    "description": "Optional inclusive minimum raw-message timestamp.",
                },
                "time_to": {
                    "anyOf": [{"type": "number"}, {"type": "string"}],
                    "description": "Optional inclusive maximum raw-message timestamp.",
                },
            },
            "required": ["query"],
        },
    },
    {
        "name": "lcm_load_session",
        "description": "Load an ordered raw-message transcript page for one explicit session_id.",
        "parameters": {
            "type": "object",
            "properties": {
                "session_id": {"type": "string", "description": "Explicit LCM session id to load."},
                "limit": {"type": "integer", "description": "Maximum raw messages to return.", "default": 100},
                "max_content_chars": {
                    "type": "integer",
                    "description": "Maximum content characters to include per message.",
                    "default": 4000,
                },
                "after_store_id": {
                    "type": "integer",
                    "description": "Exclusive cursor for pagination.",
                    "default": 0,
                },
                "roles": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Optional role filter.",
                },
                "time_from": {
                    "type": "number",
                    "description": "Optional inclusive minimum message timestamp.",
                },
                "time_to": {
                    "type": "number",
                    "description": "Optional inclusive maximum message timestamp.",
                },
            },
            "required": ["session_id"],
        },
    },
    {
        "name": "lcm_describe",
        "description": "Inspect a current-session summary node, externalized payload, or top-level DAG overview.",
        "parameters": {
            "type": "object",
            "properties": {
                "node_id": {"type": "integer", "description": "Summary node ID to inspect."},
                "externalized_ref": {
                    "type": "string",
                    "description": "Externalized payload ref filename to inspect.",
                },
            },
            "required": [],
        },
    },
    {
        "name": "lcm_expand",
        "description": "Recover detail behind a summary node, externalized payload, or raw message.",
        "parameters": {
            "type": "object",
            "properties": {
                "node_id": {"type": "integer", "description": "Summary node ID to expand."},
                "externalized_ref": {
                    "type": "string",
                    "description": "Externalized payload ref filename to expand.",
                },
                "store_id": {"type": "integer", "description": "Raw message store_id to fetch."},
                "session_id": {
                    "type": "string",
                    "description": "Optional session id override (for example, expand a cross-session grep hit in its owning session).",
                },
                "max_tokens": {"type": "integer", "description": "Token budget for returned content.", "default": 4000},
                "source_offset": {
                    "type": "integer",
                    "description": "Source pagination offset for node_id mode.",
                    "default": 0,
                },
                "source_limit": {
                    "type": "integer",
                    "description": "Maximum immediate sources to return from source_offset. If a returned source marks content_truncated=true, continue from its own store_id + content_offset.",
                },
                "content_offset": {
                    "type": "integer",
                    "description": "Character offset used to continue oversized content.",
                    "default": 0,
                },
            },
            "required": [],
        },
    },
    {
        "name": "lcm_expand_query",
        "description": "Answer a natural-language question using expanded LCM context from the current session.",
        "parameters": {
            "type": "object",
            "properties": {
                "prompt": {"type": "string", "description": "The question or task to answer from expanded LCM context."},
                "query": {"type": "string", "description": "Optional search query used to find candidate summaries."},
                "node_ids": {
                    "type": "array",
                    "items": {"type": "integer"},
                    "description": "Optional explicit summary node IDs.",
                },
                "max_results": {"type": "integer", "description": "Max candidate summaries.", "default": 5},
                "max_tokens": {"type": "integer", "description": "Max answer tokens.", "default": 2000},
                "context_max_tokens": {
                    "type": "integer",
                    "description": "Expanded context budget for the auxiliary LLM.",
                    "default": 32000,
                },
            },
            "required": ["prompt"],
        },
    },
    {
        "name": "lcm_status",
        "description": "Get a quick health overview of the LCM engine for the current session.",
        "parameters": {"type": "object", "properties": {}, "required": []},
    },
    {
        "name": "lcm_doctor",
        "description": "Run diagnostics on the LCM database/configuration, including payload GC preview/apply via gc mode.",
        "parameters": {"type": "object", "properties": {}, "required": []},
    },
]

# Tools whose registered value depends on the host forwarding the live
# in-memory ``messages`` list to plugin tool handlers (their schemas carry a
# ``messages`` parameter used for lossless LCM ingest). Everything else in
# TOOL_SCHEMAS works without that capability.
MESSAGE_DEPENDENT_TOOLS = frozenset((
    "tracedecay_lcm_compress",
    "tracedecay_lcm_preflight",
))

STANDARD_HERMES_LCM_PROVIDER = "cursor"

LCM_PROVIDER_LOCAL_TOOL_NAMES = frozenset((
    "tracedecay_lcm_compress",
    "tracedecay_lcm_describe",
    "tracedecay_lcm_doctor",
    "tracedecay_lcm_expand",
    "tracedecay_lcm_expand_query",
    "tracedecay_lcm_preflight",
    "tracedecay_lcm_session_boundary",
))

# Direct duplicates of the memory provider's own tool surface
# (fact_store / fact_feedback / memory_status). Skipped at register() time
# when tracedecay is the active memory.provider so the same store is not
# exposed twice per API call. tracedecay_message_search stays registered —
# the provider does not expose transcript search.
MEMORY_PROVIDER_TOOLS = frozenset((
    "tracedecay_fact_store",
    "tracedecay_fact_feedback",
    "tracedecay_memory_status",
))

# Tool names successfully registered with this host. Consulted by the
# first-turn guidance nudge so it never advertises tools that are not
# actually registered.
_REGISTERED_TOOL_NAMES = set()
_CONTEXT_TOOL_NAMES = set()
_HOST_FORWARDS_MESSAGES = None

# Auxiliary task key for pre-compaction extraction. Upgraded to the
# plugin-registered "lcm_extraction" task when the host supports
# ctx.register_auxiliary_task (users can then pin its model under
# auxiliary.lcm_extraction); otherwise hermes' generic "extraction"
# defaults apply.
_EXTRACTION_TASK = {"key": "extraction"}

def _active_memory_provider(ctx=None):
    """The memory.provider configured for this profile, if any."""
    config = getattr(ctx, "config", None) if ctx is not None else None
    if isinstance(config, dict):
        memory = config.get("memory")
        if isinstance(memory, dict) and memory.get("provider"):
            return str(memory.get("provider"))
    try:
        import yaml
        config_path = os.path.join(tools.hermes_home_dir(), "config.yaml")
        with open(config_path, encoding="utf-8-sig") as config_file:
            raw = yaml.safe_load(config_file) or {}
        memory = raw.get("memory")
        if isinstance(memory, dict) and memory.get("provider"):
            return str(memory.get("provider"))
    except Exception:
        pass
    return None

def _make_wrapped_lcm_handler(tool_name: str, engine):
    def _wrapped(args: dict, **kwargs) -> str:
        return engine.handle_tool_call(tool_name, args, **kwargs)
    return _wrapped

def _host_forwards_registered_tool_messages(ctx) -> bool:
    capability = getattr(ctx, "context_engine_tool_handlers_receive_messages", False)
    if callable(capability):
        try:
            capability = capability()
        except Exception:
            return False
    return bool(capability)

def _pre_llm_call(*args, **kwargs):
    # Inject guidance ONLY on the first turn: the hook result is appended to
    # the user message, so emitting it every turn would change every turn
    # boundary and break the conversation's prompt-cache prefix. Skip it
    # entirely when no tracedecay tools actually registered on this host —
    # advertising unregistered tools invites hallucinated calls.
    if not kwargs.get("is_first_turn"):
        return None
    if not _REGISTERED_TOOL_NAMES:
        return None
    if not _plugin_toggle("nudge", True):
        return None
    return (
        "Prefer tracedecay tools for codebase exploration, symbol lookup, call graphs, "
        "impact analysis, affected files, and architectural navigation before broad file reads."
    )

def _tracedecay_status(raw_args: str = ""):
    raw = tools.call_tracedecay_tool("tracedecay_status", {})
    try:
        payload = json.loads(json.loads(raw)["content"][0]["text"])
    except Exception:
        return raw
    if not isinstance(payload, dict) or payload.get("error"):
        return raw
    lines = ["tracedecay status:"]
    for label, key in (
        ("project", "project_root"),
        ("files", "file_count"),
        ("nodes", "node_count"),
        ("edges", "edge_count"),
        ("branch", "branch"),
        ("db", "db_path"),
        ("last sync", "last_sync"),
    ):
        value = payload.get(key)
        if value not in (None, ""):
            lines.append(f"  {label}: {value}")
    if len(lines) == 1:
        return raw
    return "\n".join(lines)

def _bridge_preview(value, limit: int = 2048) -> str:
    if isinstance(value, str):
        preview = value
    else:
        try:
            preview = json.dumps(value, sort_keys=True)
        except Exception:
            preview = repr(value)
    if len(preview) > limit:
        return preview[:limit] + "...[truncated]"
    return preview


_LCM_CONTRACT_KEYS = frozenset((
    "answer",
    "context_blocks",
    "expansion",
    "frontier",
    "lcm",
    "matches",
    "needs_synthesis",
    "replay_messages",
    "context_recovery_hint",
    "should_compress",
    "status",
    "summary_request",
))
_RETRIEVAL_HANDLE_KEYS = ("handle", "response_handle", "retrieval_handle")

def _json_or_none(text):
    if not isinstance(text, str):
        return None
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return None

def _content_text_candidates(content):
    if isinstance(content, str):
        return [content]
    if not isinstance(content, list):
        return []
    parts = [
        item.get("text")
        for item in content
        if isinstance(item, dict) and isinstance(item.get("text"), str)
    ]
    candidates = []
    if parts:
        candidates.append("".join(parts))
        joined = "\n".join(parts)
        if joined != candidates[0]:
            candidates.append(joined)
        candidates.extend(parts)
    return candidates

def _decode_content_json(content):
    for candidate in _content_text_candidates(content):
        decoded = _json_or_none(candidate)
        if decoded is not None:
            return decoded
    return None

def _looks_like_lcm_contract(value):
    return isinstance(value, dict) and bool(_LCM_CONTRACT_KEYS.intersection(value))

def _retrieval_handle(value):
    if not isinstance(value, dict):
        return None
    if value.get("truncated") is not True and not value.get("retrieve_tool"):
        if not any(key in value for key in ("response_handle", "retrieval_handle")):
            return None
    for key in _RETRIEVAL_HANDLE_KEYS:
        handle = value.get(key)
        if isinstance(handle, str) and handle.strip():
            return handle.strip()
    return None

def _lcm_retrieve_kwargs(args: dict, kwargs: dict) -> dict:
    retrieve_kwargs = dict(kwargs or {})
    if retrieve_kwargs.get("project_root"):
        return retrieve_kwargs
    if isinstance(args, dict):
        root = args.get("response_handle_project_root") or args.get("project_root")
        if not root and args.get("storage_scope") == "hermes_profile":
            root = args.get("hermes_home")
        if isinstance(root, str) and root.strip():
            retrieve_kwargs["project_root"] = root.strip()
    return retrieve_kwargs

def _decode_tool_payload(value, name: str, args: dict, kwargs: dict, depth: int = 0, seen_handles=None):
    if depth > 8:
        return value
    if seen_handles is None:
        seen_handles = set()
    if isinstance(value, str):
        decoded = _json_or_none(value)
        if decoded is None:
            return value
        return _decode_tool_payload(decoded, name, args, kwargs, depth + 1, seen_handles)
    if not isinstance(value, dict):
        return value

    handle = _retrieval_handle(value)
    if handle and name.startswith("tracedecay_lcm_"):
        if handle in seen_handles:
            return value
        seen_handles.add(handle)
        retrieved = call_tracedecay_json(
            "tracedecay_retrieve",
            {"handle": handle},
            **_lcm_retrieve_kwargs(args, kwargs),
        )
        if isinstance(retrieved, dict) and not retrieved.get("error"):
            return _decode_tool_payload(retrieved, name, args, kwargs, depth + 1, seen_handles)
        return retrieved

    if _looks_like_lcm_contract(value):
        return value

    if "content" in value:
        decoded = _decode_content_json(value.get("content"))
        if decoded is not None:
            return _decode_tool_payload(decoded, name, args, kwargs, depth + 1, seen_handles)

    return value

def call_tracedecay_json(name: str, args: dict, **kwargs) -> dict:
    raw = tools.call_tracedecay_tool(name, args, **kwargs)
    try:
        outer = json.loads(raw)
    except json.JSONDecodeError:
        return {
            "error": "tracedecay tool returned invalid JSON",
            "raw_preview": _bridge_preview(raw),
        }
    if isinstance(outer, dict) and "error" in outer:
        return outer
    if not isinstance(outer, dict):
        return {
            "error": "tracedecay tool response missing text content",
            "raw_preview": _bridge_preview(raw),
        }
    if not _looks_like_lcm_contract(outer) and "content" not in outer:
        return {
            "error": "tracedecay tool response missing text content",
            "raw_preview": _bridge_preview(raw),
        }
    if "content" in outer and not _content_text_candidates(outer.get("content")):
        return {
            "error": "tracedecay tool response missing text content",
            "raw_preview": _bridge_preview(raw),
        }
    payload = _decode_tool_payload(outer, name, args, kwargs)
    if isinstance(payload, dict) and "content" in outer and payload is outer:
        return {
            "error": "tracedecay tool returned invalid nested JSON",
            "text_preview": _bridge_preview(outer.get("content")),
        }
    if not isinstance(payload, dict):
        return {
            "error": "tracedecay tool response missing text content",
            "raw_preview": _bridge_preview(raw),
        }
    return payload

def _memory_schema(tracedecay_name: str, hermes_name: str, action: str = None) -> dict:
    for schema in schemas.TOOL_SCHEMAS:
        if schema.get("name") == tracedecay_name:
            parameters = json.loads(json.dumps(schema.get("parameters", {})))
            if action is not None:
                properties = parameters.get("properties")
                if isinstance(properties, dict):
                    properties.pop("action", None)
                required = parameters.get("required")
                if isinstance(required, list):
                    required = [field for field in required if field != "action"]
                    if required:
                        parameters["required"] = required
                    else:
                        parameters.pop("required", None)
            return {
                "name": hermes_name,
                "description": MEMORY_ACTION_DESCRIPTIONS.get(
                    hermes_name, schema.get("description", "")
                ),
                "parameters": parameters,
            }
    return {
        "name": hermes_name,
        "description": f"Tracedecay memory tool {hermes_name}.",
        "parameters": {"type": "object", "properties": {}},
    }

def _lcm_tool_schemas() -> list:
    return list(LCM_NATIVE_SCHEMAS)

def _decode_tool_args(arguments):
    if arguments is None:
        return {}
    if isinstance(arguments, dict):
        return arguments
    if isinstance(arguments, str):
        if not arguments.strip():
            return {}
        try:
            return json.loads(arguments)
        except json.JSONDecodeError:
            return {"arguments": arguments}
    return {"arguments": arguments}

def _normalize_memory_tool_call(name, arguments):
    if isinstance(name, dict):
        function = name.get("function") or {}
        tool_name = name.get("name") or function.get("name")
        tool_args = name.get("arguments", function.get("arguments", arguments))
        return tool_name, _decode_tool_args(tool_args)
    return name, _decode_tool_args(arguments)

def _tracedecay_binary_available() -> bool:
    if os.path.dirname(tools.TRACEDECAY_BIN):
        return Path(tools.TRACEDECAY_BIN).is_file() and os.access(tools.TRACEDECAY_BIN, os.X_OK)
    return shutil.which(tools.TRACEDECAY_BIN) is not None

def _storage_args(project_root=None, hermes_home=None):
    """Storage args for LCM/session state in the unified tracedecay store."""
    if project_root:
        return {"project_root": str(project_root)}
    home = hermes_home or _resolve_hermes_home()
    if home:
        return {"storage_scope": "hermes_profile", "hermes_home": str(home)}
    return {}

# Conventional config home: a `plugins.tracedecay` block in the profile
# config.yaml (the same `plugins.<name>` convention bundled Hermes plugins
# use). Keys are flat and mirror the host-config attribute names the
# `_configured_*` / `_lcm_*_setting` helpers read. Every key the plugin
# consults is declared here (default, dashboard description) so
# register_config_defaults()/get_config_field_meta() expose the real
# surface instead of just the install pin.
PLUGIN_CONFIG_FIELDS = {
    "project_root": ("", "Code project pinned for code-graph tool calls (set by `tracedecay install --agent hermes --project-root`)."),
    "nudge": (True, "Inject the first-turn tracedecay tool guidance nudge."),
    "sync_turn": (True, "Mirror each completed turn into the LCM raw store."),
    "prefetch": (True, "Background fact recall injected at turn start."),
    "context_threshold": ("", "Compression trigger as a fraction of the context window (default: hermes compression.threshold)."),
    "threshold_tokens": ("", "Absolute compression trigger in tokens (overrides context_threshold)."),
    "context_length": ("", "Context window override when the host does not report one."),
    "fresh_tail_count": ("", "Newest messages always kept verbatim (default 64)."),
    "leaf_chunk_tokens": ("", "Token size of LCM leaf summary chunks (default 20000)."),
    "dynamic_leaf_chunk_enabled": ("", "Scale leaf chunk size with the context window."),
    "dynamic_leaf_chunk_max": ("", "Ceiling for dynamic leaf chunk sizing (default 40000)."),
    "max_assembly_tokens": ("", "Hard cap for assembled replay tokens (0 = derive from context)."),
    "reserve_tokens_floor": ("", "Tokens reserved below the context window when deriving the assembly cap."),
    "summary_fan_in": ("", "Summary nodes condensed per parent (default 4)."),
    "incremental_max_depth": ("", "Maximum condensation depth (default 1)."),
    "summary_model": ("", "Primary auxiliary model for LCM summaries."),
    "summary_fallback_models": ("", "Comma-separated fallback models for LCM summaries."),
    "summary_timeout_ms": ("", "Auxiliary summary timeout in milliseconds."),
    "summary_circuit_breaker_failure_threshold": ("", "Failures before a summary route is cooled down (default 2)."),
    "summary_circuit_breaker_cooldown_seconds": ("", "Cooldown seconds for a tripped summary route (default 300)."),
    "custom_instructions": ("", "Extra instructions appended to the LCM summary prompt."),
    "expansion_model": ("", "Model used for lcm_expand_query synthesis."),
    "expansion_context_tokens": ("", "Expanded-context budget for lcm_expand_query (default 32000)."),
    "expansion_timeout_ms": ("", "lcm_expand_query synthesis timeout in milliseconds."),
    "extraction_enabled": ("", "Run pre-compaction decision/insight extraction."),
    "extraction_model": ("", "Model used for pre-compaction extraction."),
    "extraction_output_path": ("", "Extraction output path surfaced in the extraction contract."),
    "ignore_session_patterns": ("", "Comma-separated session-id patterns LCM ignores."),
    "stateless_session_patterns": ("", "Comma-separated session-id patterns treated as stateless."),
    "ignore_message_patterns": ("", "Comma-separated message patterns excluded from ingest."),
}

PLUGIN_CONFIG_DEFAULTS = {
    key: default for key, (default, _description) in PLUGIN_CONFIG_FIELDS.items()
}

def _plugin_config_defaults():
    # The install-time pin lives in the profile config.yaml itself
    # (plugins.tracedecay.project_root), so the defaults carry no pin.
    return dict(PLUGIN_CONFIG_DEFAULTS)

def _plugin_toggle(name, default=True):
    """Read a boolean kill switch from the plugins.tracedecay config block.

    config.yaml is the home for behavioral settings (host policy: .env is
    for secrets only).
    """
    value = tools.plugin_config_block().get(name)
    if value is None:
        return default
    if isinstance(value, str):
        normalized = value.strip().lower()
        if normalized in ("0", "false", "no", "off"):
            return False
        if normalized in ("1", "true", "yes", "on"):
            return True
        return default
    return bool(value)

class _ConfigChain:
    """Attribute-style config wrapper layering plugins.tracedecay under a host config object."""

    def __init__(self, primary, block):
        self._primary = primary
        self._block = dict(block)

    def __getattr__(self, name):
        if name.startswith("_"):
            raise AttributeError(name)
        value = getattr(self._primary, name, None)
        if value is None:
            value = self._block.get(name)
        if value is None:
            raise AttributeError(name)
        return value

def _with_plugin_block(config, hermes_home=None):
    """Layer the profile's plugins.tracedecay config block under a host config.

    Host-provided values always win; the block fills the gaps so profile
    config.yaml settings reach the engine/provider without bespoke env vars.
    """
    block = {
        key: value
        for key, value in tools.plugin_config_block(hermes_home).items()
        if value is not None and value != ""
    }
    if not block:
        return config
    if config is None:
        return dict(block)
    if isinstance(config, dict):
        merged = dict(block)
        for key, value in config.items():
            if value is not None:
                merged[key] = value
        return merged
    return _ConfigChain(config, block)

def _configured_hermes_home(config):
    if config is None:
        return None
    if isinstance(config, dict):
        return config.get("hermes_home") or config.get("home")
    for attr in ("hermes_home", "home"):
        value = getattr(config, attr, None)
        if value:
            return value
    return None

def _configured_project_root(config):
    # Profiles can pin the indexed project via a `project_root` config key
    # (kwargs from the host take precedence; cwd is the last fallback).
    if config is None:
        return None
    if isinstance(config, dict):
        value = config.get("project_root") or config.get("tracedecay_project_root")
        return str(value) if value else None
    for attr in ("project_root", "tracedecay_project_root"):
        value = getattr(config, attr, None)
        if value:
            return str(value)
    return None

def _has_tracedecay_index(path):
    if not (isinstance(path, str) and path.strip() and os.path.isabs(path)):
        return False
    return os.path.isdir(os.path.join(path, ".tracedecay"))

def _code_project_root(explicit=None, cwd=None, configured=None):
    if explicit:
        return str(explicit)
    if _has_tracedecay_index(cwd):
        return str(cwd)
    if configured:
        return str(configured)
    if isinstance(cwd, str) and os.path.isabs(cwd):
        return str(cwd)
    return None

def _resolve_hermes_home(config=None, hermes_home=None):
    for candidate in (
        hermes_home,
        _configured_hermes_home(config),
        os.environ.get("HERMES_HOME"),
    ):
        if candidate:
            return str(candidate)
    if _hermes_get_hermes_home is not None:
        try:
            resolved = _hermes_get_hermes_home()
            if resolved:
                return str(resolved)
        except Exception:
            pass
    fallback = os.path.expanduser("~/.hermes")
    return fallback or None

def _configured_value(config, *names, default=None):
    if config is None:
        return default
    if isinstance(config, dict):
        for name in names:
            if name in config and config[name] is not None:
                return config[name]
        return default
    for name in names:
        value = getattr(config, name, None)
        if value is not None:
            return value
    return default

def _configured_int(config, *names, default=None):
    value = _configured_value(config, *names, default=default)
    if value is None:
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None

def _configured_bool(config, *names, default=None):
    value = _configured_value(config, *names, default=default)
    if value is None:
        return None
    if isinstance(value, str):
        return value.strip().lower() in ("1", "true", "yes", "on")
    return bool(value)

def _parse_pattern_list(raw):
    return [part.strip() for part in str(raw).split(",") if part.strip()]

# Env-aware settings mirroring hermes-lcm LCMConfig.from_env: documented LCM_*
# env vars take precedence over host ctx.config attributes, which take
# precedence over the hermes-lcm hardcoded defaults.

def _lcm_str_setting(config, env_key, *names, default=None):
    env_value = os.environ.get(env_key)
    if env_value is not None:
        return env_value
    value = _configured_value(config, *names)
    return value if value is not None else default

def _lcm_int_setting(config, env_key, *names, default=None):
    raw = os.environ.get(env_key)
    if raw is not None:
        try:
            return int(raw)
        except (TypeError, ValueError):
            pass
    return _configured_int(config, *names, default=default)

def _lcm_reserve_tokens_floor_setting(config, context_length):
    raw = os.environ.get("LCM_RESERVE_TOKENS_FLOOR")
    if raw is not None:
        try:
            return int(raw)
        except (TypeError, ValueError):
            pass
    configured = _configured_value(config, "reserve_tokens_floor")
    if configured is not None:
        try:
            return int(configured)
        except (TypeError, ValueError):
            return 0
    try:
        return 4096 if int(context_length or 0) > 0 else 0
    except (TypeError, ValueError):
        return 0

def _lcm_float_setting(config, env_key, *names, default=None):
    raw = os.environ.get(env_key)
    if raw is not None:
        try:
            return float(raw)
        except (TypeError, ValueError):
            pass
    value = _configured_value(config, *names)
    if value is not None:
        try:
            return float(value)
        except (TypeError, ValueError):
            pass
    return default

def _lcm_bool_setting(config, env_key, *names, default=None):
    raw = os.environ.get(env_key)
    if raw is not None:
        normalized = raw.strip().lower()
        if normalized in ("1", "true", "yes", "on"):
            return True
        if normalized in ("0", "false", "no", "off"):
            return False
    return _configured_bool(config, *names, default=default)

def _lcm_list_setting(config, env_key, *names, default=None):
    raw = os.environ.get(env_key)
    if raw is not None:
        return _parse_pattern_list(raw)
    value = _configured_value(config, *names)
    if value is None:
        return default
    if isinstance(value, str):
        return _parse_pattern_list(value)
    if isinstance(value, (list, tuple)):
        return [str(item).strip() for item in value if str(item).strip()]
    return default

def _config_bool_disabled(value):
    if isinstance(value, bool):
        return value is False
    if isinstance(value, (int, float)):
        return value == 0
    if isinstance(value, str):
        normalized = value.strip().lower()
        if normalized in ("0", "false", "no", "off"):
            return True
        try:
            return float(normalized) == 0
        except ValueError:
            return False
    return False

def _hermes_yaml_compression_threshold(default, hermes_home=None):
    # Port of hermes-lcm config._hermes_compression_threshold: read the main
    # Hermes compression.threshold from {HERMES_HOME}/config.yaml when no LCM
    # override exists. Disabled Hermes compression must not leak its threshold.
    home = (
        hermes_home
        or os.environ.get("HERMES_HOME")
        or os.path.join(os.path.expanduser("~"), ".hermes")
    )
    cfg_path = Path(home) / "config.yaml"
    try:
        text = cfg_path.read_text()
    except Exception:
        return default
    try:
        import yaml
    except Exception:
        yaml = None
    try:
        if yaml is not None:
            cfg = yaml.safe_load(text) or {}
            compression = cfg.get("compression") or {}
            if _config_bool_disabled(compression.get("enabled")):
                return default
            value = compression.get("threshold")
            if value is None:
                return default
            return float(value)

        in_compression = False
        direct_indent = None
        compression_disabled = False
        threshold_value = None
        for raw_line in text.splitlines():
            line = raw_line.split('#', 1)[0].rstrip()
            if not line.strip():
                continue
            if not line.startswith((" ", "\t")):
                in_compression = line.strip() == "compression:"
                direct_indent = None
                continue
            if not in_compression:
                continue
            indent = len(line) - len(line.lstrip(" \t"))
            if direct_indent is None:
                direct_indent = indent
            if indent != direct_indent or ":" not in line:
                continue
            key, raw_value = line.strip().split(":", 1)
            value = raw_value.strip().strip("'\"")
            if key == "enabled" and _config_bool_disabled(value):
                compression_disabled = True
            elif key == "threshold":
                threshold_value = value
        if compression_disabled or threshold_value is None:
            return default
        return float(threshold_value)
    except Exception:
        return default

def _hermes_yaml_auxiliary_compression_timeout_ms(default, hermes_home=None):
    # Port of hermes-lcm config._hermes_auxiliary_compression_timeout_ms:
    # read auxiliary.compression.timeout (seconds) from config.yaml and expose
    # it in milliseconds for LCM summary timeout parity.
    home = (
        hermes_home
        or os.environ.get("HERMES_HOME")
        or os.path.join(os.path.expanduser("~"), ".hermes")
    )
    cfg_path = Path(home) / "config.yaml"
    try:
        text = cfg_path.read_text()
    except Exception:
        return default
    try:
        import yaml
    except Exception:
        yaml = None
    try:
        if yaml is not None:
            cfg = yaml.safe_load(text) or {}
            auxiliary = cfg.get("auxiliary") or {}
            compression = auxiliary.get("compression") or {}
            value = compression.get("timeout")
            if value is None:
                return default
            return int(float(value) * 1000)

        in_auxiliary = False
        in_compression = False
        auxiliary_indent = None
        compression_indent = None
        for raw_line in text.splitlines():
            line = raw_line.split('#', 1)[0].rstrip()
            if not line.strip():
                continue
            indent = len(line) - len(line.lstrip(" \t"))
            stripped = line.strip()
            if indent == 0:
                in_auxiliary = stripped == "auxiliary:"
                in_compression = False
                auxiliary_indent = None
                compression_indent = None
                continue
            if not in_auxiliary:
                continue
            if auxiliary_indent is None:
                auxiliary_indent = indent
            if indent == auxiliary_indent:
                if stripped == "compression:":
                    in_compression = True
                    compression_indent = None
                    continue
                in_compression = False
                compression_indent = None
                continue
            if not in_compression:
                continue
            if compression_indent is None:
                compression_indent = indent
            if indent != compression_indent or ":" not in stripped:
                continue
            key, raw_value = stripped.split(":", 1)
            if key == "timeout":
                return int(float(raw_value.strip().strip("'\"")) * 1000)
        return default
    except Exception:
        return default

def _lcm_summary_timeout_ms(config, hermes_home=None):
    raw = os.environ.get("LCM_SUMMARY_TIMEOUT_MS")
    if raw is not None:
        try:
            return int(raw)
        except (TypeError, ValueError):
            pass
    configured = _configured_int(config, "summary_timeout_ms")
    if configured is not None:
        return configured
    return _hermes_yaml_auxiliary_compression_timeout_ms(60000, hermes_home=hermes_home)

def _summary_circuit_breaker_settings(config):
    threshold = _lcm_clamped_int_setting(
        config,
        "LCM_SUMMARY_CIRCUIT_BREAKER_FAILURE_THRESHOLD",
        "summary_circuit_breaker_failure_threshold",
        default=2,
        minimum=1,
    )
    cooldown = _lcm_clamped_int_setting(
        config,
        "LCM_SUMMARY_CIRCUIT_BREAKER_COOLDOWN_SECONDS",
        "summary_circuit_breaker_cooldown_seconds",
        default=300,
        minimum=0,
    )
    return threshold, cooldown

def _lcm_context_threshold(config, hermes_home=None):
    raw = os.environ.get("LCM_CONTEXT_THRESHOLD")
    if raw is not None:
        try:
            return float(raw)
        except (TypeError, ValueError):
            pass
    configured = _configured_value(config, "context_threshold")
    if configured is not None:
        try:
            return float(configured)
        except (TypeError, ValueError):
            pass
    return _hermes_yaml_compression_threshold(0.75, hermes_home=hermes_home)

def _configured_threshold_tokens(config, hermes_home=None, context_length_override=None):
    explicit = _configured_int(config, "threshold_tokens")
    if explicit is not None:
        return explicit
    context_length = context_length_override
    if context_length is None:
        context_length = _configured_int(
            config,
            "context_length",
            "max_context_tokens",
            "model_context_tokens",
        )
    if context_length is None:
        return None
    try:
        return int(int(context_length) * float(_lcm_context_threshold(config, hermes_home=hermes_home)))
    except (TypeError, ValueError):
        return None

def _lcm_config_args(config, hermes_home=None, runtime_context_length=None) -> dict:
    context_length = runtime_context_length
    if context_length is None:
        context_length = _configured_int(
            config,
            "context_length",
            "max_context_tokens",
            "model_context_tokens",
        )
    args = {
        "fresh_tail_count": _lcm_int_setting(config, "LCM_FRESH_TAIL_COUNT", "fresh_tail_count", default=64),
        "leaf_chunk_tokens": _lcm_int_setting(config, "LCM_LEAF_CHUNK_TOKENS", "leaf_chunk_tokens", default=20000),
        "dynamic_leaf_chunk_enabled": _lcm_bool_setting(
            config,
            "LCM_DYNAMIC_LEAF_CHUNK_ENABLED",
            "dynamic_leaf_chunk_enabled",
            default=False,
        ),
        "dynamic_leaf_chunk_max": _lcm_int_setting(
            config,
            "LCM_DYNAMIC_LEAF_CHUNK_MAX",
            "dynamic_leaf_chunk_max",
            default=40000,
        ),
        "max_assembly_tokens": _lcm_int_setting(config, "LCM_MAX_ASSEMBLY_TOKENS", "max_assembly_tokens", default=0),
        # Hermes derives an assembly cap of context_length - reserve_tokens_floor
        # when both are positive; pass both through so tracedecay can apply the
        # same derivation (reserve_tokens_floor defaults to 0 = disabled).
        "reserve_tokens_floor": _lcm_reserve_tokens_floor_setting(config, context_length),
        "context_length": context_length,
        "summary_fan_in": _lcm_int_setting(
            config,
            "LCM_CONDENSATION_FANIN",
            "summary_fan_in",
            "condensation_fanin",
            default=4,
        ),
        # hermes-lcm caps condensation at depth 1 by default; pass the knob
        # through so the Rust engine can enforce the same ceiling.
        "incremental_max_depth": _lcm_int_setting(
            config,
            "LCM_INCREMENTAL_MAX_DEPTH",
            "incremental_max_depth",
            default=1,
        ),
    }
    threshold_tokens = _configured_threshold_tokens(
        config,
        hermes_home=hermes_home,
        context_length_override=context_length,
    )
    if threshold_tokens is not None:
        args["threshold_tokens"] = threshold_tokens
    for env_key, name in (
        ("LCM_IGNORE_SESSION_PATTERNS", "ignore_session_patterns"),
        ("LCM_STATELESS_SESSION_PATTERNS", "stateless_session_patterns"),
        ("LCM_IGNORE_MESSAGE_PATTERNS", "ignore_message_patterns"),
    ):
        patterns = _lcm_list_setting(config, env_key, name)
        if patterns:
            args[name] = patterns
    return {key: value for key, value in args.items() if value is not None}


def _lcm_gc_config_args(config):
    gc_config = {}
    for env_key, name, default in (
        ("LCM_PAYLOAD_GC_GRACE_SECONDS", "grace_seconds", None),
        ("LCM_PAYLOAD_REAP_MISSING_METADATA_AFTER_SECONDS", "reap_missing_after", None),
        ("LCM_PAYLOAD_REAP_MISSING_METADATA_ENABLED", "reap_missing_enabled", None),
        ("LCM_PAYLOAD_GC_MAX_BATCH_SIZE", "max_batch_size", None),
        ("LCM_PAYLOAD_GC_BACKUP_BEFORE_REAP", "backup_before_reap", None),
        ("LCM_PAYLOAD_GC_INTERVAL_SECONDS", "interval_seconds", None),
        ("LCM_PAYLOAD_GC_ENABLED", "gc_enabled", None),
    ):
        if name.endswith("enabled") or name == "backup_before_reap":
            value = _lcm_bool_setting(config, env_key, name, default=default)
        else:
            value = _lcm_int_setting(config, env_key, name, default=default)
        if value is not None:
            gc_config[name] = value
    return {"gc_config": gc_config} if gc_config else {}

def _lcm_expansion_model(config):
    value = _lcm_str_setting(config, "LCM_EXPANSION_MODEL", "expansion_model", default="")
    return str(value or "").strip()

def _lcm_clamped_int_setting(config, env_key, *names, default, minimum=1):
    value = _lcm_int_setting(config, env_key, *names, default=default)
    if value is None:
        value = default
    return max(minimum, int(value))

def _lcm_expansion_settings(config):
    return {
        "model": _lcm_expansion_model(config),
        "context_tokens": _lcm_clamped_int_setting(
            config,
            "LCM_EXPANSION_CONTEXT_TOKENS",
            "expansion_context_tokens",
            default=32000,
            minimum=1,
        ),
        "timeout_ms": _lcm_clamped_int_setting(
            config,
            "LCM_EXPANSION_TIMEOUT_MS",
            "expansion_timeout_ms",
            default=120000,
            minimum=1,
        ),
    }

def _lcm_expansion_context_tokens(config):
    return _lcm_expansion_settings(config)["context_tokens"]

def _lcm_expansion_timeout_ms(config):
    return _lcm_expansion_settings(config)["timeout_ms"]

def _lcm_extraction_settings(config):
    return {
        "enabled": bool(
            _lcm_bool_setting(
                config,
                "LCM_EXTRACTION_ENABLED",
                "extraction_enabled",
                default=False,
            )
        ),
        "model": str(
            _lcm_str_setting(config, "LCM_EXTRACTION_MODEL", "extraction_model", default="") or ""
        ).strip(),
        "output_path": str(
            _lcm_str_setting(
                config,
                "LCM_EXTRACTION_OUTPUT_PATH",
                "extraction_output_path",
                default="",
            )
            or ""
        ).strip(),
    }

def _lcm_extraction_enabled(config):
    return _lcm_extraction_settings(config)["enabled"]

def _lcm_extraction_model(config):
    return _lcm_extraction_settings(config)["model"]

def _lcm_extraction_output_path(config):
    return _lcm_extraction_settings(config)["output_path"]

def _apply_lcm_option_overrides(args: dict, kwargs: dict, keys) -> None:
    for key in keys:
        if key in kwargs and kwargs[key] is not None:
            args[key] = kwargs.pop(key)

REASONING_TAGS = ("think", "thinking", "reasoning", "thought", "REASONING_SCRATCHPAD")
FALLBACK_MARKER = "[deterministic compression fallback]"
RETRY_WORTHY_AUXILIARY_ERRORS = (
    "context length",
    "maximum context",
    "max context",
    "token limit",
    "too many tokens",
    "prompt is too long",
    "input too long",
    "request too large",
    "timed out",
    "timeout",
)

def _strip_reasoning(text: str) -> str:
    output = text or ""
    for tag in REASONING_TAGS:
        escaped = re.escape(tag)
        output = re.sub(
            rf"<{escaped}>.*?</{escaped}>",
            "",
            output,
            flags=re.IGNORECASE | re.DOTALL,
        )
    return output.strip()

def _messages_hash(messages):
    # Keep a full-content hash to preserve debounce correctness: any message
    # change must invalidate the signature and trigger preflight.
    try:
        payload = json.dumps(messages or [], sort_keys=True, ensure_ascii=False, separators=(",", ":"))
    except Exception:
        payload = repr(messages)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

def _llm_response_text(response) -> str:
    if isinstance(response, str):
        return response
    if isinstance(response, dict):
        content = response.get("content")
        if isinstance(content, str):
            return content
        choices = response.get("choices")
        if isinstance(choices, list) and choices:
            message = choices[0].get("message") if isinstance(choices[0], dict) else None
            if isinstance(message, dict) and isinstance(message.get("content"), str):
                return message["content"]
    choices = getattr(response, "choices", None)
    if choices:
        message = getattr(choices[0], "message", None)
        content = getattr(message, "content", None)
        if isinstance(content, str):
            return content
    return "" if response is None else str(response)

def _message_content(message) -> str:
    if not isinstance(message, dict):
        return str(message)
    content = message.get("content", "")
    if isinstance(content, str):
        return content
    if isinstance(content, dict) and isinstance(content.get("text"), str):
        return content["text"]
    if isinstance(content, list):
        parts = [
            item.get("text")
            for item in content
            if isinstance(item, dict) and isinstance(item.get("text"), str)
        ]
        if parts:
            return "\n\n".join(parts)
    return "" if content is None else str(content)

def _summary_source_messages(source_messages):
    normalized = []
    for message in source_messages or []:
        if not isinstance(message, dict):
            normalized.append({"role": "user", "content": str(message)})
            continue
        entry = {
            "role": message.get("role") or "user",
            "content": _message_content(message),
        }
        if message.get("tool_calls"):
            entry["tool_calls"] = message["tool_calls"]
        if message.get("tool_call_id"):
            entry["tool_call_id"] = message["tool_call_id"]
        normalized.append(entry)
    return normalized

_TOKEN_ENCODER = None
_TOKEN_ENCODER_CHECKED = False

def _token_encoder():
    global _TOKEN_ENCODER, _TOKEN_ENCODER_CHECKED
    if _TOKEN_ENCODER_CHECKED:
        return _TOKEN_ENCODER
    _TOKEN_ENCODER_CHECKED = True
    try:
        import tiktoken
        _TOKEN_ENCODER = tiktoken.get_encoding("cl100k_base")
    except Exception:
        _TOKEN_ENCODER = None
    return _TOKEN_ENCODER

def _count_tokens(text):
    # tiktoken when available; otherwise the same ceil(len/4) chars-per-token
    # estimate the Rust side uses everywhere ((len+3)//4 — see
    # dashboard/token_count.rs chars_estimate), so token numbers reported
    # from Hermes reconcile with dashboard numbers.
    if not text:
        return 0
    encoder = _token_encoder()
    if encoder is not None:
        try:
            return len(encoder.encode(text))
        except Exception:
            pass
    return (len(text) + 3) // 4

def _tool_call_arguments_text(arguments):
    if isinstance(arguments, str):
        return arguments
    if arguments is None:
        return ""
    try:
        return json.dumps(arguments, ensure_ascii=False)
    except Exception:
        return str(arguments)

def _count_message_tokens(message):
    total = 4
    if not isinstance(message, dict):
        return total + _count_tokens(str(message))
    total += _count_tokens(_message_content(message))
    for tool_call in message.get("tool_calls") or []:
        if isinstance(tool_call, dict):
            function = tool_call.get("function") or {}
            total += _count_tokens(str(function.get("name") or ""))
            total += _count_tokens(_tool_call_arguments_text(function.get("arguments")))
        total += 3
    return total

def _count_messages_tokens(messages):
    return sum(_count_message_tokens(message) for message in messages or [])

def _matched_tool_call_ids(messages):
    matched = set()
    for message in messages or []:
        if isinstance(message, dict) and message.get("role") == "tool":
            tool_id = str(message.get("tool_call_id") or "").strip()
            if tool_id:
                matched.add(tool_id)
    return matched

def _summary_tool_call_id(tool_call):
    if isinstance(tool_call, dict):
        return str(tool_call.get("id") or "").strip()
    return ""

def _truncate_serialized_content(content):
    if len(content) > 3000:
        return content[:2000] + "\n...[truncated]...\n" + content[-800:]
    return content

def _serialize_summary_messages(messages):
    # Mirrors hermes-lcm engine._serialize_messages: labeled per-role text with
    # matched tool-call enrichment and long-content truncation. Redaction and
    # externalization stay in the Rust ingest pipeline.
    parts = []
    matched_tool_ids = _matched_tool_call_ids(messages)
    for message in messages or []:
        if not isinstance(message, dict):
            parts.append(f"[USER]: {message}")
            continue
        role = str(message.get("role") or "unknown")
        content = _message_content(message)
        if role == "tool":
            tool_id = str(message.get("tool_call_id") or "").strip()
            parts.append(f"[TOOL RESULT {tool_id}]: {_truncate_serialized_content(content)}")
            continue
        if role == "assistant":
            tool_calls = message.get("tool_calls") or []
            matched_tool_calls = [
                tool_call
                for tool_call in tool_calls
                if not _summary_tool_call_id(tool_call)
                or _summary_tool_call_id(tool_call) in matched_tool_ids
            ]
            content = _truncate_serialized_content(content)
            if matched_tool_calls:
                tool_call_parts = []
                for tool_call in matched_tool_calls:
                    if isinstance(tool_call, dict):
                        function = tool_call.get("function") or {}
                        name = function.get("name") or "?"
                        arguments = _tool_call_arguments_text(function.get("arguments"))
                        if len(arguments) > 500:
                            arguments = arguments[:400] + "..."
                        tool_call_parts.append(f"  {name}({arguments})")
                content += "\n[Tool calls:\n" + "\n".join(tool_call_parts) + "\n]"
            parts.append(f"[ASSISTANT]: {content}")
            continue
        parts.append(f"[{role.upper()}]: {_truncate_serialized_content(content)}")
    return "\n\n".join(parts)

def _normalized_focus_topic(focus_topic, max_chars=160):
    normalized = " ".join(str(focus_topic or "").split())
    if len(normalized) <= max_chars:
        return normalized
    return normalized[: max(0, max_chars - 1)].rstrip() + "…"

def _build_l1_focus_brief(focus_topic):
    topic = _normalized_focus_topic(focus_topic)
    if not topic:
        return ""
    return (
        "Focus brief:\n"
        f"Primary focus: {topic}\n"
        "Preserve concrete decisions, constraints, files, commands, identifiers, and current state for this focus.\n"
        "Spend roughly 60-70% of the summary budget on the focus when relevant.\n"
        "Do not discard unrelated blockers or active tasks just because they are off-focus.\n"
    )

def _build_l2_focus_brief(focus_topic):
    topic = _normalized_focus_topic(focus_topic)
    if not topic:
        return ""
    return (
        "Focus brief:\n"
        f"Primary focus: {topic}\n"
        "Prefer bullets that preserve decisions, blockers, files, commands, identifiers, and current state for this focus.\n"
        "Keep other active tasks only when they are current blockers or handoff state.\n"
    )

_L1_DEPTH_GUIDANCE = {
    0: "Preserve decisions, rationale, constraints, active tasks, file paths, commands, and specific values.",
    1: "Distill into arc-level outcomes: what evolved, what was decided, current state. Drop per-turn detail.",
    2: "Capture durable narrative: decisions in effect, completed milestones, timeline. Drop process detail.",
}

def _build_l1_prompt(text, token_budget, depth, focus_topic="", custom_instructions=""):
    guidance = _L1_DEPTH_GUIDANCE.get(depth, _L1_DEPTH_GUIDANCE[2])
    focus_guidance = _build_l1_focus_brief(focus_topic)
    custom_block = ""
    if custom_instructions:
        custom_block = f"\nAdditional instructions:\n{custom_instructions}\n"
    return f"""Summarize this conversation segment for future turns.
{guidance}
Remove repetition and conversational filler.
End with: "Expand for details about: <what was compressed>"
{focus_guidance}{custom_block}

Target ~{token_budget} tokens.

CONTENT:
{text}"""

def _build_l2_prompt(text, token_budget, focus_topic="", custom_instructions=""):
    focus_guidance = _build_l2_focus_brief(focus_topic)
    custom_block = ""
    if custom_instructions:
        custom_block = f"\nAdditional instructions:\n{custom_instructions}\n"
    return f"""Compress this into bullet points. Maximum {token_budget} tokens.
Keep only: decisions made, files changed, errors hit, current state.
Drop all reasoning, alternatives considered, and process detail.
{focus_guidance}{custom_block}

CONTENT:
{text}"""

# Conservative allowlist mirroring hermes-lcm model_routing._PROVIDER_PREFIXES:
# many registry provider IDs double as OpenRouter model namespaces, so only
# explicit entries (plus non-canonical named custom providers) split into
# provider/model routes.
_LCM_PROVIDER_ROUTE_PREFIXES = frozenset({"cerebras"})

def _provider_route_is_resolvable(provider):
    provider = str(provider or "").strip().lower()
    if not provider:
        return False
    if provider.startswith("custom:"):
        provider = provider.split(":", 1)[1].strip()
        if not provider:
            return False
    try:
        from hermes_cli.auth import PROVIDER_REGISTRY
        if provider in PROVIDER_REGISTRY:
            return provider in _LCM_PROVIDER_ROUTE_PREFIXES
    except Exception:
        pass
    try:
        from hermes_cli.runtime_provider import _get_named_custom_provider
        if _get_named_custom_provider(provider):
            return True
    except Exception:
        pass
    return False

def _parse_lcm_model_override(value):
    model = str(value or "").strip()
    if not model:
        return None, ""
    provider, separator, rest = model.partition("/")
    provider = provider.strip().lower()
    rest = rest.strip()
    route_provider = provider
    if provider.startswith("custom:"):
        route_provider = provider.split(":", 1)[1].strip()
    if separator and rest and route_provider and _provider_route_is_resolvable(route_provider):
        return route_provider, rest
    return None, model

def _apply_lcm_model_route(call_kwargs, model):
    # Mirrors hermes-lcm model_routing.apply_lcm_model_route.
    provider, routed_model = _parse_lcm_model_override(model)
    if provider:
        call_kwargs["provider"] = provider
    if routed_model:
        call_kwargs["model"] = routed_model

def _deterministic_truncation(messages, limit: int = 2048) -> str:
    lines = []
    for message in messages or []:
        if isinstance(message, dict):
            role = message.get("role") or "user"
            content = _message_content(message)
        else:
            role = "user"
            content = str(message)
        if content:
            lines.append(f"{role}: {content}")
    text = "\n".join(lines).strip()
    if not text:
        text = "No auxiliary summary was available."
    max_prefix = max(0, limit - len(FALLBACK_MARKER) - 2)
    return f"{text[:max_prefix].rstrip()}\n\n{FALLBACK_MARKER}"

def _auxiliary_error_classification(error) -> str:
    message = str(error or "").lower()
    if any(pattern in message for pattern in RETRY_WORTHY_AUXILIARY_ERRORS):
        return "retry_worthy"
    return "permanent"

def _auxiliary_retry_limit(kwargs) -> int:
    try:
        limit = int(kwargs.pop("max_auxiliary_attempts", 2) or 2)
    except Exception:
        limit = 2
    return min(max(limit, 1), 8)

def _next_smaller_source_limit(source_messages, current_limit=None):
    source_count = len(source_messages or [])
    if source_count <= 1:
        return None
    next_limit = max(1, source_count // 2)
    if current_limit is not None:
        try:
            next_limit = min(next_limit, int(current_limit))
        except Exception:
            pass
    if next_limit >= source_count:
        next_limit = source_count - 1
    return max(1, next_limit)

def _normalize_extraction_items(text):
    cleaned = str(text or "").strip()
    if not cleaned:
        return []
    items = []
    for line in cleaned.splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        if stripped.startswith(("- ", "* ")):
            stripped = stripped[2:].strip()
        items.append(stripped)
    if not items:
        items = [cleaned]
    return items

def _extraction_route_payload(route, extraction_result):
    if extraction_result is None:
        return route
    # Route-envelope contract with Rust extraction.rs:
    # keys are `route` and `pre_compaction_extraction`.
    payload = {"pre_compaction_extraction": extraction_result}
    if route:
        payload["route"] = route
    return json.dumps(payload, ensure_ascii=False)

def _with_auxiliary_metadata(
    result,
    *,
    attempts,
    retry_status=None,
    error_classification=None,
    fallback_used=False,
):
    if not isinstance(result, dict):
        result = {}
    if attempts or retry_status is not None or error_classification is not None or fallback_used:
        result.setdefault("auxiliary_attempts", attempts)
    if retry_status is not None:
        result.setdefault("auxiliary_retry_status", retry_status)
    if error_classification is not None:
        result.setdefault("auxiliary_error_classification", error_classification)
    if fallback_used:
        result["fallback_used"] = True
    return result

def _auxiliary_error_result(first, *, attempts, retry_status, error_classification, error):
    result = {}
    if isinstance(first, dict):
        for key in (
            "summary_nodes_created",
            "summary_nodes",
            "replay_messages",
            "replay_token_estimate",
            "replay_over_budget",
            "context_recovery_hint",
            "frontier",
            "summary_request",
        ):
            if key in first:
                result[key] = first[key]
    result.setdefault("summary_nodes_created", 0)
    result.setdefault("summary_nodes", [])
    result.setdefault("replay_messages", [])
    result.setdefault("frontier", {"current_frontier_store_id": None, "maintenance_debt": []})
    result["status"] = "error"
    result["reason"] = (
        "auxiliary_summary_permanent_failure"
        if error_classification == "permanent"
        else "auxiliary_summary_retry_exhausted"
    )
    result["error"] = str(error)
    return _with_auxiliary_metadata(
        result,
        attempts=attempts,
        retry_status=retry_status,
        error_classification=error_classification,
    )

def _replay_message_list(value):
    """Validate a replay_messages payload as an OpenAI-format message list.

    Returns a list copy when every entry is a role-tagged dict, else None.
    """
    if not isinstance(value, list):
        return None
    for item in value:
        if not isinstance(item, dict) or not item.get("role"):
            return None
    return list(value)

def _compression_replay_is_compacted(result):
    if not isinstance(result, dict):
        return False
    return any(bool(result.get(key)) for key in (
        "contract_truncated",
        "replay_messages_truncated_for_mcp",
        "replay_messages_compacted_for_mcp",
    ))

def _compacted_lcm_result_error(result):
    payload = dict(result) if isinstance(result, dict) else {}
    payload["status"] = "error"
    payload.setdefault("reason", "mcp_compacted_compression_payload")
    payload.setdefault(
        "error",
        payload.get("mcp_truncation_reason")
        or "LCM compression payload was compacted for MCP; full replay was not recovered",
    )
    return payload

def _bounded_expand_query_answer(text: str, max_tokens: int):
    try:
        token_budget = int(max_tokens or 2000)
    except Exception:
        token_budget = 2000
    char_limit = max(1, token_budget) * 4
    answer = (text or "").strip()
    if len(answer) <= char_limit:
        return answer, False
    return answer[:char_limit].rstrip(), True

def _expand_query_degraded_payload(retrieval, reason: str, *, timeout_seconds=None):
    payload = {}
    if isinstance(retrieval, dict):
        for key in (
            "status",
            "prompt",
            "query",
            "model",
            "max_tokens",
            "context_max_tokens",
            "context_truncated",
            "context_pagination",
            "node_ids",
            "matches",
            "provider",
            "session_id",
            "storage_scope",
        ):
            if key in retrieval:
                payload[key] = retrieval[key]
    payload["status"] = payload.get("status") or "ok"
    payload["needs_synthesis"] = False
    payload["degraded"] = True
    payload["error"] = reason
    if timeout_seconds is not None:
        payload["timeout_seconds"] = timeout_seconds
    return payload

def _synthesize_expand_query_payload(retrieval, agent=None, **kwargs):
    if not isinstance(retrieval, dict) or not retrieval.get("needs_synthesis"):
        return retrieval
    client = _resolve_auxiliary_client(agent)
    if client is None or not callable(getattr(client, "call_llm", None)):
        return _expand_query_degraded_payload(
            retrieval,
            "Hermes auxiliary_client.call_llm is unavailable",
        )

    synthesis_prompt = retrieval.get("synthesis_prompt") or {}
    context_blocks = retrieval.get("context_blocks") or []
    system_prompt = synthesis_prompt.get("system") or (
        "You answer questions using expanded LCM retrieval context. "
        "Be concise, factual, and grounded in the provided context. "
        "If the context is insufficient, say so plainly."
    )
    user_prompt = synthesis_prompt.get("user") or (
        f"QUESTION:\n{retrieval.get('prompt', '')}\n\n"
        "EXPANDED CONTEXT:\n"
        f"{json.dumps(context_blocks, ensure_ascii=False, indent=2)}"
    )
    max_tokens = retrieval.get("max_tokens") or kwargs.get("max_tokens") or 2000
    timeout = kwargs.get("timeout") or kwargs.get("expansion_timeout") or 60
    call_kwargs = {
        "task": "compression",
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        "max_tokens": max_tokens,
        "timeout": timeout,
    }
    model = kwargs.get("model") or retrieval.get("model")
    if model:
        call_kwargs["model"] = model
    try:
        response = client.call_llm(**call_kwargs)
    except TimeoutError:
        return _expand_query_degraded_payload(
            retrieval,
            f"lcm_expand_query synthesis timed out after {float(timeout):.3g}s",
            timeout_seconds=timeout,
        )
    except Exception as exc:
        # The auxiliary client raises RuntimeError / provider SDK / httpx
        # errors too; letting those escape loses the retrieval entirely
        # behind a generic registry error. Degrade with the retrieval
        # payload intact instead.
        return _expand_query_degraded_payload(
            retrieval,
            f"lcm_expand_query synthesis failed: {exc}",
        )

    answer = _strip_reasoning(_llm_response_text(response)).strip()
    if not answer:
        return _expand_query_degraded_payload(
            retrieval,
            "lcm_expand_query synthesis returned an empty answer",
        )
    bounded_answer, truncated = _bounded_expand_query_answer(answer, max_tokens)
    payload = dict(retrieval)
    payload.pop("context_blocks", None)
    payload.pop("synthesis_prompt", None)
    payload["status"] = payload.get("status") or "ok"
    payload["needs_synthesis"] = False
    payload["answer"] = bounded_answer
    if truncated:
        payload["answer_truncated"] = True
    return payload

def _handle_lcm_expand_query(args, **kwargs) -> str:
    kwargs = dict(kwargs)
    agent = kwargs.pop("agent", None)
    args = dict(args or {})
    args.setdefault("provider", STANDARD_HERMES_LCM_PROVIDER)
    retrieval = call_tracedecay_json("tracedecay_lcm_expand_query", args, **kwargs)
    payload = _synthesize_expand_query_payload(retrieval, agent=agent, **kwargs)
    return json.dumps(payload)

def _copy_without_none(source: dict) -> dict:
    return {key: value for key, value in source.items() if value is not None}

def _tokens_from_native_max(max_tokens):
    if max_tokens is None:
        return None
    try:
        return max(1, min(8192, int(max_tokens) * 4))
    except Exception:
        return None

def _native_expand_target(args: dict):
    provided = [key for key in ("node_id", "store_id", "externalized_ref") if args.get(key) is not None]
    if len(provided) > 1:
        return None, "lcm_expand expects exactly one of node_id, store_id, or externalized_ref"
    if not provided:
        return None, None
    key = provided[0]
    if key == "node_id":
        return {"kind": "summary_node", "node_id": str(args[key])}, None
    if key == "store_id":
        return {"kind": "raw_message", "store_id": args[key]}, None
    return {"kind": "external_payload", "payload_ref": args[key]}, None

def _native_describe_target(args: dict):
    provided = [key for key in ("node_id", "externalized_ref") if args.get(key) is not None]
    if len(provided) > 1:
        return None, "lcm_describe expects at most one of node_id or externalized_ref"
    if not provided:
        return {"kind": "session"}, None
    key = provided[0]
    if key == "node_id":
        return {"kind": "summary_node", "node_id": str(args[key])}, None
    return {"kind": "external_payload", "payload_ref": args[key]}, None

def _translate_lcm_args(native_name: str, args: dict) -> dict:
    translated = dict(args or {})
    if native_name == "lcm_grep":
        if "session_scope" in translated:
            translated["scope"] = translated.pop("session_scope")
        else:
            translated.setdefault("scope", "current")
        if "time_from" in translated:
            translated["start_time"] = translated.pop("time_from")
        if "time_to" in translated:
            translated["end_time"] = translated.pop("time_to")
        return translated
    if native_name == "lcm_load_session":
        if "max_content_chars" in translated:
            translated["content_limit"] = translated.pop("max_content_chars")
        if "time_from" in translated:
            translated["start_time"] = translated.pop("time_from")
        if "time_to" in translated:
            translated["end_time"] = translated.pop("time_to")
        return translated
    if native_name == "lcm_describe":
        if "target" not in translated:
            target, error = _native_describe_target(translated)
            if error is not None:
                return {"error": error}
            translated["target"] = target
        translated.pop("node_id", None)
        translated.pop("externalized_ref", None)
        return translated
    if native_name == "lcm_expand":
        if "target" not in translated:
            target, error = _native_expand_target(translated)
            if error is not None:
                return {"error": error}
            if target is not None:
                translated["target"] = target
        for public_key in ("node_id", "store_id", "externalized_ref"):
            translated.pop(public_key, None)
        content_limit = _tokens_from_native_max(translated.pop("max_tokens", None))
        if content_limit is not None and "content_limit" not in translated:
            translated["content_limit"] = content_limit
        return translated
    return translated

_ENGINE_DEFAULT_SESSION = "__default__"

class _EngineSessionState:
    """Mutable per-conversation engine state.

    The host registers ONE engine instance and shares it across every
    AIAgent in the process (gateway sessions, parallel delegate_task
    children), so conversation-scoped fields must never live directly on
    the engine.
    """

    _FIELDS = (
        "agent",
        "model",
        "last_prompt_tokens",
        "last_real_prompt_tokens",
        "last_completion_tokens",
        "last_total_tokens",
        "context_length",
        "threshold_tokens",
        "compression_count",
        "last_compress_result",
        "_last_compress_aborted",
        "_last_summary_error",
        "_runtime_context_length",
        "_session_start_context_length",
        "_last_preflight_signature",
    )

    def __init__(self):
        self.agent = None
        self.model = ""
        self.last_prompt_tokens = 0
        self.last_real_prompt_tokens = 0
        self.last_completion_tokens = 0
        self.last_total_tokens = 0
        # Host-contract token state (run_agent.py reads these directly; the
        # minimum-context guard in agent/agent_init.py checks context_length).
        self.context_length = 0
        self.threshold_tokens = 0
        self.compression_count = 0
        # Compression abort/diagnostic state read by
        # agent/conversation_compression.py after each compress() call.
        self.last_compress_result = None
        self._last_compress_aborted = False
        self._last_summary_error = None
        self._runtime_context_length = None
        self._session_start_context_length = None
        self._last_preflight_signature = None

    def adopt(self, other):
        """Carry conversation state across a compression-rotation rebind."""
        for field in self._FIELDS:
            setattr(self, field, getattr(other, field))

def _engine_session_property(field):
    """Engine attribute view onto the calling context's session state.

    Keeps the host contract intact: run_agent.py reads (and
    conversation_compression.py occasionally writes) these as plain
    attributes on the shared engine instance, and each access resolves to
    the session bound to the calling thread.
    """

    def _get(self):
        return getattr(self._state(), field)

    def _set(self, value):
        setattr(self._state(), field, value)

    return property(_get, _set)

class TraceDecayContextEngine(ContextEngine):
    def __init__(self, config=None, hermes_home=None):
        # Per-session mutable state, guarded by an RLock. Calls that carry
        # no session id (compress, update_from_response, should_compress,
        # ...) resolve the session bound to the calling thread, falling
        # back to the most recently bound session. That keeps concurrent
        # gateway sessions / delegate_task children from clobbering each
        # other whenever the host gives us a binding signal, and at minimum
        # serializes state access when it does not.
        self._state_lock = threading.RLock()
        self._session_states = {}
        self._thread_sessions = threading.local()
        self.active_session_id = None
        self._host_config = config
        self.hermes_home = _resolve_hermes_home(config, hermes_home)
        self.config = _with_plugin_block(config, self.hermes_home)
        self.project_root = _configured_project_root(self.config)
        # Auxiliary-route circuit breakers are process-global on purpose:
        # a broken summary model is broken for every session.
        self._route_failures = {}
        self._cooldown_until = {}

    agent = _engine_session_property("agent")
    model = _engine_session_property("model")
    last_prompt_tokens = _engine_session_property("last_prompt_tokens")
    last_real_prompt_tokens = _engine_session_property("last_real_prompt_tokens")
    last_completion_tokens = _engine_session_property("last_completion_tokens")
    last_total_tokens = _engine_session_property("last_total_tokens")
    context_length = _engine_session_property("context_length")
    threshold_tokens = _engine_session_property("threshold_tokens")
    compression_count = _engine_session_property("compression_count")
    last_compress_result = _engine_session_property("last_compress_result")
    _last_compress_aborted = _engine_session_property("_last_compress_aborted")
    _last_summary_error = _engine_session_property("_last_summary_error")
    _runtime_context_length = _engine_session_property("_runtime_context_length")
    _session_start_context_length = _engine_session_property("_session_start_context_length")
    _last_preflight_signature = _engine_session_property("_last_preflight_signature")

    def _session_key(self, session_id=None):
        if session_id:
            return str(session_id)
        bound = getattr(self._thread_sessions, "session_id", None)
        if bound:
            return bound
        return self.active_session_id or _ENGINE_DEFAULT_SESSION

    def _state(self, session_id=None):
        key = self._session_key(session_id)
        with self._state_lock:
            state = self._session_states.get(key)
            if state is None:
                state = _EngineSessionState()
                self._session_states[key] = state
            return state

    @property
    def name(self) -> str:
        return "tracedecay"

    def _bind_session(self, session_id=None, hermes_home=None, project_root=None, **kwargs):
        if session_id is not None:
            session_key = str(session_id)
            with self._state_lock:
                if session_key not in self._session_states:
                    state = _EngineSessionState()
                    # Compression rotation continues the same conversation:
                    # carry model/window/counters over from the predecessor
                    # session's state (old_session_id on boundary starts,
                    # else whatever this thread was bound to).
                    source_key = str(kwargs.get("old_session_id") or "") or self._session_key()
                    source = self._session_states.get(source_key)
                    if source is not None:
                        state.adopt(source)
                        state._last_preflight_signature = None
                    self._session_states[session_key] = state
            # Bind the calling thread so session-less calls (compress,
            # update_from_response, should_compress) resolve this session.
            self._thread_sessions.session_id = session_key
            self.active_session_id = session_key
        if kwargs.get("config") is not None:
            self._host_config = kwargs.get("config")
        if "context_length" in kwargs:
            applied_context_length = self._apply_context_length(kwargs.get("context_length"))
            if applied_context_length is not None:
                self._session_start_context_length = applied_context_length
        next_agent = kwargs.get("agent")
        if next_agent is not None:
            self.agent = next_agent
        explicit_hermes_home = hermes_home or kwargs.get("hermes_home")
        if explicit_hermes_home or kwargs.get("config") is not None or self.hermes_home is None:
            next_hermes_home = _resolve_hermes_home(
                kwargs.get("config", self._host_config),
                explicit_hermes_home,
            )
            if next_hermes_home:
                self.hermes_home = next_hermes_home
        # Re-layer the profile's plugins.tracedecay block now that the host
        # config and hermes_home are settled for this session.
        self.config = _with_plugin_block(self._host_config, self.hermes_home)
        next_project_root = _code_project_root(
            explicit=project_root or kwargs.get("project_root"),
            cwd=kwargs.get("cwd"),
            configured=_configured_project_root(self.config),
        )
        if next_project_root:
            self.project_root = next_project_root

    def initialize(self, session_id=None, hermes_home=None, project_root=None, **kwargs):
        self._bind_session(session_id, hermes_home, project_root, **kwargs)

    def on_session_start(self, session_id=None, hermes_home=None, project_root=None, **kwargs):
        bound_session_id = self.active_session_id
        self._bind_session(session_id, hermes_home, project_root, **kwargs)
        self._report_compression_boundary(session_id, bound_session_id, kwargs)

    def on_session_end(self, session_id=None, messages=None, **kwargs):
        # Real session boundary: drop the per-session record so long-lived
        # gateway processes do not accumulate dead conversation state.
        key = self._session_key(session_id)
        with self._state_lock:
            self._session_states.pop(key, None)
        if getattr(self._thread_sessions, "session_id", None) == key:
            self._thread_sessions.session_id = None
        if self.active_session_id == key:
            self.active_session_id = None

    def _apply_context_length(self, context_length):
        """Track a context length on the host-contract attrs.

        run_agent.py logs ``context_length``/``threshold_tokens`` directly and
        the minimum-context guard in agent/agent_init.py reads
        ``context_length`` — leaving them 0 logged a bogus 0-token window and
        silently bypassed that guard.
        """
        try:
            parsed = int(context_length)
        except (TypeError, ValueError):
            return None
        if parsed <= 0:
            return None
        self.context_length = parsed
        threshold_tokens = _configured_threshold_tokens(
            self.config,
            hermes_home=self.hermes_home,
            context_length_override=parsed,
        )
        if threshold_tokens is None:
            threshold_tokens = int(
                parsed * _lcm_context_threshold(self.config, hermes_home=self.hermes_home)
            )
        self.threshold_tokens = threshold_tokens
        return parsed

    def update_model(self, model, context_length, base_url="", api_key="", provider="", api_mode=""):
        self.model = str(model or "")
        self._runtime_context_length = self._apply_context_length(context_length)

    def update_from_response(self, usage):
        # Required by newer Hermes ContextEngine ABCs (abstract method).
        # Tracks normalized token usage; run_agent.py reads these attrs.
        usage = usage or {}

        def _as_int(value):
            try:
                return int(value)
            except (TypeError, ValueError):
                return 0

        self.last_prompt_tokens = _as_int(
            usage.get("prompt_tokens") or usage.get("input_tokens")
        )
        self.last_real_prompt_tokens = self.last_prompt_tokens
        self.last_completion_tokens = _as_int(
            usage.get("completion_tokens") or usage.get("output_tokens")
        )
        self.last_total_tokens = _as_int(usage.get("total_tokens")) or (
            self.last_prompt_tokens + self.last_completion_tokens
        )

    def _effective_context_length(self):
        if self._runtime_context_length is not None:
            return self._runtime_context_length
        if self._session_start_context_length is not None:
            return self._session_start_context_length
        return None

    def _tool_args(self, session_id=None):
        args = _storage_args(self.project_root, self.hermes_home)
        args["session_id"] = session_id if session_id is not None else self.active_session_id
        return args

    def _report_compression_boundary(self, session_id, bound_session_id, kwargs):
        # Mirrors Hermes' compression-boundary session starts: hand the
        # bound/old session ids to tracedecay so it can record a boundary-skip
        # cooldown when carry-over did not continue from the bound session.
        boundary_reason = str(kwargs.get("boundary_reason") or "")
        old_session_id = str(kwargs.get("old_session_id") or "")
        if (
            boundary_reason != "compression"
            or not old_session_id
            or not session_id
            or old_session_id == session_id
        ):
            return
        args = _storage_args(self.project_root, self.hermes_home)
        args.update({
            "provider": STANDARD_HERMES_LCM_PROVIDER,
            "session_id": session_id,
            "old_session_id": old_session_id,
            "boundary_reason": boundary_reason,
        })
        if bound_session_id:
            args["bound_session_id"] = bound_session_id
        try:
            tools.call_tracedecay_tool("tracedecay_lcm_session_boundary", args)
        except Exception as exc:
            logger.warning("LCM session boundary report failed: %s", exc)

    def should_compress_preflight(self, messages, current_tokens=None, **kwargs):
        """ABC contract: quick pre-flight check returning a BOOL.

        The dict-shaped probe (status/reason/replay_messages) stays
        available as ``_preflight_probe`` for internal callers — an error
        dict returned here would be truthy and trigger bogus compactions.
        """
        result = self._preflight_probe(messages, current_tokens=current_tokens, **kwargs)
        if isinstance(result, dict):
            original = list(messages or [])
            replay = _replay_message_list(result.get("replay_messages"))
            if (
                replay is not None
                and replay != original
                and (replay or not original)
                and not _compression_replay_is_compacted(result)
            ):
                return True
            return bool(result.get("should_compress"))
        return False

    def _preflight_probe(self, messages, current_tokens=None, **kwargs):
        args = self._tool_args()
        args.update(
            _lcm_config_args(
                self.config,
                self.hermes_home,
                runtime_context_length=self._effective_context_length(),
            )
        )
        args.update({
            "provider": STANDARD_HERMES_LCM_PROVIDER,
            "session_id": self.active_session_id,
            "messages": messages,
            "current_tokens": current_tokens,
        })
        _apply_lcm_option_overrides(args, kwargs, (
            "threshold_tokens",
            "max_assembly_tokens",
            "leaf_chunk_tokens",
            "max_source_messages",
            "summary_fan_in",
            "incremental_max_depth",
            "fresh_tail_count",
            "dynamic_leaf_chunk_enabled",
            "dynamic_leaf_chunk_max",
            "context_length",
            "reserve_tokens_floor",
            "ignore_session_patterns",
            "stateless_session_patterns",
            "ignore_message_patterns",
        ))
        return call_tracedecay_json("tracedecay_lcm_preflight", args, **kwargs)

    def should_compress(self, prompt_tokens=None, **kwargs):
        # The host probes this on EVERY agent-loop iteration (up to ~90
        # times per turn). Below the tracked threshold the subprocess
        # answer is always "no compression", so gate locally and only pay
        # the spawn when the conversation is actually near its budget (or
        # when no threshold is known yet).
        state = self._state()
        try:
            tokens = int(prompt_tokens) if prompt_tokens is not None else None
        except (TypeError, ValueError):
            tokens = None
        if tokens is not None and state.threshold_tokens and tokens < state.threshold_tokens:
            return False
        response = self._preflight_probe([], current_tokens=prompt_tokens, **kwargs)
        if isinstance(response, dict):
            return bool(response.get("should_compress"))
        return False

    def has_content_to_compress(self, messages, current_tokens=None, **kwargs):
        del current_tokens, kwargs
        non_empty = [
            message
            for message in (messages or [])
            if str((message or {}).get("content") or "").strip()
        ]
        return len(non_empty) >= 2

    def should_defer_preflight_to_real_usage(self, rough_tokens=None):
        if not (
            isinstance(self.last_compress_result, dict)
            and self.last_compress_result.get("status") == "compressed"
        ):
            return False
        try:
            rough = int(rough_tokens or 0)
            real = int(self.last_real_prompt_tokens or 0)
        except (TypeError, ValueError):
            return False
        if real <= 0 or rough <= real:
            return False
        # Defer only when the rough estimate is plausibly near the last real
        # usage; very large estimates should still trigger a backend preflight.
        return rough <= real * 10

    def carry_over_new_session_context(self, old_session_id, new_session_id):
        old_session_id = str(old_session_id or "")
        new_session_id = str(new_session_id or "")
        if not new_session_id:
            return
        self._bind_session(new_session_id, old_session_id=old_session_id)
        self._report_compression_boundary(
            new_session_id,
            old_session_id,
            {"boundary_reason": "compression", "old_session_id": old_session_id},
        )

    def status(self, session_id=None, **kwargs):
        args = self._tool_args(session_id)
        args.update(_lcm_gc_config_args(self.config))
        return call_tracedecay_json("tracedecay_lcm_status", args, **kwargs)

    def get_tool_schemas(self):
        return _lcm_tool_schemas()

    def get_status(self):
        storage = _storage_args(self.project_root, self.hermes_home)
        project_root = storage.get("project_root")
        last_result = self.last_compress_result
        if not isinstance(last_result, dict):
            last_result = {"status": "never_ran"}
        now = time.time()
        route_cooldowns = {
            route: max(0, int(deadline - now))
            for route, deadline in self._cooldown_until.items()
            if deadline > now
        }
        return {
            "engine": self.name,
            "session_id": self.active_session_id,
            "active_session_id": self.active_session_id,
            "storage_scope": "profile_sharded",
            "hermes_home": self.hermes_home,
            "lcm_project_root": project_root,
            "lcm_session_db_path": None,
            "storage_note": (
                "Hermes LCM conversation state is stored in the unified "
                "user-level tracedecay store; unpinned profiles use hermes_profile storage."
            ),
            "project_root": self.project_root,
            "tracedecay_binary_path": tools.TRACEDECAY_BIN,
            "tracedecay_binary_available": _tracedecay_binary_available(),
            "context_engine_tool_names": sorted(
                schema["name"] for schema in self.get_tool_schemas()
            ),
            "route_failures": dict(self._route_failures),
            "cooldown_routes": sorted(self._cooldown_until.keys()),
            "route_cooldowns": route_cooldowns,
            "last_compress_result": last_result,
            "awaiting_real_usage_after_compression": (
                isinstance(self.last_compress_result, dict)
                and self.last_compress_result.get("status") == "compressed"
                and not self.last_real_prompt_tokens
            ),
            "live_ingest": {
                "registered_tool_names": sorted(_REGISTERED_TOOL_NAMES),
                "context_tool_names": sorted(_CONTEXT_TOOL_NAMES),
                "host_forwards_messages": _HOST_FORWARDS_MESSAGES,
                "message_dependent_tools_registered": bool(
                    MESSAGE_DEPENDENT_TOOLS.intersection(_REGISTERED_TOOL_NAMES)
                ),
                "gate_reason": (
                    "not_registered"
                    if not _REGISTERED_TOOL_NAMES and not _CONTEXT_TOOL_NAMES
                    else (
                        "host_does_not_forward_messages"
                        if _HOST_FORWARDS_MESSAGES is False
                        else "registered"
                    )
                ),
            },
        }

    def _current_turn_preflight(self, messages, **kwargs):
        if not messages or not self.active_session_id:
            return
        signature = f"{self.active_session_id}:{_messages_hash(messages)}"
        if signature == self._last_preflight_signature:
            return
        args = _storage_args(self.project_root, self.hermes_home)
        args.update(
            _lcm_config_args(
                self.config,
                self.hermes_home,
                runtime_context_length=self._effective_context_length(),
            )
        )
        args.update({
            "session_id": self.active_session_id,
            "messages": messages,
        })
        _apply_lcm_option_overrides(args, kwargs, (
            "current_tokens",
            "threshold_tokens",
            "max_assembly_tokens",
            "leaf_chunk_tokens",
            "max_source_messages",
            "summary_fan_in",
            "incremental_max_depth",
            "fresh_tail_count",
            "dynamic_leaf_chunk_enabled",
            "dynamic_leaf_chunk_max",
            "context_length",
            "reserve_tokens_floor",
            "ignore_session_patterns",
            "stateless_session_patterns",
            "ignore_message_patterns",
        ))
        try:
            tools.call_tracedecay_tool("tracedecay_lcm_preflight", args, **_copy_without_none({
                "project_root": kwargs.get("project_root"),
            }))
            self._last_preflight_signature = signature
        except Exception as exc:
            logger.warning("LCM current-turn preflight failed: %s", exc)

    def handle_tool_call(self, name, arguments=None, **kwargs) -> str:
        tool_name, tool_args = _normalize_memory_tool_call(name, arguments)
        native_name = tool_name
        tracedecay_name = LCM_TOOL_ALIASES.get(native_name)
        if tracedecay_name is None and native_name in LCM_DIRECT_TOOL_NAMES:
            tracedecay_name = native_name
            native_name = LCM_DIRECT_TO_NATIVE.get(native_name, native_name)
        if tracedecay_name is None:
            return tools.error_payload(f"unknown LCM tool: {tool_name}")

        messages = kwargs.get("messages")
        preflight_kwargs = dict(kwargs)
        preflight_kwargs.pop("messages", None)
        self._current_turn_preflight(messages, **preflight_kwargs)

        tool_args = _translate_lcm_args(native_name, dict(tool_args))
        if tool_args.get("error"):
            return json.dumps({"error": tool_args["error"]})
        storage_args = _storage_args(self.project_root, self.hermes_home)
        for key, value in storage_args.items():
            tool_args.setdefault(key, value)
        if tracedecay_name in LCM_PROVIDER_LOCAL_TOOL_NAMES:
            tool_args.setdefault("provider", STANDARD_HERMES_LCM_PROVIDER)
        if tracedecay_name == "tracedecay_lcm_compress" and self.project_root:
            tool_args.setdefault("response_handle_project_root", self.project_root)
        if native_name in ("lcm_status", "lcm_doctor"):
            tool_args.update(_lcm_gc_config_args(self.config))
        if self.active_session_id:
            tool_args.setdefault("session_id", self.active_session_id)

        if tracedecay_name == "tracedecay_lcm_expand_query":
            expand_kwargs = dict(preflight_kwargs)
            agent = expand_kwargs.pop("agent", None) or self.agent
            return _handle_lcm_expand_query(tool_args, agent=agent, **expand_kwargs)
        return tools.call_tracedecay_tool(tracedecay_name, tool_args, **preflight_kwargs)

    def expand_query(self, prompt, query=None, node_ids=None, **kwargs):
        kwargs = dict(kwargs)
        args = self._tool_args(kwargs.pop("session_id", None))
        args["provider"] = STANDARD_HERMES_LCM_PROVIDER
        args["prompt"] = prompt
        if query is not None:
            args["query"] = query
        if node_ids is not None:
            args["node_ids"] = node_ids
        for key in ("max_results", "max_tokens", "context_max_tokens"):
            if key in kwargs and kwargs[key] is not None:
                args[key] = kwargs[key]
        if "context_max_tokens" not in args:
            args["context_max_tokens"] = _lcm_expansion_context_tokens(self.config)
        retrieval = call_tracedecay_json("tracedecay_lcm_expand_query", args, **kwargs)
        synthesis_kwargs = dict(kwargs)
        synthesis_agent = synthesis_kwargs.pop("agent", None) or self.agent
        if synthesis_kwargs.get("model") is None:
            expansion_model = _lcm_expansion_model(self.config)
            if expansion_model:
                synthesis_kwargs["model"] = expansion_model
        if (
            synthesis_kwargs.get("timeout") is None
            and synthesis_kwargs.get("expansion_timeout") is None
        ):
            synthesis_kwargs["expansion_timeout"] = _lcm_expansion_timeout_ms(self.config) / 1000
        return _synthesize_expand_query_payload(retrieval, agent=synthesis_agent, **synthesis_kwargs)

    def _auxiliary_routes(self, summary_request=None, **kwargs):
        routes = (
            kwargs.get("routes")
            or kwargs.get("auxiliary_routes")
            or (summary_request or {}).get("routes")
        )
        if isinstance(routes, dict):
            routes = [routes]
        defaults = {}
        for key in ("model", "temperature", "max_tokens", "timeout"):
            if kwargs.get(key) is not None:
                defaults[key] = kwargs[key]
        if "timeout" not in defaults:
            timeout_ms = _lcm_summary_timeout_ms(self.config, hermes_home=self.hermes_home)
            if timeout_ms:
                defaults["timeout"] = timeout_ms / 1000
        if not routes:
            if defaults.get("model") is not None:
                routes = [{}]
            else:
                # Mirror hermes-lcm escalation._summary_model_chain: the
                # configured summary_model plus summary_fallback_models form
                # the default route chain, falling back to one task-default
                # route when nothing is configured.
                primary = str(
                    _lcm_str_setting(self.config, "LCM_SUMMARY_MODEL", "summary_model", default="") or ""
                )
                fallbacks = _lcm_list_setting(
                    self.config,
                    "LCM_SUMMARY_FALLBACK_MODELS",
                    "summary_fallback_models",
                    default=[],
                )
                chain = []
                for model in [primary, *(fallbacks or [])]:
                    normalized_model = str(model or "").strip()
                    if normalized_model not in chain:
                        chain.append(normalized_model)
                if not chain:
                    chain.append("")
                routes = [{"model": model} if model else {} for model in chain]
        normalized = []
        for route in routes:
            if not isinstance(route, dict):
                route = {"model": str(route)}
            normalized.append({**defaults, **route})
        return normalized

    def _call_auxiliary_summary(self, prompt, messages, **kwargs):
        client = _resolve_auxiliary_client(getattr(self, "agent", None))
        summary_request = kwargs.get("summary_request")
        allow_retry_signal = bool(kwargs.pop("allow_retry_signal", False))
        accepts_result = kwargs.pop("accepts_result", None)
        route_kwargs = dict(kwargs)
        route_kwargs.pop("summary_request", None)
        routes = self._auxiliary_routes(summary_request, **route_kwargs)
        breaker_threshold, breaker_cooldown_seconds = _summary_circuit_breaker_settings(self.config)
        last_error = None
        last_classification = None
        rejected_text = None
        rejected_route = None
        rejected_model = None
        if client is None or not callable(getattr(client, "call_llm", None)):
            last_error = RuntimeError("Hermes auxiliary_client.call_llm is unavailable")
        else:
            now = time.time()
            for route in routes:
                route_name = route.get("route") or route.get("name") or route.get("model") or "default"
                route_key = str(route_name)
                if self._cooldown_until.get(route_key, 0) > now:
                    continue
                call_kwargs = {
                    "task": "compression",
                    # Hermes escalation sends the full prompt (guidance plus
                    # serialized CONTENT block) as a single user message.
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": route.get("temperature", 0.3),
                    "max_tokens": route.get("max_tokens", 2048),
                    "timeout": route.get("timeout", 60),
                }
                _apply_lcm_model_route(call_kwargs, route.get("model"))
                try:
                    response = client.call_llm(**call_kwargs)
                    text = _strip_reasoning(_llm_response_text(response))
                    if not text:
                        raise RuntimeError("Hermes auxiliary summary was empty")
                    if accepts_result is not None and not accepts_result(text):
                        rejected_text = text
                        rejected_route = route_key
                        rejected_model = route.get("model")
                        failures = self._route_failures.get(route_key, 0) + 1
                        self._route_failures[route_key] = failures
                        if failures >= breaker_threshold:
                            self._cooldown_until[route_key] = (
                                time.time() + breaker_cooldown_seconds
                            )
                        continue
                    self._route_failures.pop(route_key, None)
                    self._cooldown_until.pop(route_key, None)
                    return {
                        "status": "ok",
                        "text": text,
                        "route": route_key,
                        "model": route.get("model"),
                    }
                except Exception as exc:
                    last_error = exc
                    last_classification = _auxiliary_error_classification(exc)
                    failures = self._route_failures.get(route_key, 0) + 1
                    self._route_failures[route_key] = failures
                    if failures >= breaker_threshold:
                        self._cooldown_until[route_key] = (
                            time.time() + breaker_cooldown_seconds
                        )
        if rejected_text is not None:
            return {
                "status": "rejected",
                "text": rejected_text,
                "route": rejected_route,
                "model": rejected_model,
                "error": "Hermes auxiliary summary was not smaller than source",
                "error_classification": "non_compressing_summary",
            }
        if last_error is not None:
            last_classification = last_classification or _auxiliary_error_classification(last_error)
            if allow_retry_signal and last_classification == "retry_worthy":
                return {
                    "status": "retry",
                    "error": str(last_error),
                    "error_classification": last_classification,
                }
            if allow_retry_signal and last_classification == "permanent":
                return {
                    "status": "error",
                    "error": str(last_error),
                    "error_classification": last_classification,
                }
        fallback = {
            "status": "fallback",
            "text": _deterministic_truncation(messages),
            "route": "deterministic_fallback",
            "model": None,
        }
        if last_error is not None:
            fallback["error"] = str(last_error)
            fallback["error_classification"] = last_classification
        return fallback

    def _summarize_with_escalation(self, source_messages, focus_topic="", **kwargs):
        # Port of hermes-lcm escalation.summarize_with_escalation: L1 detailed
        # summary, L2 aggressive bullets at reduced budget, then deterministic
        # L3 truncation. Each LLM rung accepts a result only when its token
        # estimate is below the source token estimate.
        serialized = _serialize_summary_messages(source_messages)
        source_tokens = _count_messages_tokens(source_messages)
        # Mirrors the leaf budget in hermes-lcm engine._summarize_leaf_chunk_with_rescue.
        token_budget = min(12000, max(2000, int(source_tokens * 0.20)))
        custom_instructions = str(
            _lcm_str_setting(self.config, "LCM_CUSTOM_INSTRUCTIONS", "custom_instructions", default="") or ""
        )
        l2_budget_ratio = _lcm_float_setting(
            self.config,
            "LCM_L2_BUDGET_RATIO",
            "l2_budget_ratio",
            default=0.50,
        )
        if l2_budget_ratio is None:
            l2_budget_ratio = 0.50
        l3_truncate_tokens = (
            _lcm_int_setting(self.config, "LCM_L3_TRUNCATE_TOKENS", "l3_truncate_tokens", default=512) or 512
        )

        def accepts_result(text):
            return source_tokens <= 0 or _count_tokens(text) < source_tokens

        l1_kwargs = dict(kwargs)
        l1_kwargs["accepts_result"] = accepts_result
        l1_kwargs["max_tokens"] = token_budget * 2
        l1_prompt = _build_l1_prompt(
            serialized,
            token_budget,
            0,
            focus_topic=focus_topic,
            custom_instructions=custom_instructions,
        )
        rung_failures = []

        def record_rung_failure(level, summary):
            rung_failures.append({
                "level": level,
                "status": summary.get("status"),
                "route": summary.get("route"),
                "model": summary.get("model"),
                "error": summary.get("error"),
                "error_classification": summary.get("error_classification"),
            })

        summary = self._call_auxiliary_summary(l1_prompt, source_messages, **l1_kwargs)
        if summary.get("status") == "ok":
            return summary
        record_rung_failure(1, summary)

        l2_budget = max(1, int(token_budget * l2_budget_ratio))
        l2_kwargs = dict(kwargs)
        l2_kwargs["accepts_result"] = accepts_result
        l2_kwargs["max_tokens"] = l2_budget * 2
        l2_prompt = _build_l2_prompt(
            serialized,
            l2_budget,
            focus_topic=focus_topic,
            custom_instructions=custom_instructions,
        )
        summary = self._call_auxiliary_summary(l2_prompt, source_messages, **l2_kwargs)
        if summary.get("status") == "ok":
            if rung_failures:
                summary["rung_failures"] = rung_failures
            return summary
        record_rung_failure(2, summary)
        if summary.get("status") == "fallback":
            summary.setdefault("error", "Hermes auxiliary summary was not smaller than source")
            summary.setdefault("error_classification", "non_compressing_summary")
        fallback = {
            "status": "fallback",
            "text": _deterministic_truncation(source_messages, limit=max(1, l3_truncate_tokens * 4)),
            "route": "deterministic_fallback",
            "model": None,
            "error": summary.get("error") or "Hermes auxiliary summary was not smaller than source",
            "error_classification": summary.get("error_classification") or "non_compressing_summary",
            "rung_failures": rung_failures,
        }
        if summary.get("error"):
            fallback["auxiliary_error"] = summary.get("error")
        if summary.get("error_classification"):
            fallback["auxiliary_error_classification"] = summary.get("error_classification")
        return fallback

    def _run_pre_compaction_extraction(self, summary_request, source_messages):
        if not _lcm_extraction_enabled(self.config):
            return None
        if not source_messages:
            return {"status": "no_source"}
        extraction_request = (
            summary_request.get("extraction_request") if isinstance(summary_request, dict) else None
        )
        prompt = extraction_request.get("prompt") if isinstance(extraction_request, dict) else None
        if not isinstance(prompt, str) or not prompt.strip():
            return {
                "status": "failed_non_blocking",
                "error": "LCM extraction envelope missing prompt",
                "model": None,
                "output_path": None,
            }
        extraction_model = str(
            _lcm_extraction_model(self.config)
            or _lcm_str_setting(self.config, "LCM_SUMMARY_MODEL", "summary_model", default="")
            or ""
        ).strip()
        timeout_seconds = _lcm_summary_timeout_ms(self.config, hermes_home=self.hermes_home) / 1000
        # Intentional divergence from upstream hermes-lcm: Rust stores extraction results in
        # summary-node metadata instead of writing daily markdown files. We still surface
        # output_path in the extraction contract for config/API parity.
        output_path = _lcm_extraction_output_path(self.config)
        client = _resolve_auxiliary_client(getattr(self, "agent", None))
        if client is None or not callable(getattr(client, "call_llm", None)):
            return {
                "status": "failed_non_blocking",
                "error": "Hermes auxiliary_client.call_llm is unavailable",
                "model": extraction_model or None,
                "output_path": output_path or None,
            }
        call_kwargs = {
            "task": _EXTRACTION_TASK["key"],
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.2,
            "max_tokens": 2000,
        }
        _apply_lcm_model_route(call_kwargs, extraction_model)
        if timeout_seconds is not None:
            call_kwargs["timeout"] = timeout_seconds
        try:
            response = client.call_llm(**call_kwargs)
        except Exception as exc:
            return {
                "status": "failed_non_blocking",
                "error": str(exc),
                "model": extraction_model or None,
                "output_path": output_path or None,
            }
        cleaned = _strip_reasoning(_llm_response_text(response)).strip()
        if not cleaned or cleaned == "NOTHING_TO_EXTRACT":
            return {
                "status": "nothing_to_extract",
                "model": extraction_model or None,
                "output_path": output_path or None,
            }
        return {
            "status": "ok",
            "items": _normalize_extraction_items(cleaned),
            "text": cleaned,
            "model": extraction_model or None,
            "output_path": output_path or None,
        }

    def compress(self, messages, current_tokens=None, focus_topic=None, **kwargs):
        """Compact ``messages`` and return the new message LIST.

        The hermes ContextEngine ABC contract is a message list: the host
        appends to the returned value, re-estimates tokens on it, and adopts
        it as the live transcript (agent/conversation_compression.py). The
        raw tracedecay result dict is kept on ``self.last_compress_result``
        for diagnostics/tests. On error or no-op the input list is returned
        unchanged so the host takes its abort/no-op path instead of
        corrupting the conversation.
        """
        original = list(messages or [])
        self._last_compress_aborted = False
        self._last_summary_error = None
        try:
            result = self._compress_to_result(
                original,
                current_tokens=current_tokens,
                focus_topic=focus_topic,
                **kwargs,
            )
        except Exception as exc:
            logger.warning("tracedecay compression failed: %s", exc)
            self._last_compress_aborted = True
            self._last_summary_error = str(exc)
            return original
        self.last_compress_result = result if isinstance(result, dict) else {}
        if (
            not isinstance(result, dict)
            or result.get("error")
            or result.get("status") == "error"
        ):
            self._last_compress_aborted = True
            if isinstance(result, dict):
                self._last_summary_error = str(
                    result.get("error") or result.get("reason") or "compression error"
                )
            else:
                self._last_summary_error = "invalid compression result"
            return original
        if _compression_replay_is_compacted(result):
            self._last_compress_aborted = True
            self._last_summary_error = str(
                result.get("mcp_truncation_reason")
                or result.get("reason")
                or "compression replay was compacted for MCP"
            )
            return original
        replay = _replay_message_list(result.get("replay_messages"))
        if replay is None or (not replay and original):
            # No usable replay window — treat as a no-op; the host detects
            # ``compressed == messages`` and skips session rotation.
            self._last_compress_aborted = True
            self._last_summary_error = str(result.get("reason") or "no usable replay")
            return original
        if replay == original:
            return original
        if replay != original:
            self.compression_count += 1
        return replay

    def _compress_to_result(self, messages, current_tokens=None, focus_topic=None, **kwargs):
        summarizer = kwargs.pop("summarizer", None) or {"mode": "hermes_auxiliary"}
        force = bool(kwargs.pop("force", False))
        max_auxiliary_attempts = _auxiliary_retry_limit(kwargs)
        lcm_option_keys = (
            "expected_current_frontier_store_id",
            "threshold_tokens",
            "max_assembly_tokens",
            "leaf_chunk_tokens",
            "max_source_messages",
            "summary_fan_in",
            "incremental_max_depth",
            "fresh_tail_count",
            "dynamic_leaf_chunk_enabled",
            "dynamic_leaf_chunk_max",
            "context_length",
            "reserve_tokens_floor",
            "ignore_session_patterns",
            "stateless_session_patterns",
            "ignore_message_patterns",
        )
        args = self._tool_args()
        args.update(
            _lcm_config_args(
                self.config,
                self.hermes_home,
                runtime_context_length=self._effective_context_length(),
            )
        )
        args.update({
            "provider": STANDARD_HERMES_LCM_PROVIDER,
            "messages": messages,
            "current_tokens": current_tokens,
            "focus_topic": focus_topic,
            "summarizer": summarizer,
        })
        if self.project_root:
            args["response_handle_project_root"] = self.project_root
        _apply_lcm_option_overrides(args, kwargs, lcm_option_keys)
        if force and not args.get("max_assembly_tokens"):
            try:
                current = int(current_tokens or 0)
            except (TypeError, ValueError):
                current = 0
            if current > 1:
                args["max_assembly_tokens"] = current - 1
            elif args.get("threshold_tokens"):
                try:
                    threshold = int(args["threshold_tokens"])
                    if threshold > 1:
                        args["max_assembly_tokens"] = threshold - 1
                except (TypeError, ValueError):
                    pass

        attempts = 0
        retry_status = None
        error_classification = None
        fallback_used = False
        attempt_args = dict(args)

        while attempts < max_auxiliary_attempts:
            first = call_tracedecay_json("tracedecay_lcm_compress", attempt_args, **kwargs)
            if _compression_replay_is_compacted(first):
                return _with_auxiliary_metadata(
                    _compacted_lcm_result_error(first),
                    attempts=attempts,
                    retry_status=retry_status,
                    error_classification=error_classification,
                    fallback_used=fallback_used,
                )
            if first.get("status") != "needs_summary":
                return _with_auxiliary_metadata(
                    first,
                    attempts=attempts,
                    retry_status=retry_status,
                    error_classification=error_classification,
                    fallback_used=fallback_used,
                )

            summary_request = first.get("summary_request") or {}
            source_messages = _summary_source_messages(
                summary_request.get("source_messages") or messages
            )
            attempts += 1
            extraction_result = self._run_pre_compaction_extraction(
                summary_request,
                source_messages,
            )
            summary = self._summarize_with_escalation(
                source_messages,
                focus_topic=summary_request.get("focus_topic") or focus_topic or "",
                summary_request=summary_request,
                allow_retry_signal=True,
                **kwargs,
            )
            summary_status = summary.get("status")
            if summary_status in ("retry", "error"):
                error_classification = summary.get("error_classification") or (
                    "retry_worthy" if summary_status == "retry" else "permanent"
                )
                smaller_limit = _next_smaller_source_limit(
                    source_messages,
                    attempt_args.get("max_source_messages"),
                )
                if (
                    summary_status == "retry"
                    and smaller_limit is not None
                    and attempts < max_auxiliary_attempts
                ):
                    retry_status = "retried"
                    attempt_args = dict(args)
                    attempt_args["max_source_messages"] = smaller_limit
                    continue
                retry_status = "retry_exhausted" if summary_status == "retry" else "not_retryable"
                return _auxiliary_error_result(
                    first,
                    attempts=attempts,
                    retry_status=retry_status,
                    error_classification=error_classification,
                    error=summary.get("error"),
                )

            if summary_status == "fallback":
                fallback_used = True
                retry_status = retry_status or "fallback_summary"
                error_classification = summary.get("error_classification") or error_classification

            provided_args = dict(attempt_args)
            provided_route = _extraction_route_payload(summary.get("route"), extraction_result)
            provided_args["summarizer"] = {
                "mode": "provided",
                "summary_text": summary["text"],
                "route": provided_route,
            }
            result = call_tracedecay_json("tracedecay_lcm_compress", provided_args, **kwargs)
            if _compression_replay_is_compacted(result):
                result = _compacted_lcm_result_error(result)
            return _with_auxiliary_metadata(
                result,
                attempts=attempts,
                retry_status=retry_status,
                error_classification=error_classification,
                fallback_used=fallback_used,
            )

class TracedecayMemoryProvider(MemoryProvider):
    provider_id = "tracedecay"

    def __init__(self):
        self.hermes_home = None
        self.project_root = None
        self.session_id = None
        self.agent_context = ""
        self._prefetch_lock = threading.Lock()
        self._prefetch_cache = {}
        self._sync_turn_sequence = 0

    @property
    def name(self) -> str:
        return "tracedecay"

    def is_available(self) -> bool:
        return _tracedecay_binary_available()

    def initialize(self, session_id=None, **kwargs):
        self.hermes_home = kwargs.get("hermes_home") or _resolve_hermes_home()
        config = _with_plugin_block(kwargs.get("config"), self.hermes_home)
        self.project_root = _code_project_root(
            explicit=kwargs.get("project_root"),
            cwd=kwargs.get("cwd"),
            configured=_configured_project_root(config),
        )
        self.session_id = session_id
        # Execution context ("", "cron", "flush", ...): cron/flush runs are
        # not primary conversations and must not write turn state (cron
        # already passes skip_memory=True by design; this is the belt for
        # hosts that initialize the provider anyway).
        self.agent_context = str(kwargs.get("agent_context") or "")

    def post_setup(self, hermes_home, config):
        """`hermes memory setup` hand-off: verify the binary and warm the store.

        tracedecay_memory_status repairs derived holographic vectors/banks
        and creates the resolved user-level store on first touch, so one call doubles
        as install repair + initialization.
        """
        if not _tracedecay_binary_available():
            print(
                f"  tracedecay binary not found at {tools.TRACEDECAY_BIN} — "
                "install it (cargo install tracedecay) and re-run `hermes memory setup`."
            )
            return
        home = str(hermes_home or self.hermes_home or _resolve_hermes_home())
        resolved_config = _with_plugin_block(config, home)
        project_root = _code_project_root(configured=_configured_project_root(resolved_config))
        storage = _storage_args(project_root, home)
        status = call_tracedecay_json("tracedecay_memory_status", storage, **storage)
        if isinstance(status, dict) and not status.get("error"):
            facts = status.get("fact_count", status.get("facts"))
            suffix = f" ({facts} facts)" if facts is not None else ""
            label = project_root or home
            print(f"  tracedecay memory store ready at {label}{suffix}.")
        else:
            detail = status.get("error") if isinstance(status, dict) else status
            print(f"  tracedecay memory store check failed: {detail}")

    def system_prompt_block(self):
        # Built once per session by the host during system prompt assembly —
        # cache-stable, unlike per-turn pre_llm_call injection.
        return (
            "tracedecay memory is active: durable facts live in the holographic "
            "fact store. Use fact_store(action='add') for facts worth keeping "
            "across sessions and fact_store(action='search') for explicit "
            "recall; relevant facts are also prefetched automatically."
        )

    def _prefetch_key(self, session_id=""):
        return str(session_id or self.session_id or "")

    def queue_prefetch(self, query, *, session_id=""):
        """Queue background fact recall for the NEXT turn.

        The host calls this after each turn completes; a daemon thread runs
        the subprocess recall and parks the result for prefetch() to consume
        (same shape as the honcho/mem0 providers).
        """
        if not _plugin_toggle("prefetch", True):
            return
        text = str(query or "").strip()
        if not text:
            return
        key = self._prefetch_key(session_id)

        def _worker():
            try:
                recall = self._recall_facts(text)
            except Exception as exc:
                logger.debug("tracedecay queue_prefetch failed: %s", exc)
                return
            with self._prefetch_lock:
                self._prefetch_cache[key] = recall

        threading.Thread(
            target=_worker, name="tracedecay-memory-prefetch", daemon=True
        ).start()

    def prefetch(self, query, *, session_id=""):
        """Return the recall queued at the end of the previous turn.

        Called inline before every API call, so it must never block on a
        subprocess (memory_provider ABC: be fast here, recall in
        queue_prefetch). Empty string means nothing relevant yet.
        """
        del query
        with self._prefetch_lock:
            return self._prefetch_cache.pop(self._prefetch_key(session_id), "") or ""

    def _recall_facts(self, query):
        """Subprocess fact recall, formatted for cache-safe injection.

        Kept small on purpose: the recall is injected context, and the MCP
        envelope truncates tool payloads past 15k chars, so a low limit
        keeps giant facts from corrupting the response JSON.
        """
        text = str(query or "").strip()
        if not text:
            return ""
        try:
            args = _storage_args(self.project_root, self.hermes_home)
            args.update({"action": "search", "query": text[:512], "limit": 3})
            payload = call_tracedecay_json("tracedecay_fact_store", args, **args)
        except Exception as exc:
            logger.debug("tracedecay memory prefetch failed: %s", exc)
            return ""
        if not isinstance(payload, dict) or payload.get("error"):
            return ""
        facts = payload.get("facts") or payload.get("results") or []
        lines = []
        for item in facts:
            if not isinstance(item, dict):
                continue
            # Search results nest the row under "fact" (with match scores
            # beside it); list results are flat fact rows.
            fact = item.get("fact") if isinstance(item.get("fact"), dict) else item
            content = str(fact.get("content") or "").strip()
            if not content:
                continue
            if len(content) > 600:
                content = content[:600].rstrip() + "..."
            fact_id = fact.get("fact_id")
            prefix = f"[fact {fact_id}] " if fact_id is not None else ""
            lines.append(f"- {prefix}{content}")
        return "\n".join(lines)

    def sync_turn(self, user_content, assistant_content, *, session_id="", messages=None):
        """Persist the completed turn into the LCM raw store.

        Uses tracedecay_lcm_preflight's lossless active-message ingest (the
        same content-cursored path the context engine uses), so the raw
        store grows every turn instead of only when compression fires.
        """
        if not _plugin_toggle("sync_turn", True):
            return
        if self.agent_context in ("cron", "flush"):
            # Non-primary execution contexts must not write turn state.
            return
        sid = session_id or self.session_id
        if not sid:
            return
        forwarded_messages = isinstance(messages, list) and bool(messages)
        turn_messages = messages if forwarded_messages else None
        if not turn_messages:
            turn_messages = []
            if user_content:
                turn_messages.append({"role": "user", "content": str(user_content)})
            if assistant_content:
                turn_messages.append({"role": "assistant", "content": str(assistant_content)})
        if not turn_messages:
            return
        if not forwarded_messages:
            self._sync_turn_sequence += 1
            batch_id = self._sync_turn_sequence
            timestamp_ns = time.time_ns()
            for idx, entry in enumerate(turn_messages):
                role = str(entry.get("role") or "user")
                entry["id"] = f"tracedecay_sync_{batch_id}_{timestamp_ns}_{idx}_{role}"
        args = _storage_args(self.project_root, self.hermes_home)
        args.update({
            "provider": STANDARD_HERMES_LCM_PROVIDER,
            "session_id": sid,
            "messages": turn_messages,
        })
        try:
            tools.call_tracedecay_tool(
                "tracedecay_lcm_preflight",
                args,
                **args,
            )
        except Exception as exc:
            logger.debug("tracedecay sync_turn ingest failed: %s", exc)

    def on_memory_write(self, action, target, content, metadata=None):
        """Mirror built-in memory tool writes into the fact store."""
        if action not in ("add", "replace"):
            # Removals carry no stable fact identity to mirror.
            return
        text = str(content or "").strip()
        if not text:
            return
        fact_metadata = {"hermes_action": str(action), "hermes_target": str(target)}
        for key in ("session_id", "platform", "write_origin", "tool_name"):
            value = (metadata or {}).get(key)
            if value:
                fact_metadata[key] = str(value)
        fact_args = {
            "action": "add",
            "content": text,
            "category": "user_pref" if target == "user" else "general",
            "metadata": fact_metadata,
        }
        try:
            args = _storage_args(self.project_root, self.hermes_home)
            args.update(fact_args)
            tools.call_tracedecay_tool("tracedecay_fact_store", args, **args)
        except Exception as exc:
            logger.debug("tracedecay on_memory_write mirror failed: %s", exc)

    def get_config_schema(self):
        # Consumed by `hermes memory setup` / `hermes memory status`.
        return [
            {
                "key": "project_root",
                "description": "Absolute path of the project tracedecay serves (install-time pin)",
                "default": tools.config_pinned_project_root() or "",
            },
        ]

    def get_config_defaults(self):
        # Hermes layers these under DEFAULT_CONFIG so plugins.tracedecay
        # exists in loaded configs without core changes.
        return {"plugins": {"tracedecay": _plugin_config_defaults()}}

    def get_config_field_meta(self):
        # Dashboard config-page hints for every provider-owned dot-path.
        return {
            f"plugins.tracedecay.{key}": {
                "category": "plugins",
                "description": description,
            }
            for key, (_default, description) in PLUGIN_CONFIG_FIELDS.items()
        }

    def save_config(self, values, hermes_home):
        # `hermes memory setup` hands collected non-secret values here; the
        # conventional home is the plugins.tracedecay block. Prefer hermes'
        # canonical raw-read + save path (config lock, atomic write,
        # managed-config guard) and only fall back to raw YAML outside a
        # hermes install.
        updates = {key: value for key, value in (values or {}).items()}
        if _hermes_cli_config is not None and callable(
            getattr(_hermes_cli_config, "save_config", None)
        ):
            try:
                reader = getattr(_hermes_cli_config, "read_raw_config", None)
                existing = reader() if callable(reader) else None
                if not isinstance(existing, dict):
                    existing = {}
                plugins_cfg = existing.get("plugins")
                if not isinstance(plugins_cfg, dict):
                    plugins_cfg = {}
                block = plugins_cfg.get("tracedecay")
                if not isinstance(block, dict):
                    block = {}
                block.update(updates)
                plugins_cfg["tracedecay"] = block
                existing["plugins"] = plugins_cfg
                _hermes_cli_config.save_config(existing)
                return
            except Exception as exc:
                logger.warning(
                    "tracedecay save_config via hermes_cli.config failed; falling back to raw YAML: %s",
                    exc,
                )
        config_path = Path(hermes_home) / "config.yaml"
        try:
            import yaml
            existing = {}
            if config_path.exists():
                with open(config_path, encoding="utf-8-sig") as config_file:
                    existing = yaml.safe_load(config_file) or {}
            plugins_cfg = existing.get("plugins")
            if not isinstance(plugins_cfg, dict):
                plugins_cfg = {}
            block = plugins_cfg.get("tracedecay")
            if not isinstance(block, dict):
                block = {}
            block.update(updates)
            plugins_cfg["tracedecay"] = block
            existing["plugins"] = plugins_cfg
            with open(config_path, "w", encoding="utf-8") as config_file:
                yaml.dump(existing, config_file, default_flow_style=False)
        except Exception as exc:
            logger.warning("tracedecay memory provider save_config failed: %s", exc)

    def get_tool_schemas(self):
        # Collapsed surface (12 -> 3): fact_store(action=...) covers the nine
        # fixed-action fact_* aliases, which remain dispatchable through
        # handle_tool_call for compatibility with older transcripts/configs
        # but no longer cost per-call schema footprint.
        return [
            _memory_schema("tracedecay_fact_store", "fact_store"),
            _memory_schema("tracedecay_fact_feedback", "fact_feedback"),
            _memory_schema("tracedecay_memory_status", "memory_status"),
        ]

    def handle_tool_call(self, name, arguments=None, **kwargs) -> str:
        tool_name, tool_args = _normalize_memory_tool_call(name, arguments)
        mapping = MEMORY_TOOL_MAP.get(tool_name)
        if mapping is None:
            return tools.error_payload(f"unknown memory tool: {tool_name}")
        tracedecay_name = mapping["tracedecay_name"]
        fixed_args = mapping.get("fixed_args")
        if fixed_args:
            tool_args = dict(tool_args)
            tool_args.update(fixed_args)
        return tools.call_tracedecay_tool(tracedecay_name, tool_args, **kwargs)

def register(ctx):
    global _HOST_FORWARDS_MESSAGES
    ctx.register_hook("pre_llm_call", _pre_llm_call)
    # Declare the plugins.tracedecay config block so its keys exist in
    # load_config() even before the user edits config.yaml.
    register_config_defaults = getattr(ctx, "register_config_defaults", None)
    if callable(register_config_defaults):
        try:
            register_config_defaults({"plugins": {"tracedecay": _plugin_config_defaults()}})
        except Exception as exc:
            logger.warning("tracedecay config defaults registration failed: %s", exc)
    # Declare the extraction side-LLM as a proper auxiliary task so users can
    # pin its provider/model under auxiliary.lcm_extraction instead of the
    # generic auto-resolved "extraction" defaults.
    register_auxiliary_task = getattr(ctx, "register_auxiliary_task", None)
    if callable(register_auxiliary_task):
        try:
            register_auxiliary_task(
                "lcm_extraction",
                display_name="LCM extraction",
                description="tracedecay pre-compaction decision/insight extraction",
                defaults={"provider": "auto", "model": "", "timeout": 60},
            )
        except Exception as exc:
            logger.debug("tracedecay auxiliary task registration failed: %s", exc)
        else:
            _EXTRACTION_TASK["key"] = "lcm_extraction"
    register_command = getattr(ctx, "register_command", None)
    if callable(register_command):
        register_command(
            "/tracedecay_status",
            _tracedecay_status,
            description="Show tracedecay project status.",
        )

    if callable(getattr(ctx, "register_memory_provider", None)):
        ctx.register_memory_provider(TracedecayMemoryProvider())

    context_config = getattr(ctx, "config", None)
    context_hermes_home = (
        getattr(ctx, "hermes_home", None)
        or getattr(ctx, "_hermes_home", None)
    )
    context_engine = TraceDecayContextEngine(
        config=context_config,
        hermes_home=context_hermes_home,
    )
    if callable(getattr(ctx, "register_context_engine", None)):
        ctx.register_context_engine(context_engine)

    # Direct tool registration is split by capability:
    #   - Code-graph / memory / transcript tools register UNCONDITIONALLY.
    #     They work without host message forwarding, and hermes defers
    #     registered non-core tools through its tool-search bridge
    #     (tools/tool_search.py), so schema footprint is not a blocker.
    #   - Only the live-ingest LCM verbs whose schemas take the in-memory
    #     ``messages`` list (MESSAGE_DEPENDENT_TOOLS) and the context-engine
    #     native tool mirrors stay gated behind the message-forwarding
    #     capability flag — without forwarding their ingest piggyback can
    #     never fire (the host still mounts the native LCM tools itself via
    #     context_engine.get_tool_schemas()).
    register_tool = getattr(ctx, "register_tool", None)
    host_forwards_messages = _host_forwards_registered_tool_messages(ctx)
    _HOST_FORWARDS_MESSAGES = host_forwards_messages if callable(register_tool) else None
    tracedecay_is_memory_provider = _active_memory_provider(ctx) == "tracedecay"
    if callable(register_tool):
        for schema in schemas.TOOL_SCHEMAS:
            name = schema["name"]
            if name in MESSAGE_DEPENDENT_TOOLS and not host_forwards_messages:
                continue
            if name in MEMORY_PROVIDER_TOOLS and tracedecay_is_memory_provider:
                # The active memory provider already exposes this store as
                # fact_store/fact_feedback/memory_status — registering the
                # prefixed twins would double the schema footprint.
                continue
            handler = _handle_lcm_expand_query if name == "tracedecay_lcm_expand_query" else tools.make_handler(name)
            try:
                register_tool(
                    name=name,
                    toolset="tracedecay",
                    schema=schema,
                    handler=handler,
                )
            except Exception as exc:
                logger.warning(
                    "tracedecay tool registration failed for %s; continuing: %s",
                    name,
                    exc,
                )
            else:
                _REGISTERED_TOOL_NAMES.add(name)
        if host_forwards_messages:
            for schema in context_engine.get_tool_schemas():
                name = schema["name"]
                try:
                    register_tool(
                        name=name,
                        toolset="context_engine",
                        schema=schema,
                        handler=_make_wrapped_lcm_handler(name, context_engine),
                        description=schema.get("description", ""),
                    )
                except Exception as exc:
                    logger.warning(
                        "tracedecay LCM tool registration failed for %s; continuing with context-engine schemas: %s",
                        name,
                        exc,
                    )
                else:
                    _CONTEXT_TOOL_NAMES.add(name)
        else:
            logger.info(
                "tracedecay LCM registered live-ingest tools skipped: this Hermes host does not forward messages to registered tool handlers; context-engine compression still receives host messages"
            )
    else:
        logger.info(
            "tracedecay direct tool registration unavailable on this Hermes host; continuing with context-engine schemas"
        )

    skills_dir = Path(__file__).parent / "skills"
    skill_path = skills_dir / "tracedecay" / "SKILL.md"
    register_skill = getattr(ctx, "register_skill", None)
    if skill_path.exists() and callable(register_skill):
        # Newer Hermes derives the namespace from the plugin name and
        # rejects ':' in skill names, so register the bare name.
        try:
            register_skill("tracedecay", skill_path)
        except Exception as exc:
            logger.warning("tracedecay skill registration failed: %s", exc)