"""MemOS Local — Hermes memory provider (Reflect2Evolve V7 core).

Implements the ``agent.memory_provider.MemoryProvider`` interface exposed
by the hermes-agent host (see
``hermes-agent/agent/memory_provider.py``). All heavy lifting lives in the
Node.js ``memos-local-plugin`` core; this adapter is a thin Python client
that speaks JSON-RPC 2.0 over stdio to the packaged Node bridge.

Discovery
---------
The hermes-agent host discovers memory providers via
``plugins/memory/__init__.py::load_memory_provider`` which:

  1. Looks for a ``register(ctx)`` function and calls it with a
     ``_ProviderCollector`` that has ``register_memory_provider(provider)``.
  2. Falls back to finding a ``MemoryProvider`` subclass in the module.

We support **both** entry points.

Activation
----------
Set ``memory.provider: memtensor`` in ``~/.hermes/config.yaml`` (or the
relevant `$HERMES_HOME`).

Lifecycle mapping (V7 §0.2)
---------------------------

| Hermes hook          | Our action                                    |
| -------------------- | --------------------------------------------- |
| ``initialize``       | acquire shared bridge; open logical session   |
| ``on_turn_start``    | record turn count; stash message              |
| ``prefetch``         | ``turn.start`` RPC → Tier 1+2+3 retrieval     |
| ``queue_prefetch``   | no-op; real prefetch runs before the turn     |
| ``sync_turn``        | persist a synchronous ``turn.end`` RPC        |
| ``on_session_end``   | close this logical session                    |
| ``on_pre_compress``  | extract a short memory summary               |
| ``on_delegation``    | record a subagent outcome as a trace         |
| ``get_tool_schemas`` | expose memory, skill, and environment tools   |
| ``handle_tool_call`` | dispatch to MemOS JSON-RPC tool methods       |
| ``shutdown``         | release provider lease                        |

Threading: all JSON-RPC calls are synchronous. One process-scoped runtime
owns bridge keepalive and reconnect; providers retain per-session state.
"""

from __future__ import annotations

import contextlib
import json
import logging
import os
import re
import sys
import threading
import time
import weakref

from pathlib import Path
from typing import Any


# Add our own directory to sys.path so the submodule imports below work
# whether hermes-agent loaded us bundled or via the user-plugin namespace.
_PLUGIN_DIR = Path(__file__).resolve().parent
if str(_PLUGIN_DIR) not in sys.path:
    sys.path.insert(0, str(_PLUGIN_DIR))

from bridge_client import BridgeError, MemosBridgeClient  # noqa: E402
from daemon_manager import (  # noqa: E402
    ensure_bridge_running,
    ensure_viewer_daemon,
    kill_zombie_bridges,
)
from runtime_home import resolve_runtime_home  # noqa: E402
from shared_bridge_runtime import (  # noqa: E402
    HERMES_HOOK_DISPATCHER,
    SHARED_BRIDGE_REGISTRY,
    SharedBridgeLease,
)


try:  # pragma: no cover — host-provided base class, absent in unit tests
    from agent.memory_provider import MemoryProvider  # type: ignore
except Exception:  # pragma: no cover

    class MemoryProvider:  # type: ignore[no-redef]
        """Fallback base class used when running outside hermes-agent host.

        Defines only the attributes the adapter reads so ``pyright`` and
        ``pytest`` stay happy in standalone test runs.
        """


logger = logging.getLogger(__name__)

PLUGIN_ID = "memos-local-hermes"


def _read_plugin_version() -> str:
    """Read the npm package version that owns this Hermes adapter."""
    package_json = _PLUGIN_DIR.parents[2] / "package.json"
    try:
        payload = json.loads(package_json.read_text(encoding="utf-8"))
    except (OSError, ValueError, TypeError):
        return "dev"
    version = payload.get("version")
    return version.strip() if isinstance(version, str) and version.strip() else "dev"


PLUGIN_VERSION = _read_plugin_version()
_TOOL_FAILURE_REPAIR_HINT = (
    "This tool has failed multiple times in a row. You may want to call "
    "`memos_search` for relevant past experience before deciding what to do next."
)
_TOOL_FAILURE_HINT_THRESHOLD = 3
_COMPRESSION_CONTEXT_MAX_CHARS = 6_000


def _shared_bridge_enabled() -> bool:
    """Use the safe process-scoped bridge unless an operator opts out."""
    mode = os.environ.get("MEMOS_HERMES_BRIDGE_MODE", "shared").strip().lower()
    return mode not in {"legacy", "per_provider", "disabled", "false", "0"}


def _resolved_memos_runtime_home() -> Path:
    """Mirror the Node resolver closely enough to isolate distinct databases."""
    return resolve_runtime_home(plugin_root=_PLUGIN_DIR.parents[2])


def _memos_runtime_env_snapshot(runtime_home: Path | None = None) -> dict[str, str]:
    """Freeze the environment inputs that select a MemOS data home."""
    memos_home = os.environ.get("MEMOS_HOME", "").strip()
    if memos_home:
        resolved_home = runtime_home or Path(memos_home).expanduser().resolve()
        return {"MEMOS_HOME": str(resolved_home)}
    config_file = os.environ.get("MEMOS_CONFIG_FILE", "").strip()
    if config_file:
        return {
            "MEMOS_HOME": "",
            "MEMOS_CONFIG_FILE": str(Path(config_file).expanduser().resolve()),
        }
    resolved_home = runtime_home or _resolved_memos_runtime_home()
    return {"MEMOS_HOME": str(resolved_home)}


def _shared_bridge_runtime_key(runtime_home: Path | None = None) -> tuple[str, ...]:
    """Use the data home as the authoritative shared-runtime boundary."""
    resolved_home = runtime_home or _resolved_memos_runtime_home()
    return (str(resolved_home), "hermes", "stdio")


def _prepare_shared_bridge(
    runtime_home: Path | None = None,
    *,
    cleanup_legacy_zombies: bool = False,
) -> None:
    """Prepare bridge/viewer state without crossing data-home boundaries."""
    ensure_bridge_running()
    if cleanup_legacy_zombies:
        # The legacy scanner cannot distinguish data homes. Shared runtimes
        # instead rely on the Python scoped singleton and the CJS scoped PID
        # guard so one healthy home is never reaped while another starts.
        with contextlib.suppress(Exception):
            zombies = kill_zombie_bridges()
            if zombies:
                logger.info("MemOS: killed %d zombie bridge(s)", zombies)
    try:
        ensure_viewer_daemon(runtime_home=runtime_home)
    except Exception as err:
        logger.warning("MemOS: viewer daemon check failed — %s", err)


def _long_rpc_timeout_default() -> float:
    """Resolve the timeout used for long-running JSON-RPC calls.

    After 1-2 hours of Hermes use the memory / capture / reflection
    pipeline grows past the 30s JSON-RPC default and surfaces as
    ``[timeout] memory.search did not respond within 30.0s`` and
    ``[timeout] turn.end did not respond within 30.0s`` in the host
    logs (issue #2028). ``feedback.submit`` already opts into 75s;
    ``sync_turn``'s ``_ensure_bridge`` also uses 75s. Aligning the
    heavy retrieval / capture RPCs with the same 75s ceiling gives
    the pipeline enough headroom without turning genuinely hung
    calls into an indefinite wait. The value is overridable via
    ``MEMOS_HERMES_LONG_RPC_TIMEOUT`` for site-specific tuning; any
    unparseable / non-positive value falls back to the default.
    """
    raw = os.environ.get("MEMOS_HERMES_LONG_RPC_TIMEOUT", "")
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return 75.0
    if value <= 0:
        return 75.0
    return value


_LONG_RPC_TIMEOUT = _long_rpc_timeout_default()


def _prefetch_rpc_timeout_default() -> float:
    """Resolve the latency budget for Hermes' foreground memory lookup.

    Hermes places its own short deadline around ``prefetch``. Reusing the
    long capture timeout here lets the bridge continue work after the host
    has already moved on. Keep a separate, configurable ceiling below the
    host's default and forward the corresponding absolute deadline to core.
    """
    raw = os.environ.get("MEMOS_HERMES_PREFETCH_RPC_TIMEOUT", "")
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return 6.0
    if not value > 0:
        return 6.0
    # Hermes currently abandons external providers after 8 seconds. Keep at
    # least one second for Python thread scheduling and response assembly even
    # when an operator overrides the default.
    return min(value, 7.0)


_PREFETCH_RPC_TIMEOUT = _prefetch_rpc_timeout_default()
_PREFETCH_RESPONSE_RESERVE_SECONDS = 0.25


def _remaining_rpc_timeout(
    deadline_monotonic: float | None,
    requested_timeout: float | None,
) -> float | None:
    """Bound one blocking bridge step by a shared end-to-end deadline."""
    if deadline_monotonic is None:
        return requested_timeout
    remaining = deadline_monotonic - time.monotonic()
    if remaining <= 0:
        raise BridgeError("timeout", "foreground prefetch deadline exceeded")
    if requested_timeout is None:
        return remaining
    return min(requested_timeout, remaining)


_HERMES_INTERNAL_REVIEW_PREFIXES = (
    "review the conversation above and consider saving to memory if appropriate.",
    "review the conversation above and update the skill library.",
    "review the conversation above and update two things:",
    "review the conversation above and consider saving or updating a skill if appropriate.",
    "review the conversation above and consider whether a skill should be saved or updated.",
)


def _is_hermes_internal_review_prompt(message: str) -> bool:
    """Return True for Hermes' own background memory/skill review turns."""
    normalized = " ".join((message or "").strip().lower().split())
    if not normalized:
        return False
    return any(normalized.startswith(prefix) for prefix in _HERMES_INTERNAL_REVIEW_PREFIXES)


def _is_explicit_delegation_request(message: str) -> bool:
    """Return True when the user explicitly asks Hermes to use a subagent."""
    text = " ".join((message or "").strip().lower().split())
    if not text:
        return False
    delegation_terms = (
        "subagent",
        "sub-agent",
        "sub agent",
        "delegate",
        "delegation",
        "子代理",
        "子任务",
        "派一个",
        "派发",
    )
    return any(term in text for term in delegation_terms)


def _is_verifier_feedback_prompt(message: str) -> bool:
    """Return True for explicit evaluator/verifier feedback turns."""
    text = " ".join((message or "").strip().lower().split())
    if not text:
        return False

    # Strong markers: formal verifier feedback
    strong_markers = (
        "本任务评为反例",
        "本任务评为正例",
        "verifier feedback",
        "verification feedback",
        "task rated as counterexample",
        "task is rated as counterexample",
        "r <= -0.5",
        "r≤-0.5",
        "r >= 0.5",
        "r≥0.5",
    )
    if any(marker in text for marker in strong_markers):
        return True
    if re.search(r"\br\s*(?:<=|>=|≤|≥)\s*-?\d+(?:\.\d+)?", text):
        return True

    # User correction markers: natural corrective feedback
    correction_markers = (
        "不对",
        "错了",
        "不是",
        "不行",
        "不对的",
        "写错了",
        "做错了",
        "理解错了",
        "wrong",
        "incorrect",
        "not right",
        "not correct",
        "that's wrong",
        "this is wrong",
    )
    if any(marker in text for marker in correction_markers):
        return True

    # Weak markers: require "feedback/反馈" + action keywords
    if "feedback" not in text and "反馈" not in text:
        return False
    feedback_markers = (
        "failed",
        "failure",
        "pass",
        "passed",
        "success",
        "succeeded",
        "should",
        "avoid",
        "next time",
        "失败",
        "成功",
        "应该",
        "不要",
        "下次",
    )
    return any(marker in text for marker in feedback_markers)


def _feedback_polarity(message: str) -> str:
    text = " ".join((message or "").strip().lower().split())
    if re.search(r"r\s*(?:<=|≤)\s*-?0\.5", text):
        return "negative"
    if "反例" in text:
        return "negative"
    if any(
        term in text
        for term in (
            "failed",
            "failure",
            "wrong",
            "incorrect",
            "not acceptable",
            "错误",
            "失败",
            "不对",
        )
    ):
        return "negative"
    if re.search(r"r\s*(?:>=|≥)\s*0\.5", text):
        return "positive"
    if "正例" in text:
        return "positive"
    if any(
        term in text
        for term in ("passed", "success", "succeeded", "correct", "great", "成功", "通过", "正确")
    ):
        return "positive"
    return "neutral"


def _feedback_magnitude(message: str, polarity: str) -> float:
    text = " ".join((message or "").strip().lower().split())
    match = re.search(r"\br\s*(?:=|:|<=|>=|≤|≥)\s*(-?\d+(?:\.\d+)?)", text)
    if match:
        with contextlib.suppress(Exception):
            return max(0.0, min(1.0, abs(float(match.group(1)))))
    return 1.0 if polarity in {"positive", "negative"} else 0.6


class MemTensorProvider(MemoryProvider):
    """MemOS Reflect2Evolve memory for hermes-agent.

    Wraps a JSON-RPC client around the shared ``memos-local-plugin`` core.

    Only methods that Hermes actually calls are overridden here; every
    optional hook stays default so future versions of the base class can
    grow without breaking us.
    """

    def __init__(self) -> None:
        self._bridge: MemosBridgeClient | SharedBridgeLease | None = None
        self._shared_bridge = _shared_bridge_enabled()
        self._bridge_generation = 0
        self._reconnect_lock = threading.Lock()
        self._session_open_lock = threading.Lock()
        self._runtime_home: Path | None = None
        self._runtime_env: dict[str, str] = {}
        self._session_id: str = ""
        self._episode_id: str = ""
        self._hermes_home: str = ""
        self._agent_identity: str = "hermes"
        self._platform: str = "cli"
        self._last_host_runtime: dict[str, str] = {}
        self._turn_number: int = 0
        # Last user turn text — used by `sync_turn` to compose `turn.end`.
        self._last_user_text: str = ""
        # Single-flight prefetch coordination.
        self._prefetch_lock = threading.Lock()
        self._prefetch_result: str = ""
        self._prefetch_thread: threading.Thread | None = None
        # Exact memory context injected by the most recent real prefetch.
        # Compression hooks only read this cache; they must never call the
        # lifecycle-mutating `turn.start` RPC.
        self._state_lock = threading.Lock()
        self._last_injected_context: str = ""
        self._active_turn_key: str = ""
        # Hermes does not pass the live conversation history to `prefetch`.
        # Once `on_pre_compress` fires, this timestamp separates history
        # that has fallen out of the raw model window from newer turns that
        # are still visible. Before the first compression it stays unset so
        # the core preserves the legacy whole-session no-repeat behaviour.
        self._visible_context_start_ts: int | None = None
        # Tool calls accumulated via the Hermes `post_tool_call` plugin
        # hook — flushed alongside user/assistant text in `sync_turn`.
        self._tool_calls: list[dict[str, Any]] = []
        # Reasoning text captured via the `post_llm_call` hook for the
        # current turn. Hermes' MemoryProvider.sync_turn signature only
        # carries the visible assistant text; reasoning lives on the
        # `assistant` message's `reasoning` field. We capture it from
        # `post_llm_call`'s `conversation_history` so the viewer can
        # show the model's thinking like OpenClaw does.
        self._turn_thinking: str = ""
        self._hook_registered = False
        self._bridge_keepalive_stop = threading.Event()
        self._bridge_keepalive_thread: threading.Thread | None = None
        self._session_close_thread: threading.Thread | None = None
        self._session_close_requested_for = ""
        # Hermes runs background memory/skill reviewers by forking an agent and
        # appending a synthetic user turn. That turn is instruction plumbing,
        # not a human utterance, so it must not become a MemOS trace.
        self._skip_current_turn = False
        # Track the last trace ID for feedback submission
        self._last_trace_id: str = ""
        self._tool_failure_streaks: dict[str, int] = {}

    # ─── Identity ─────────────────────────────────────────────────────────

    @property
    def name(self) -> str:  # type: ignore[override]
        return "memtensor"

    def is_available(self) -> bool:  # type: ignore[override]
        try:
            return ensure_bridge_running(probe_only=True)
        except Exception:
            return False

    # ─── Lifecycle ────────────────────────────────────────────────────────

    def initialize(self, session_id: str, **kwargs: Any) -> None:  # type: ignore[override]
        """Called once at agent startup.

        kwargs always include ``hermes_home`` and ``platform``. We stash
        them so the bridge can resolve the right `~/.hermes/memos-plugin/`
        and log the originating channel.

        Shared mode (the default) acquires a lightweight lease on one
        process-scoped Node bridge. Each provider still owns its logical
        session/episode state. ``MEMOS_HERMES_BRIDGE_MODE=legacy`` retains
        the old per-provider bridge as a one-release rollback path.
        """
        previous_bridge = self._bridge
        if previous_bridge is not None:
            old_pid = getattr(previous_bridge, "pid", "?")
            logger.info(
                "MemOS: releasing previous bridge handle (pid=%s) before re-init",
                old_pid,
            )
            self.on_session_end([])
            close_thread = self._session_close_thread
            if close_thread is not None and close_thread.is_alive():
                close_thread.join(timeout=5.5)
            with contextlib.suppress(Exception):
                previous_bridge.close()
            self._bridge = None
            self._bridge_generation = 0

        self._session_id = session_id or self._session_id
        self._session_close_requested_for = ""
        self._bridge_keepalive_stop.clear()
        with self._state_lock:
            self._last_injected_context = ""
            self._active_turn_key = ""
            self._visible_context_start_ts = None
        self._hermes_home = str(kwargs.get("hermes_home") or "")
        self._platform = str(kwargs.get("platform") or "cli")
        self._agent_identity = str(kwargs.get("agent_identity") or "hermes")
        self._shared_bridge = _shared_bridge_enabled()
        self._runtime_home = _resolved_memos_runtime_home()
        self._runtime_env = _memos_runtime_env_snapshot(self._runtime_home)

        new_bridge: MemosBridgeClient | SharedBridgeLease | None = None
        try:
            runtime_home = self._runtime_home
            runtime_env = dict(self._runtime_env)
            if self._shared_bridge:
                new_bridge = SHARED_BRIDGE_REGISTRY.acquire(
                    _shared_bridge_runtime_key(runtime_home),
                    client_factory=lambda home=str(runtime_home), env=runtime_env: (
                        MemosBridgeClient(
                            runtime_home=home,
                            extra_env=env,
                        )
                    ),
                    before_spawn=lambda home=runtime_home: _prepare_shared_bridge(home),
                    host_handlers={
                        "host.llm.complete": self._handle_host_llm_complete,
                    },
                )
            else:
                _prepare_shared_bridge(runtime_home, cleanup_legacy_zombies=True)
                new_bridge = MemosBridgeClient(
                    runtime_home=str(runtime_home),
                    extra_env=runtime_env,
                )
                new_bridge.register_host_handler(
                    "host.llm.complete",
                    self._handle_host_llm_complete,
                )
            self._bridge = new_bridge
            self._open_session(session_id, timeout=60.0)
            mode = "shared" if self._shared_bridge else "legacy"
            runtime_id = getattr(new_bridge, "runtime_id", "per-provider")
            logger.info(
                "MemOS: bridge ready mode=%s runtime=%s generation=%d pid=%s "
                "session=%s platform=%s (episode deferred)",
                mode,
                runtime_id,
                self._bridge_generation,
                getattr(new_bridge, "pid", "?"),
                self._session_id,
                self._platform,
            )
        except Exception as err:
            logger.warning("MemOS: bridge init failed — %s", err)
            if new_bridge is not None:
                with contextlib.suppress(Exception):
                    new_bridge.close()
            self._bridge = None
            self._bridge_generation = 0
        # Register a Hermes plugin hook to capture tool calls as they
        # happen. The `post_tool_call` hook fires after every tool
        # dispatch (write_file, terminal, search_files, etc.) with the
        # tool name, arguments, and result. We accumulate them and
        # flush in `sync_turn`.
        self._register_tool_call_hook()
        self._start_bridge_keepalive()

    def system_prompt_block(self) -> str:  # type: ignore[override]
        return (
            "# MemOS Memory\n"
            "Persistent long-term memory is active. Call `memos_search`, "
            "`memos_get`, `memos_timeline`, `memos_environment`, "
            "`memos_skill_list`, or `memos_skill_get` when prior context or learned "
            "procedures would help. Relevant memories are automatically "
            "injected at the start of every turn.\n\n"
            "**Not the same as repo skills:** Hermes' `<available_skills>` / "
            "`skill_view(name=…)` load **repository SKILL.md** files. "
            "`memos_skill_get` / `memos_skill_list` refer to **MemOS-crystallized** "
            "skills (learned from your runs). If both apply, you may use "
            "both: repo skills for product conventions, MemOS skills for "
            "workflows proven on *your* past tasks."
        )

    # ─── Episode tracking ─────────────────────────────────────────────────
    #
    # We DON'T call `episode.open` ourselves. The core's `onTurnStart`
    # (RPC `turn.start`) automatically opens / reopens / boundary-cuts
    # an episode based on V7 §0.1 relation classification. Calling
    # `episode.open` from the adapter creates an orphan episode that
    # never receives any traces — and our `episode.close` then closes
    # that empty orphan, leaving the *real* episode (the one the
    # pipeline auto-created) without the close trigger that fires
    # reflect → reward → L2 / L3 / Skill.
    #
    # The real episode id surfaces in the `turn.start` response's
    # `query.episodeId` field; we stash it here so `on_session_end`
    # can close the right one.

    # ─── Tool call capture via Hermes plugin hook ──────────────────────────

    def _matches_session(self, session_id: str = "") -> bool:
        """Return True when a global Hermes hook belongs to this provider."""
        return not session_id or not self._session_id or session_id == self._session_id

    def _runtime_namespace(self) -> dict[str, Any]:
        profile_id = (self._agent_identity or "").strip() or "default"
        normalized_home = self._hermes_home.replace("\\", "/").rstrip("/")
        if normalized_home:
            marker = "/profiles/"
            if marker in normalized_home:
                profile_id = normalized_home.rsplit(marker, 1)[-1].split("/", 1)[0] or profile_id
            elif normalized_home.endswith("/.hermes") and profile_id in ("", "hermes"):
                profile_id = "default"
        return {
            "agentKind": "hermes",
            "profileId": profile_id,
            "profileLabel": profile_id,
        }

    def _record_namespace(self) -> dict[str, Any]:
        """Namespace used for write-path records.

        Hermes delegation hooks can be global and occasionally arrive through
        a provider instance whose `profileId` fell back to `default` while
        `agent_identity` still carries the real profile label (for example
        coder10). For writes, prefer the concrete non-default label so
        subagent outcome traces inherit the parent profile instead of leaking
        into hermes/default.
        """
        ns = dict(self._runtime_namespace())
        label = (self._agent_identity or ns.get("profileLabel") or "").strip()
        profile_id = str(ns.get("profileId") or "").strip()
        if profile_id in ("", "default", "hermes") and label and label not in ("default", "hermes"):
            ns["profileId"] = label
            ns["profileLabel"] = label
        return ns

    def _register_tool_call_hook(self) -> None:
        try:
            from hermes_cli.plugins import (
                get_plugin_manager,  # pyright: ignore[reportMissingImports]
            )

            mgr = get_plugin_manager()
            HERMES_HOOK_DISPATCHER.bind(mgr, self)
            self._hook_registered = True
            logger.debug("MemOS: bound provider session=%s to shared hooks", self._session_id)
        except Exception as err:
            logger.debug("MemOS: could not register tool hook — %s", err)

    def _on_transform_tool_result(
        self,
        tool_name: str = "",
        arguments: dict | None = None,
        result: str = "",
        task_id: str | None = None,
        **kwargs: Any,
    ) -> str | None:
        """Append a small repair hint after repeated same-turn tool failures."""
        session_id = str(kwargs.get("session_id") or kwargs.get("sessionId") or "")
        if not self._matches_session(session_id):
            return None

        tool = str(tool_name or kwargs.get("toolName") or "unknown_tool")
        if not self._tool_result_failed(result, kwargs):
            self._tool_failure_streaks.pop(tool, None)
            return None

        count = self._tool_failure_streaks.get(tool, 0) + 1
        self._tool_failure_streaks[tool] = count
        if count < _TOOL_FAILURE_HINT_THRESHOLD:
            return None
        if _TOOL_FAILURE_REPAIR_HINT in (result or ""):
            return None
        text = (result or "").rstrip()
        return f"{text}\n\n{_TOOL_FAILURE_REPAIR_HINT}" if text else _TOOL_FAILURE_REPAIR_HINT

    @staticmethod
    def _tool_result_failed(result: str, payload: dict[str, Any]) -> bool:
        for key in ("is_error", "isError", "error", "failed"):
            value = payload.get(key)
            if value is True:
                return True
            if isinstance(value, str) and value.strip():
                return True
        try:
            parsed = json.loads(result or "")
        except Exception:
            parsed = None
        if isinstance(parsed, dict):
            error = parsed.get("error")
            if error is True:
                return True
            if isinstance(error, str) and error.strip():
                return True
            if parsed.get("is_error") is True or parsed.get("isError") is True:
                return True
        normalized = " ".join((result or "").strip().lower().split())
        if not normalized:
            return False
        failure_prefixes = (
            "error:",
            "failed:",
            "failure:",
            "exception:",
            "traceback ",
            "traceback:",
            "command failed",
            "tool failed",
        )
        if normalized.startswith(failure_prefixes):
            return True
        return " traceback (most recent call last)" in normalized

    def _on_post_tool_call(
        self,
        *,
        tool_name: str = "",
        args: dict | None = None,
        result: str = "",
        tool_call_id: str = "",
        session_id: str = "",
        **kw: Any,
    ) -> None:
        """Accumulate a tool call record for the current turn.

        We keep the host's ``tool_call_id`` on a private ``_id`` field so
        ``_on_post_llm_call`` can later attach the assistant message's
        ``reasoning`` (the model's "thinking before this tool") to the
        right entry. Hermes/OpenAI-compatible providers may surface the
        same call under ``id``, ``call_id``, or ``response_item_id``; keep
        all aliases so post-LLM and post-tool events can be merged even
        when a particular tool omits one field. Private fields are stripped
        before the JSON-RPC send.
        """
        if not self._matches_session(session_id):
            return
        ids = self._tool_call_ids(
            {
                "id": tool_call_id,
                "call_id": kw.get("call_id"),
                "response_item_id": kw.get("response_item_id"),
                "tool_call_id": kw.get("tool_call_id"),
            }
        )
        input_text = (
            json.dumps(args, ensure_ascii=False) if isinstance(args, dict) else str(args or "")
        )
        timing = self._coerce_tool_timing(kw)

        existing = self._find_tool_call(ids)
        if existing is not None:
            existing["name"] = tool_name or existing.get("name") or "unknown_tool"
            existing["input"] = input_text or existing.get("input", "")
            existing["output"] = (result or "")[:4000]
            existing["_ids"] = sorted(set((existing.get("_ids") or []) + ids))
            existing["_id"] = existing.get("_id") or (ids[0] if ids else "")
            if existing.get("_id"):
                existing["toolCallId"] = existing["_id"]
            if timing:
                existing.update(timing)
            return

        call = {
            "name": tool_name,
            "input": input_text,
            "output": (result or "")[:4000],
            "_id": ids[0] if ids else "",
            "_ids": ids,
            "toolCallId": ids[0] if ids else "",
        }
        if timing:
            call.update(timing)
        self._tool_calls.append(call)

    def _coerce_tool_timing(self, payload: dict[str, Any]) -> dict[str, int] | None:
        """Preserve real tool timing if Hermes exposes it in hook kwargs."""
        started = self._coerce_epoch_ms(
            payload.get("startedAt")
            or payload.get("started_at")
            or payload.get("startTime")
            or payload.get("start_time")
        )
        ended = self._coerce_epoch_ms(
            payload.get("endedAt")
            or payload.get("ended_at")
            or payload.get("endTime")
            or payload.get("end_time")
        )
        if started is not None and ended is not None and ended > started:
            return {"startedAt": started, "endedAt": ended}

        duration = self._coerce_duration_ms(
            payload.get("durationMs")
            or payload.get("duration_ms")
            or payload.get("elapsedMs")
            or payload.get("elapsed_ms")
            or payload.get("latencyMs")
            or payload.get("latency_ms")
        )
        if duration is not None and duration > 0:
            end_ms = int(time.time() * 1000)
            return {"startedAt": end_ms - duration, "endedAt": end_ms}

        return None

    @staticmethod
    def _coerce_epoch_ms(value: Any) -> int | None:
        if isinstance(value, int | float):
            numeric = float(value)
        elif isinstance(value, str):
            try:
                numeric = float(value)
            except ValueError:
                return None
        else:
            return None
        if numeric <= 0:
            return None
        # Accept seconds or milliseconds.
        if numeric < 10_000_000_000:
            numeric *= 1000
        return int(numeric)

    @staticmethod
    def _coerce_duration_ms(value: Any) -> int | None:
        if isinstance(value, int | float):
            numeric = float(value)
        elif isinstance(value, str):
            try:
                numeric = float(value)
            except ValueError:
                return None
        else:
            return None
        if numeric <= 0:
            return None
        return int(numeric)

    def _on_post_llm_call(
        self,
        *,
        conversation_history: list[dict[str, Any]] | None = None,
        user_message: str = "",
        session_id: str = "",
        **_kw: Any,
    ) -> None:
        """Capture reasoning content from assistant messages in this turn.

        Hermes' ``_build_assistant_message`` writes the model's reasoning
        text into ``msg["reasoning"]`` (extended thinking, OpenAI o1
        ``reasoning_content``, etc.). The default ``MemoryProvider.sync_turn``
        only carries plain ``user_content`` / ``assistant_content``, so we
        fish the reasoning out of the conversation history fired with the
        ``post_llm_call`` hook and stash it for the upcoming ``sync_turn``.

        We walk through assistant messages of the current turn (those
        after the most recent user message). For each message that
        contains ``tool_calls``, we attach two pieces of pre-tool context
        to each captured tool call:

        * ``thinkingBefore`` — private/model-native reasoning.
        * ``assistantTextBefore`` — visible assistant narration emitted in
          the same message before the tool call.

        The final reasoning (the message that produced the user-facing
        reply) becomes the turn-level ``agentThinking``.
        """
        if not self._matches_session(session_id):
            return
        if not conversation_history:
            return

        # Find the last user message and walk forward from there.
        last_user_idx = -1
        for i, msg in enumerate(conversation_history):
            if msg.get("role") == "user":
                last_user_idx = i

        # Build maps keyed by tool_call_id so post-tool events can be
        # merged with the canonical assistant message later.
        thinking_by_id: dict[str, str] = {}
        assistant_text_by_id: dict[str, str] = {}
        ordered_tool_calls: list[dict[str, Any]] = []
        ordered_object_ids: set[int] = set()
        # Reasoning of the message that produced the final reply (no
        # tool_calls in that message) becomes the turn-level thinking.
        final_reasoning = ""

        for msg in conversation_history[last_user_idx + 1 :]:
            if msg.get("role") != "assistant":
                continue
            r = msg.get("reasoning")
            r_str = r.strip() if isinstance(r, str) and r.strip() else ""
            content_str = self._assistant_text(msg.get("content"))
            tcs = msg.get("tool_calls")
            if isinstance(tcs, list) and tcs:
                # Reasoning preceded these tool calls.
                for tc in tcs:
                    if not isinstance(tc, dict):
                        continue
                    ids = self._tool_call_ids(tc)
                    if r_str:
                        for tc_id in ids:
                            thinking_by_id[tc_id] = r_str
                    if content_str:
                        for tc_id in ids:
                            assistant_text_by_id[tc_id] = content_str

                    existing = self._find_tool_call(ids)
                    # Some Hermes tools (for example planner/todo-style
                    # host tools) appear in the assistant message but do
                    # not fire `post_tool_call`. Add a placeholder so the
                    # trace still records the tool decision and reasoning;
                    # `post_tool_call` will merge real output later if it
                    # eventually arrives.
                    if existing is None:
                        existing = {
                            "name": self._tool_name(tc),
                            "input": self._tool_input(tc),
                            "output": "",
                            "thinkingBefore": r_str or "",
                            "assistantTextBefore": content_str or "",
                            "_id": ids[0] if ids else "",
                            "_ids": ids,
                            "toolCallId": ids[0] if ids else "",
                        }
                        self._tool_calls.append(existing)
                    else:
                        # Preserve output captured by post_tool_call, but
                        # let the LLM message supply canonical order,
                        # input/name aliases, and thinkingBefore.
                        existing["name"] = existing.get("name") or self._tool_name(tc)
                        existing["input"] = existing.get("input") or self._tool_input(tc)
                        existing["thinkingBefore"] = r_str or existing.get("thinkingBefore", "")
                        existing["assistantTextBefore"] = content_str or existing.get(
                            "assistantTextBefore", ""
                        )
                        existing["_ids"] = sorted(set((existing.get("_ids") or []) + ids))
                        existing["_id"] = existing.get("_id") or (ids[0] if ids else "")
                        if existing.get("_id"):
                            existing["toolCallId"] = existing["_id"]

                    marker = id(existing)
                    if marker not in ordered_object_ids:
                        ordered_tool_calls.append(existing)
                        ordered_object_ids.add(marker)
            else:
                # Plain assistant reply — overwrite final_reasoning so we
                # keep the LATEST one (mirrors Hermes' ``last_reasoning``).
                if r_str:
                    final_reasoning = r_str

        # Make the turn payload follow the LLM-declared tool order. This
        # matters when post_tool_call fires for later tools before
        # post_llm_call backfills earlier planner/todo calls.
        if ordered_tool_calls:
            remaining = [tc for tc in self._tool_calls if id(tc) not in ordered_object_ids]
            self._tool_calls = ordered_tool_calls + remaining

        # Attach thinkingBefore to matching captured tool calls.
        for tc in self._tool_calls:
            ids = tc.get("_ids") or ([tc.get("_id")] if tc.get("_id") else [])
            for tc_id in ids:
                if tc_id and tc_id in thinking_by_id:
                    tc["thinkingBefore"] = thinking_by_id[tc_id]
                    break
            for tc_id in ids:
                if tc_id and tc_id in assistant_text_by_id:
                    tc["assistantTextBefore"] = assistant_text_by_id[tc_id]
                    break

        self._turn_thinking = final_reasoning

    @staticmethod
    def _assistant_text(content: Any) -> str:
        """Extract visible assistant text from Hermes/OpenAI message content."""
        if isinstance(content, str):
            return content.strip()
        if isinstance(content, list):
            parts: list[str] = []
            for block in content:
                if isinstance(block, str):
                    text = block.strip()
                elif isinstance(block, dict):
                    raw = block.get("text") or block.get("content")
                    text = raw.strip() if isinstance(raw, str) else ""
                else:
                    text = ""
                if text:
                    parts.append(text)
            return "\n".join(parts).strip()
        return ""

    @staticmethod
    def _tool_call_ids(raw: dict[str, Any]) -> list[str]:
        ids: list[str] = []
        for key in ("id", "call_id", "response_item_id", "tool_call_id"):
            value = raw.get(key)
            if isinstance(value, str) and value and value not in ids:
                ids.append(value)
        return ids

    @staticmethod
    def _tool_name(raw: dict[str, Any]) -> str:
        fn = raw.get("function")
        if isinstance(fn, dict) and isinstance(fn.get("name"), str):
            return fn["name"]
        name = raw.get("name")
        return name if isinstance(name, str) and name else "unknown_tool"

    @staticmethod
    def _tool_input(raw: dict[str, Any]) -> str:
        fn = raw.get("function")
        if isinstance(fn, dict):
            args = fn.get("arguments")
            if isinstance(args, str):
                return args
            if args is not None:
                return json.dumps(args, ensure_ascii=False)
        for key in ("arguments", "args", "input"):
            args = raw.get(key)
            if isinstance(args, str):
                return args
            if args is not None:
                return json.dumps(args, ensure_ascii=False)
        return ""

    def _find_tool_call(self, ids: list[str]) -> dict[str, Any] | None:
        if not ids:
            return None
        needle = set(ids)
        for tc in self._tool_calls:
            existing = set(tc.get("_ids") or [])
            if tc.get("_id"):
                existing.add(str(tc["_id"]))
            if existing & needle:
                return tc
        return None

    # ─── Turn-level hooks ─────────────────────────────────────────────────

    def on_turn_start(self, turn_number: int, message: str, **_kwargs: Any) -> None:  # type: ignore[override]
        self._turn_number = int(turn_number or 0)
        self._skip_current_turn = _is_hermes_internal_review_prompt(message)
        self._last_user_text = "" if self._skip_current_turn else (message or "").strip()
        with self._state_lock:
            self._last_injected_context = ""
            self._active_turn_key = (
                f"{self._session_id}:{self._turn_number}"
                if self._session_id and self._turn_number > 0
                else ""
            )
        # Reset per-turn buffers so reasoning / tool calls captured here
        # belong only to this turn.
        self._turn_thinking = ""
        self._tool_calls = []
        self._tool_failure_streaks = {}

    def prefetch(self, query: str, *, session_id: str = "") -> str:  # type: ignore[override]
        """Inject relevant memories ahead of the next model call.

        If ``queue_prefetch`` already ran for this turn, return the
        cached result immediately. Otherwise synchronously run
        ``turn.start`` against the bridge (small overhead).
        """
        deadline_monotonic = time.monotonic() + _PREFETCH_RPC_TIMEOUT
        started_at_ms = int(time.time() * 1000)
        core_budget_seconds = max(
            0.05,
            _PREFETCH_RPC_TIMEOUT - _PREFETCH_RESPONSE_RESERVE_SECONDS,
        )
        deadline_at_ms = started_at_ms + int(core_budget_seconds * 1000)
        if self._prefetch_thread and self._prefetch_thread.is_alive():
            try:
                join_timeout = _remaining_rpc_timeout(deadline_monotonic, 5.0)
            except BridgeError:
                return ""
            self._prefetch_thread.join(timeout=join_timeout)
        with self._prefetch_lock:
            cached = self._prefetch_result
            self._prefetch_result = ""
        if self._skip_current_turn or _is_hermes_internal_review_prompt(query):
            self._skip_current_turn = True
            return ""
        suppress_injection = _is_explicit_delegation_request(query)
        if cached:
            return "" if suppress_injection else cached
        try:
            ensure_timeout = _remaining_rpc_timeout(
                deadline_monotonic,
                _PREFETCH_RPC_TIMEOUT,
            )
        except BridgeError:
            return ""
        if not self._ensure_bridge(
            session_id or self._session_id,
            timeout=min(10.0, ensure_timeout or _PREFETCH_RPC_TIMEOUT),
        ):
            return ""
        try:
            context = self._turn_start(
                query,
                session_id=session_id,
                deadline_monotonic=deadline_monotonic,
                deadline_at_ms=deadline_at_ms,
            )
            if suppress_injection:
                # Do not let remembered "do it directly" skills override an
                # explicit user request to dispatch work to a subagent.
                with self._state_lock:
                    self._last_injected_context = ""
                return ""
            with self._state_lock:
                self._last_injected_context = context[:_COMPRESSION_CONTEXT_MAX_CHARS]
            return context
        except Exception as err:
            logger.debug("MemOS: prefetch failed — %s", err)
            return ""

    def queue_prefetch(self, query: str, *, session_id: str = "") -> None:  # type: ignore[override]
        """No-op for MemOS.

        Hermes calls this AFTER ``sync_turn`` to warm the cache for a
        hypothetical next turn. In the V7 architecture each ``turn.end``
        triggers async capture / reward / induction work — running another
        ``turn.start`` against the same (already-closed) episode just
        races and produces ``episode already closed`` noise in the
        viewer's logs page. ``prefetch()`` (called BEFORE the next
        turn's LLM call) handles real retrieval; this hook is moot.
        """
        return

    def sync_turn(
        self,
        user_content: str,
        assistant_content: str,
        *,
        session_id: str = "",
    ) -> None:  # type: ignore[override]
        """Persist a completed turn immediately.

        Tool calls are captured via the Hermes ``post_tool_call``
        plugin hook (registered in ``initialize``). By the time
        ``sync_turn`` is called the full list of tool calls for this
        turn has already been accumulated in ``self._tool_calls``.
        """
        user = user_content or self._last_user_text
        assistant = assistant_content or ""
        tool_calls = self._tool_calls
        thinking = self._turn_thinking
        self._tool_calls = []
        self._turn_thinking = ""
        if self._skip_current_turn or _is_hermes_internal_review_prompt(user):
            self._skip_current_turn = False
            self._last_user_text = ""
            return
        if not self._ensure_bridge(session_id or self._session_id, timeout=75.0):
            logger.warning("MemOS: sync_turn skipped because bridge is unavailable")
            return
        logger.info(
            "MemOS: sync_turn user=%d assistant=%d tools=%d thinking=%d",
            len(user),
            len(assistant),
            len(tool_calls),
            len(thinking),
        )
        ts_ms = int(time.time() * 1000)
        is_feedback_turn = _is_verifier_feedback_prompt(user)
        feedback_submitted = False
        try:
            if user and not self._episode_id:
                self._turn_start(user, session_id=session_id or self._session_id)
            current_trace_id = self._turn_end(
                user,
                assistant,
                tool_calls,
                ts_ms,
                agent_thinking=thinking,
            )
            if is_feedback_turn:
                feedback_submitted = self._try_submit_verifier_feedback(
                    user,
                    assistant,
                    ts_ms,
                    trace_id=current_trace_id,
                )
        except Exception as err:
            if not self._is_transport_closed(err):
                logger.warning("MemOS: sync_turn turn.end failed — %s", err)
            else:
                logger.warning(
                    "MemOS: bridge transport closed during sync_turn; "
                    "reconnecting and retrying once — %s",
                    err,
                )
                try:
                    self._reconnect_bridge(session_id or self._session_id, timeout=75.0)
                    if user:
                        self._turn_start(user, session_id=session_id or self._session_id)
                    current_trace_id = self._turn_end(
                        user,
                        assistant,
                        tool_calls,
                        ts_ms,
                        agent_thinking=thinking,
                    )
                    if is_feedback_turn and not feedback_submitted:
                        feedback_submitted = self._try_submit_verifier_feedback(
                            user,
                            assistant,
                            ts_ms,
                            trace_id=current_trace_id,
                        )
                except Exception:
                    logger.exception(
                        "MemOS: sync_turn failed after bridge reconnect; "
                        "memory turn was not persisted"
                    )
        if is_feedback_turn and not feedback_submitted:
            # turn.end may time out while the bridge continues lite capture in
            # the background. Preserve the user's explicit signal at episode
            # scope instead of dropping Decision Repair entirely.
            self._try_submit_verifier_feedback(
                user,
                assistant,
                ts_ms,
                trace_id="",
                fallback=True,
            )
        if user_content:
            self._last_user_text = user_content

    def on_delegation(
        self,
        task: str,
        result: str,
        *,
        child_session_id: str = "",
        **kwargs: Any,
    ) -> None:  # type: ignore[override]
        """Record a subagent outcome.

        Hermes invokes this on the **parent** when a subagent finishes.
        We write it as a synthetic trace so decision-repair can see
        failure bursts and so Tier 2 retrieval can surface past
        delegations.
        """
        if not self._ensure_bridge(self._session_id, timeout=30.0):
            return
        try:
            if not self._episode_id and self._last_user_text:
                self._turn_start(self._last_user_text, session_id=self._session_id)
            namespace = self._record_namespace()
            hook_meta = {
                "hookKwargs": kwargs,
                "namespace": namespace,
            }
            self._bridge_request(
                "subagent.record",
                {
                    "agent": "hermes",
                    "namespace": namespace,
                    "sessionId": self._session_id,
                    "episodeId": self._episode_id or None,
                    "childSessionId": child_session_id or None,
                    "task": task,
                    "result": result,
                    "toolCalls": self._extract_child_tool_calls(child_session_id),
                    "ts": int(time.time() * 1000),
                    "meta": hook_meta,
                    "contextHints": {
                        "agentIdentity": self._agent_identity,
                        "namespace": namespace,
                        **self._host_runtime_context(),
                    },
                },
            )
        except Exception as err:
            logger.warning("MemOS: subagent.record failed — %s", err)

    def _extract_child_tool_calls(self, child_session_id: str = "") -> list[dict[str, Any]]:
        """Best-effort recovery of subagent tool calls from Hermes session JSON.

        Hermes invokes ``on_delegation`` on the parent and only passes the
        child task/result. The child transcript is still persisted under
        ``$HERMES_HOME/sessions/session_<id>.json``, so we read that file to
        preserve structured tool use in the MemOS child episode.
        """
        if not child_session_id:
            return []
        sessions_dir = (
            Path(self._hermes_home).expanduser() / "sessions"
            if self._hermes_home
            else Path.home() / ".hermes" / "sessions"
        )
        session_path = sessions_dir / f"session_{child_session_id}.json"
        try:
            payload = json.loads(session_path.read_text(encoding="utf-8"))
        except Exception as err:
            logger.debug("MemOS: child session tool extraction skipped — %s", err)
            return []

        messages = payload.get("messages")
        if not isinstance(messages, list):
            return []

        tool_outputs: dict[str, str] = {}
        for message in messages:
            if not isinstance(message, dict) or message.get("role") != "tool":
                continue
            tool_call_id = str(message.get("tool_call_id") or "")
            if tool_call_id:
                tool_outputs[tool_call_id] = str(message.get("content") or "")[:4000]

        base_ts = int(time.time() * 1000)
        calls: list[dict[str, Any]] = []
        for message in messages:
            if not isinstance(message, dict):
                continue
            raw_calls = message.get("tool_calls")
            if not isinstance(raw_calls, list):
                continue
            for raw_call in raw_calls:
                if not isinstance(raw_call, dict):
                    continue
                function = raw_call.get("function")
                if not isinstance(function, dict):
                    function = {}
                call_id = str(
                    raw_call.get("id")
                    or raw_call.get("call_id")
                    or raw_call.get("tool_call_id")
                    or ""
                )
                raw_args = function.get("arguments", raw_call.get("arguments", ""))
                output = tool_outputs.get(call_id, "")
                call: dict[str, Any] = {
                    "name": str(function.get("name") or raw_call.get("name") or "tool"),
                    "input": self._json_or_raw(raw_args),
                    "output": output,
                    "startedAt": base_ts + len(calls),
                    "endedAt": base_ts + len(calls),
                }
                parsed_output = self._json_or_raw(output)
                if isinstance(parsed_output, dict) and parsed_output.get("error"):
                    call["errorCode"] = "tool_error"
                calls.append(call)
        return calls

    @staticmethod
    def _json_or_raw(value: Any) -> Any:
        if not isinstance(value, str):
            return value
        try:
            return json.loads(value)
        except Exception:
            return value

    def on_pre_compress(self, messages: list[dict[str, Any]]) -> str:  # type: ignore[override]
        """Return the memory context injected by the current/previous turn.

        This hook must stay read-only. Calling ``turn.start`` here used to
        finalize the active lightweight episode and create a phantom episode
        on every compression pass. Reusing the exact context previously
        injected by ``prefetch`` also keeps the summary faithful to what the
        model actually saw.
        """
        with self._state_lock:
            # Everything captured before this boundary is about to leave the
            # raw conversation window. The shared retrieval core interprets
            # this as "same-session traces older than this are eligible".
            self._visible_context_start_ts = int(time.time() * 1000)
            context = self._last_injected_context.strip()
        if not context:
            return ""
        return f"MemOS memory snapshot (preserved across compression):\n{context}"

    # ─── Tool surface ─────────────────────────────────────────────────────

    @staticmethod
    def _clip(value: Any, limit: int = 1200) -> str:
        text = "" if value is None else str(value)
        return text if len(text) <= limit else text[:limit] + "..."

    @staticmethod
    def _int_arg(args: dict[str, Any], key: str, default: int, lower: int, upper: int) -> int:
        try:
            value = int(args.get(key, default))
        except Exception:
            value = default
        return max(lower, min(upper, value))

    def get_tool_schemas(self) -> list[dict[str, Any]]:  # type: ignore[override]
        return [
            {
                "name": "memos_search",
                "description": (
                    "Search the local MemOS memory (traces, policies, world models, skills). "
                    "Prefer this before claiming prior context is unavailable."
                ),
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "Short natural-language query (2–5 key words).",
                        },
                        "maxResults": {
                            "type": "integer",
                            "default": 10,
                            "minimum": 1,
                            "maximum": 50,
                        },
                        "sessionScope": {
                            "type": "boolean",
                            "default": False,
                            "description": "Restrict results to the current Hermes session only.",
                        },
                    },
                    "required": ["query"],
                },
            },
            {
                "name": "memos_get",
                "description": (
                    "Fetch the full body of a memory item by id. `kind` can be "
                    '"trace" (default), "policy", or "world_model".'
                ),
                "parameters": {
                    "type": "object",
                    "properties": {
                        "id": {"type": "string"},
                        "kind": {
                            "type": "string",
                            "enum": ["trace", "policy", "world_model"],
                            "default": "trace",
                        },
                    },
                    "required": ["id"],
                },
            },
            {
                "name": "memos_timeline",
                "description": "Return the ordered traces for an episode id.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "episodeId": {"type": "string"},
                        "limit": {"type": "integer", "default": 20, "maximum": 100},
                    },
                    "required": ["episodeId"],
                },
            },
            {
                "name": "memos_skill_list",
                "description": (
                    "List callable skills the agent can invoke. Filter by status "
                    "(candidate | active | archived)."
                ),
                "parameters": {
                    "type": "object",
                    "properties": {
                        "status": {
                            "type": "string",
                            "enum": ["candidate", "active", "archived"],
                        },
                        "limit": {
                            "type": "integer",
                            "default": 10,
                            "minimum": 1,
                            "maximum": 50,
                        },
                    },
                },
            },
            {
                "name": "memos_environment",
                "description": (
                    "Return accumulated environment knowledge (L3 world models): "
                    "structural facts, behavioral rules, and project constraints."
                ),
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "Optional keyword query; omit to list recent world models.",
                        },
                        "limit": {
                            "type": "integer",
                            "default": 5,
                            "minimum": 1,
                            "maximum": 30,
                        },
                    },
                },
            },
            {
                "name": "memos_skill_get",
                "description": "Return the full invocation guide for a crystallized skill.",
                "parameters": {
                    "type": "object",
                    "properties": {"id": {"type": "string"}},
                    "required": ["id"],
                },
            },
        ]

    def handle_tool_call(self, tool_name: str, args: dict[str, Any], **_kwargs: Any) -> str:  # type: ignore[override]
        if not self._bridge:
            return json.dumps({"error": "bridge not connected"}, ensure_ascii=False)
        try:
            if tool_name == "memos_search":
                query = (args.get("query") or "").strip()
                if not query:
                    return json.dumps({"error": "missing query"}, ensure_ascii=False)
                max_results = self._int_arg(args, "maxResults", 10, 1, 50)
                params: dict[str, Any] = {
                    "agent": "hermes",
                    "namespace": self._runtime_namespace(),
                    "query": query,
                    "topK": {
                        "tier1": max_results,
                        "tier2": max_results,
                        "tier3": max_results,
                    },
                }
                if bool(args.get("sessionScope", False)):
                    params["sessionId"] = self._session_id
                resp = self._bridge_request_with_retry(
                    "memory.search",
                    params,
                    timeout=_LONG_RPC_TIMEOUT,
                )
                return json.dumps({"hits": resp.get("hits", [])}, ensure_ascii=False)
            if tool_name == "memos_get":
                item_id = (args.get("id") or "").strip()
                if not item_id:
                    return json.dumps({"error": "missing id"}, ensure_ascii=False)
                kind = args.get("kind") or "trace"
                methods = {
                    "trace": "memory.get_trace",
                    "policy": "memory.get_policy",
                    "world_model": "memory.get_world",
                }
                method = methods.get(kind)
                if method is None:
                    return json.dumps({"error": f"unknown memory kind: {kind}"}, ensure_ascii=False)
                item = self._bridge_request_with_retry(
                    method, {"id": item_id, "namespace": self._runtime_namespace()}
                )
                if not item:
                    return json.dumps(
                        {"found": False, "kind": kind, "id": item_id}, ensure_ascii=False
                    )
                if kind == "trace":
                    body = self._clip(item.get("agentText") or item.get("body"))
                    meta = {
                        "episodeId": item.get("episodeId"),
                        "ts": item.get("ts"),
                        "value": item.get("value"),
                        "reflection": self._clip(item.get("reflection")),
                        "userText": self._clip(item.get("userText")),
                        "toolCalls": item.get("toolCalls") or [],
                    }
                elif kind == "policy":
                    body = self._clip(
                        "\n\n".join(
                            part for part in [item.get("title"), item.get("procedure")] if part
                        )
                    )
                    meta = {
                        "trigger": item.get("trigger"),
                        "verification": item.get("verification"),
                        "boundary": item.get("boundary"),
                        "gain": item.get("gain"),
                        "support": item.get("support"),
                        "status": item.get("status"),
                    }
                else:
                    body = self._clip(item.get("body"))
                    meta = {
                        "title": item.get("title"),
                        "policyIds": item.get("policyIds") or [],
                    }
                return json.dumps(
                    {
                        "found": True,
                        "kind": kind,
                        "id": item.get("id", item_id),
                        "body": body,
                        "meta": meta,
                    },
                    ensure_ascii=False,
                )
            if tool_name == "memos_timeline":
                resp = self._bridge_request_with_retry(
                    "memory.timeline",
                    {
                        "episodeId": args.get("episodeId", self._episode_id),
                        "namespace": self._runtime_namespace(),
                    },
                )
                limit = self._int_arg(args, "limit", 20, 1, 100)
                traces = resp.get("traces", [])[:limit]
                return json.dumps({"traces": traces}, ensure_ascii=False)
            if tool_name == "memos_skill_list":
                limit = self._int_arg(args, "limit", 10, 1, 50)
                params = {"limit": limit, "namespace": self._runtime_namespace()}
                if args.get("status"):
                    params["status"] = args["status"]
                return json.dumps(
                    self._bridge_request_with_retry("skill.list", params),
                    ensure_ascii=False,
                )
            if tool_name == "memos_environment":
                query = (args.get("query") or "").strip()
                limit = self._int_arg(args, "limit", 5, 1, 30)
                if not query:
                    resp = self._bridge_request_with_retry(
                        "memory.list_world_models",
                        {"limit": limit, "offset": 0, "namespace": self._runtime_namespace()},
                    )
                    return json.dumps(
                        {
                            "worldModels": [
                                {
                                    **w,
                                    "body": self._clip(w.get("body")),
                                }
                                for w in resp.get("worldModels", [])
                            ],
                            "queried": False,
                        },
                        ensure_ascii=False,
                    )
                resp = self._bridge_request_with_retry(
                    "memory.search",
                    {
                        "agent": "hermes",
                        "namespace": self._runtime_namespace(),
                        "query": query,
                        "topK": {"tier1": 0, "tier2": 0, "tier3": limit},
                    },
                    timeout=_LONG_RPC_TIMEOUT,
                )
                hits = [
                    h
                    for h in resp.get("hits", [])
                    if h.get("tier") == 3 or h.get("refKind") == "world_model"
                ]
                return json.dumps(
                    {
                        "worldModels": [
                            {
                                "id": h.get("refId") or h.get("id"),
                                "title": self._clip((h.get("snippet") or "").split("\n")[0]),
                                "body": self._clip(h.get("snippet")),
                                "policyIds": [],
                                "score": h.get("score"),
                            }
                            for h in hits[:limit]
                        ],
                        "queried": True,
                    },
                    ensure_ascii=False,
                )
            if tool_name == "memos_skill_get":
                skill_id = (args.get("id") or "").strip()
                if not skill_id:
                    return json.dumps({"error": "missing id"}, ensure_ascii=False)
                skill = self._bridge_request_with_retry(
                    "skill.get",
                    {
                        "id": skill_id,
                        "namespace": self._runtime_namespace(),
                        "recordTrial": True,
                        "sessionId": self._session_id,
                        "episodeId": self._episode_id or None,
                    },
                )
                return json.dumps({"found": bool(skill), "skill": skill}, ensure_ascii=False)
        except Exception as err:
            return json.dumps({"error": str(err)}, ensure_ascii=False)
        return json.dumps({"error": f"unknown tool: {tool_name}"}, ensure_ascii=False)

    # ─── Config schema (for `hermes memory setup`) ────────────────────────

    def get_config_schema(self) -> list[dict[str, Any]]:  # type: ignore[override]
        """Fields the host's `hermes memory setup` wizard will collect.

        Secrets go to .env; everything else to the provider config file
        written by ``save_config``.
        """
        return [
            {
                "key": "viewer_port",
                "description": "Fixed local HTTP port for the MemOS Hermes viewer.",
                "default": 18800,
                "required": False,
            },
            {
                "key": "llm_provider",
                "description": "LLM for V7 reward / l2.induction / l3.abstraction.",
                "choices": ["openai_compatible", "anthropic", "gemini", "host", "local_only"],
                "default": "openai_compatible",
                "required": False,
            },
            {
                "key": "llm_api_key",
                "description": "API key for the chosen LLM provider.",
                "secret": True,
                "env_var": "MEMOS_LLM_API_KEY",
                "required": False,
            },
            {
                "key": "embedding_provider",
                "description": "Embedding provider (local = MiniLM on-device).",
                "choices": [
                    "local",
                    "openai_compatible",
                    "gemini",
                    "cohere",
                    "voyage",
                    "mistral",
                ],
                "default": "local",
                "required": False,
            },
        ]

    def save_config(self, values: dict[str, Any], hermes_home: str) -> None:  # type: ignore[override]
        """Write non-secret config to `<hermes_home>/memos-plugin/config.yaml`."""
        if not hermes_home:
            return
        import yaml  # lazy import — hermes already ships pyyaml

        if self._runtime_home is not None:
            target_dir = self._runtime_home
        elif os.name == "nt":
            target_dir = resolve_runtime_home(plugin_root=_PLUGIN_DIR.parents[2])
        else:
            target_dir = Path(hermes_home) / "memos-plugin"
        target_dir.mkdir(parents=True, exist_ok=True)
        target = target_dir / "config.yaml"

        payload: dict[str, Any] = {"version": 1}
        if "viewer_port" in values:
            # Keep the legacy setup field for host compatibility, but the
            # Hermes adapter owns :18800. Persist the effective value so the
            # YAML file never advertises a port the runtime will not bind.
            payload["viewer"] = {"port": 18800}
        if "llm_provider" in values:
            llm: dict[str, Any] = {"provider": values["llm_provider"]}
            if values.get("llm_provider") != "local_only":
                llm["apiKey"] = ""
            payload["llm"] = llm
        if "embedding_provider" in values:
            payload["embedding"] = {"provider": values["embedding_provider"]}

        target.write_text(yaml.safe_dump(payload, sort_keys=False), encoding="utf-8")
        target.chmod(0o600)

    # ─── Session-end ──────────────────────────────────────────────────────

    def on_session_end(self, messages: list[dict[str, Any]]) -> None:  # type: ignore[override]
        with self._state_lock:
            self._last_injected_context = ""
            self._active_turn_key = ""
            self._visible_context_start_ts = None
        if not self._bridge:
            return
        # `sync_turn` already flushed completed turn data synchronously.
        # Closing the host session is not the same as ending the topic:
        # the core will pause or finalize the open episode according to
        # topic-boundary rules so interrupted Hermes sessions can resume
        # into the same task later.
        #
        # Fire session.close in a daemon thread — the response is unused, so
        # this is semantically fire-and-forget. Calling urlopen() inline blocks
        # the asyncio event loop (gateway/run.py calls us synchronously from
        # _handle_reset_command) and causes Discord heartbeat timeouts when the
        # bridge is unresponsive. 5 s timeout keeps it bounded.
        _sid = self._session_id
        if not _sid or self._session_close_requested_for == _sid:
            return
        self._session_close_requested_for = _sid

        def _close() -> None:
            try:
                self._bridge_request(
                    "session.close",
                    {"sessionId": _sid},
                    timeout=5.0,
                )
            except Exception as err:
                if self._session_close_requested_for == _sid:
                    self._session_close_requested_for = ""
                logger.debug("MemOS: session.close failed session=%s — %s", _sid, err)

        self._session_close_thread = threading.Thread(
            target=_close,
            daemon=True,
            name="memos-session-close",
        )
        self._session_close_thread.start()

    def __del__(self) -> None:
        # Safety net — if shutdown() was never called (e.g. caller forgot,
        # Hermes agent routed model change with self.agent = None), clean
        # up the bridge subprocess and keepalive thread on GC.
        if self._bridge is not None or (
            self._bridge_keepalive_thread is not None and self._bridge_keepalive_thread.is_alive()
        ):
            logger.warning(
                "MemOS: __del__ cleaning up leaked provider — shutdown() was never called"
            )
            with contextlib.suppress(Exception):
                self.shutdown()

    def shutdown(self) -> None:  # type: ignore[override]
        with self._state_lock:
            self._last_injected_context = ""
            self._active_turn_key = ""
        HERMES_HOOK_DISPATCHER.unbind(self)
        self._hook_registered = False
        self._bridge_keepalive_stop.set()
        if self._bridge_keepalive_thread and self._bridge_keepalive_thread.is_alive():
            self._bridge_keepalive_thread.join(
                timeout=12.0
            )  # Increased to cover health check timeout (10s) + margin
        if self._prefetch_thread and self._prefetch_thread.is_alive():
            self._prefetch_thread.join(timeout=5.0)
        if self._bridge:
            self.on_session_end([])
            close_thread = self._session_close_thread
            if close_thread is not None and close_thread.is_alive():
                close_thread.join(timeout=5.5)
            pid = getattr(self._bridge, "pid", "?")
            action = (
                "releasing shared bridge lease" if self._shared_bridge else "shutting down bridge"
            )
            logger.info("MemOS: %s (pid=%s session=%s)", action, pid, self._session_id)
            with contextlib.suppress(Exception):
                self._bridge.close()
            self._bridge = None
            self._bridge_generation = 0
            logger.info("MemOS: provider bridge shutdown complete (pid=%s)", pid)

    # ─── Host LLM bridge (fallback for plugin-side model failures) ────────

    def _handle_host_llm_complete(self, params: dict[str, Any]) -> dict[str, Any]:
        """Run a fallback LLM call using the host (hermes) agent's models.

        Wired into the bridge's reverse-RPC channel under the
        ``host.llm.complete`` method. Triggered when the plugin's
        configured summary or skill-evolver model fails — instead of
        bubbling the error straight up (which would stall the V7
        capture / reflection / skill pipeline), we replay the prompt
        through ``agent.auxiliary_client.call_llm`` so hermes' own
        provider stack (including its OpenRouter / Codex / custom
        endpoint resolution) handles it.

        If the host LLM also fails this raises, the bridge converts
        that into a JSON-RPC error, the LlmClient ``markFail``s, and
        the Overview card flips red — exactly matching the spec
        "if the agent's main model is also down, stop falling back
        and surface red".
        """
        messages = params.get("messages")
        if not isinstance(messages, list) or not messages:
            raise ValueError("host.llm.complete: missing messages")

        # Lazy imports — these pull in heavy deps (openai client,
        # credential pool, …) that we don't want to load until a
        # fallback is actually requested.
        try:
            from agent.auxiliary_client import call_llm  # type: ignore[import-not-found]
            from hermes_cli.runtime_provider import (  # type: ignore[import-not-found]
                resolve_runtime_provider,
            )
        except Exception as err:
            raise RuntimeError(f"host LLM bridge unavailable: {err}") from err

        # Resolve hermes' MAIN conversation provider so the fallback
        # uses exactly what the user configured for chat. Walking the
        # generic auxiliary auto-detect chain would otherwise depend
        # on env vars (`OPENROUTER_API_KEY`, `OPENAI_API_KEY`, …) that
        # often don't propagate into the bridge subprocess and would
        # leave us with no working credential. Pinning to the resolved
        # main runtime guarantees we hit the same endpoint the user
        # already authenticated for chat.
        try:
            runtime = resolve_runtime_provider()
        except Exception as err:
            raise RuntimeError(f"could not resolve hermes main runtime: {err}") from err

        main_runtime: dict[str, str] = {}
        for field in ("provider", "model", "base_url", "api_key", "api_mode"):
            value = runtime.get(field) if isinstance(runtime, dict) else None
            if isinstance(value, str) and value.strip():
                main_runtime[field] = value.strip()

        normalized = [
            {
                "role": str(m.get("role", "user")),
                "content": str(m.get("content", "")),
            }
            for m in messages
            if isinstance(m, dict)
        ]
        timeout_ms = params.get("timeoutMs")
        timeout_s: float | None = None
        if isinstance(timeout_ms, int | float) and timeout_ms > 0:
            timeout_s = float(timeout_ms) / 1000.0

        max_tokens = params.get("maxTokens")
        temperature = params.get("temperature")

        kwargs: dict[str, Any] = {
            "messages": normalized,
            # `main_runtime` makes `_resolve_auto` prefer the user's
            # main conversation provider + model over the generic auto
            # chain. If the user's main provider is also down,
            # `call_llm` raises — which is exactly the "agent's own
            # model is broken too, stop falling back" semantic we want
            # (red light on Overview).
            "main_runtime": main_runtime,
        }
        if isinstance(max_tokens, int | float) and max_tokens > 0:
            kwargs["max_tokens"] = int(max_tokens)
        if isinstance(temperature, int | float):
            kwargs["temperature"] = float(temperature)
        if timeout_s is not None:
            kwargs["timeout"] = timeout_s

        started = time.time()
        try:
            response = call_llm(**kwargs)
        except Exception as err:
            # Surface the original failure verbatim — the LlmClient
            # will tag this as a "host fallback failed" terminal error
            # and the Overview red-light path takes over.
            raise RuntimeError(f"host LLM call failed: {err}") from err

        # `call_llm` returns an OpenAI ChatCompletion-shaped object.
        # Pluck the assistant text + token usage defensively so a
        # non-standard host (e.g. Anthropic native) still produces a
        # populated response.
        text = ""
        model = ""
        usage_dict: dict[str, int] = {}
        try:
            choices = getattr(response, "choices", None) or response.get("choices", [])  # type: ignore[union-attr]
            if choices:
                first = choices[0]
                msg = getattr(first, "message", None) or first.get("message", {})  # type: ignore[union-attr]
                content = getattr(msg, "content", None) or msg.get("content", "")  # type: ignore[union-attr]
                text = str(content or "")
            model = (
                getattr(response, "model", None)
                or response.get("model", "")  # type: ignore[union-attr]
                or ""
            )
            u = getattr(response, "usage", None) or response.get("usage", None)  # type: ignore[union-attr]
            if u is not None:
                pt = getattr(u, "prompt_tokens", None)
                ct = getattr(u, "completion_tokens", None)
                tt = getattr(u, "total_tokens", None)
                if pt is None and isinstance(u, dict):
                    pt = u.get("prompt_tokens")
                    ct = u.get("completion_tokens")
                    tt = u.get("total_tokens")
                if isinstance(pt, int):
                    usage_dict["promptTokens"] = pt
                if isinstance(ct, int):
                    usage_dict["completionTokens"] = ct
                if isinstance(tt, int):
                    usage_dict["totalTokens"] = tt
        except Exception:
            logger.debug("host.llm.complete: shape parse failed", exc_info=True)

        result: dict[str, Any] = {
            "text": text,
            "model": str(model or ""),
            "durationMs": int((time.time() - started) * 1000),
        }
        if usage_dict:
            result["usage"] = usage_dict
        return result

    # ─── Internals ────────────────────────────────────────────────────────

    def _host_runtime_context(self) -> dict[str, str]:
        """Best-effort snapshot of Hermes' main conversation runtime."""
        try:
            from hermes_cli.runtime_provider import (  # type: ignore[import-not-found]
                resolve_runtime_provider,
            )

            runtime = resolve_runtime_provider()
        except Exception:
            return dict(self._last_host_runtime)

        out: dict[str, str] = {}
        if isinstance(runtime, dict):
            for source, target in (
                ("provider", "hostProvider"),
                ("model", "hostModel"),
                ("api_mode", "hostApiMode"),
                ("base_url", "hostBaseUrl"),
            ):
                value = runtime.get(source)
                if isinstance(value, str) and value.strip():
                    out[target] = value.strip()
        if out:
            self._last_host_runtime = dict(out)
        return out

    def _bridge_request(
        self,
        method: str,
        params: Any = None,
        *,
        timeout: float | None = None,
        ensure_session: bool = True,
        deadline_monotonic: float | None = None,
    ) -> dict[str, Any]:
        bridge = self._bridge
        if bridge is None:
            raise BridgeError("transport_closed", "bridge is not connected")
        if ensure_session and isinstance(bridge, SharedBridgeLease):
            generation = bridge.generation
            if generation != self._bridge_generation:
                with self._session_open_lock:
                    if bridge.generation != self._bridge_generation:
                        logger.info(
                            "MemOS: reopening logical session after shared bridge generation "
                            "change runtime=%s old_generation=%d new_generation=%d session=%s",
                            bridge.runtime_id,
                            self._bridge_generation,
                            bridge.generation,
                            self._session_id,
                        )
                        session_ceiling = (
                            timeout
                            if deadline_monotonic is not None and timeout is not None
                            else 30.0
                        )
                        session_timeout = _remaining_rpc_timeout(
                            deadline_monotonic,
                            session_ceiling,
                        )
                        self._open_session(
                            self._session_id,
                            timeout=session_timeout or 30.0,
                        )
        request_timeout = _remaining_rpc_timeout(deadline_monotonic, timeout)
        if request_timeout is None:
            return bridge.request(method, params)
        return bridge.request(method, params, timeout=request_timeout)

    def _open_session(self, session_id: str = "", *, timeout: float = 30.0) -> None:
        bridge = self._bridge
        assert bridge is not None
        requested_session = session_id or self._session_id or ""
        host_runtime = self._host_runtime_context()
        resp = bridge.request(
            "session.open",
            {
                "agent": "hermes",
                "sessionId": requested_session,
                "namespace": self._runtime_namespace(),
                "meta": {
                    "hermesHome": self._hermes_home,
                    "platform": self._platform,
                    "agentIdentity": self._agent_identity,
                    "profileId": self._runtime_namespace()["profileId"],
                    "namespace": self._runtime_namespace(),
                    **host_runtime,
                },
            },
            timeout=timeout,
        )
        self._session_id = resp.get("sessionId") or requested_session
        if isinstance(bridge, SharedBridgeLease):
            self._bridge_generation = bridge.generation
        else:
            self._bridge_generation += 1

    def _bridge_request_with_retry(
        self,
        method: str,
        params: Any,
        *,
        timeout: float | None = None,
        deadline_monotonic: float | None = None,
    ) -> dict[str, Any]:
        """Read-path helper: reconnect + retry once on ``transport_closed``.

        Layer 4 (#2028): the read-path memory tools previously issued a
        single ``self._bridge.request(...)`` and surfaced any error
        verbatim to the model. When the Node bridge has died since the
        last user turn, that first call now raises
        ``transport_closed`` fast (thanks to Layer 1/2). This helper
        mirrors the pattern ``sync_turn`` already uses: reconnect the
        bridge once and re-issue the same request. A second failure is
        left to propagate — the ``except`` block in
        ``handle_tool_call`` will surface the error text verbatim.
        """
        assert self._bridge is not None
        try:
            return self._bridge_request(
                method,
                params,
                timeout=timeout,
                deadline_monotonic=deadline_monotonic,
            )
        except BridgeError as err:
            if not self._is_transport_closed(err):
                raise
            logger.info(
                "MemOS: bridge transport closed on %s; reconnecting and retrying once — %s",
                method,
                err,
            )
            reconnect_ceiling = (
                timeout if deadline_monotonic is not None and timeout is not None else 30.0
            )
            reconnect_timeout = _remaining_rpc_timeout(
                deadline_monotonic,
                reconnect_ceiling,
            )
            self._reconnect_bridge(
                self._session_id,
                timeout=reconnect_timeout or 30.0,
            )
            assert self._bridge is not None
            return self._bridge_request(
                method,
                params,
                timeout=timeout,
                deadline_monotonic=deadline_monotonic,
            )

    def _is_transport_closed(self, err: Exception) -> bool:
        if isinstance(err, BridgeError) and err.code == "transport_closed":
            return True
        msg = str(err).lower()
        return "broken pipe" in msg or "bridge closed" in msg or "transport_closed" in msg

    def _should_reconnect_after_keepalive_failure(self, err: Exception) -> bool:
        """Decide whether a keepalive failure warrants a bridge reconnect.

        Layer 3 (#2028): the keepalive previously reconnected only on
        ``BridgeError("transport_closed", …)``. A hung Node bridge
        surfaces instead as ``BridgeError("timeout", …)`` (the client
        gave up waiting for a response); that error was dropped at
        DEBUG and the stale client kept being reused, so every
        subsequent memory tool timed out for another 30 s. Reconnect
        also when the subprocess has already exited (belt-and-braces
        for hangs that didn't raise a transport error).

        A live subprocess raising a generic (non-transport) error must
        NOT trigger a reconnect — otherwise transient parse noise
        would create a reconnect storm.
        """
        if self._is_transport_closed(err):
            return True
        if isinstance(err, BridgeError) and err.code == "timeout":
            return True
        # Ask the underlying subprocess: is it still alive?
        bridge = self._bridge
        if bridge is not None:
            try:
                exit_code = bridge._proc.poll()  # type: ignore[attr-defined]
            except Exception:
                exit_code = None
            if exit_code is not None:
                return True
        return False

    def _reconnect_bridge(self, session_id: str = "", *, timeout: float = 30.0) -> None:
        # Don't reconnect if we're shutting down
        if self._bridge_keepalive_stop.is_set():
            logger.debug("MemOS: skipping reconnect during shutdown")
            return

        with self._reconnect_lock:
            # Double-check after acquiring lock
            if self._bridge_keepalive_stop.is_set():
                logger.debug("MemOS: skipping reconnect during shutdown (after lock)")
                return

            if self._shared_bridge:
                bridge = self._bridge
                acquired_here = bridge is None
                if bridge is None:
                    if self._runtime_home is None:
                        self._runtime_home = _resolved_memos_runtime_home()
                        self._runtime_env = _memos_runtime_env_snapshot(self._runtime_home)
                    runtime_home = self._runtime_home
                    runtime_env = dict(self._runtime_env)
                    bridge = SHARED_BRIDGE_REGISTRY.acquire(
                        _shared_bridge_runtime_key(runtime_home),
                        client_factory=lambda home=str(runtime_home), env=runtime_env: (
                            MemosBridgeClient(
                                runtime_home=home,
                                extra_env=env,
                            )
                        ),
                        before_spawn=lambda home=runtime_home: _prepare_shared_bridge(home),
                        host_handlers={
                            "host.llm.complete": self._handle_host_llm_complete,
                        },
                    )
                    self._bridge = bridge
                if not isinstance(bridge, SharedBridgeLease):
                    raise BridgeError(
                        "internal",
                        "shared bridge mode has a non-shared bridge handle",
                    )
                try:
                    if acquired_here:
                        # acquire() already ensures and health-checks the shared
                        # client. Open this logical session directly; restarting
                        # here would disrupt every healthy provider using it.
                        acquired_generation = bridge.generation
                        try:
                            self._open_session(session_id, timeout=timeout)
                        except Exception as err:
                            if not self._is_transport_closed(err):
                                raise
                            bridge.reconnect(expected_generation=acquired_generation)
                            self._open_session(session_id, timeout=timeout)
                    else:
                        expected_generation = self._bridge_generation or bridge.generation
                        bridge.reconnect(expected_generation=expected_generation)
                        self._open_session(session_id, timeout=timeout)
                except Exception:
                    if acquired_here:
                        with contextlib.suppress(Exception):
                            bridge.close()
                        if self._bridge is bridge:
                            self._bridge = None
                        self._bridge_generation = 0
                    raise
                logger.info(
                    "MemOS: shared bridge session recovered runtime=%s generation=%d "
                    "pid=%s session=%s",
                    bridge.runtime_id,
                    bridge.generation,
                    bridge.pid,
                    self._session_id,
                )
                return

            old_bridge = self._bridge
            old_pid = getattr(old_bridge, "pid", None) if old_bridge else None

            if old_bridge:
                logger.info("MemOS: closing old bridge (pid=%s)", old_pid)
                with contextlib.suppress(Exception):
                    old_bridge.close()
                logger.info("MemOS: old bridge closed (pid=%s)", old_pid)

            runtime_home = self._runtime_home or _resolved_memos_runtime_home()
            _prepare_shared_bridge(runtime_home, cleanup_legacy_zombies=True)
            new_bridge: MemosBridgeClient | None = None
            try:
                runtime_home = self._runtime_home or _resolved_memos_runtime_home()
                runtime_env = dict(self._runtime_env or _memos_runtime_env_snapshot(runtime_home))
                new_bridge = MemosBridgeClient(
                    runtime_home=str(runtime_home),
                    extra_env=runtime_env,
                )
                logger.info(
                    "MemOS: new bridge created (pid=%s)",
                    getattr(new_bridge, "pid", "?"),
                )

                new_bridge.register_host_handler(
                    "host.llm.complete",
                    self._handle_host_llm_complete,
                )
                self._bridge = new_bridge
                self._open_session(session_id, timeout=timeout)
            except Exception:
                if new_bridge is not None:
                    with contextlib.suppress(Exception):
                        new_bridge.close()
                if self._bridge is new_bridge:
                    self._bridge = None
                raise

    def _ensure_bridge(self, session_id: str = "", *, timeout: float = 30.0) -> bool:
        if self._bridge:
            return True
        try:
            self._reconnect_bridge(session_id or self._session_id, timeout=timeout)
            logger.info(
                "MemOS: bridge reconnected session=%s platform=%s",
                self._session_id,
                self._platform,
            )
            return True
        except Exception as err:
            logger.warning("MemOS: bridge reconnect failed — %s", err)
            return False

    def _start_bridge_keepalive(self) -> None:
        if self._shared_bridge:
            # SharedBridgeRuntime owns the only keepalive/reconnect loop.
            return
        if self._bridge_keepalive_thread and self._bridge_keepalive_thread.is_alive():
            return
        self._bridge_keepalive_stop.clear()

        _self_ref = weakref.ref(self)

        def _run() -> None:
            while True:
                # Stop signal set (e.g. shutdown called by another thread).
                # When self is garbage-collected the weakref resolves to None
                # and we exit gracefully instead of keeping the thread + bridge
                # subprocess alive forever.
                provider = _self_ref()
                if provider is None:
                    break
                if provider._bridge_keepalive_stop.wait(5.0):
                    break
                if not provider._ensure_bridge(provider._session_id, timeout=10.0):
                    continue
                try:
                    assert provider._bridge is not None
                    provider._bridge.request("core.health", {}, timeout=10.0)
                except Exception as err:
                    if provider._should_reconnect_after_keepalive_failure(err):
                        logger.info(
                            "MemOS: bridge keepalive reconnecting after failure — %s",
                            err,
                        )
                        with contextlib.suppress(Exception):
                            provider._reconnect_bridge(provider._session_id, timeout=10.0)
                    else:
                        logger.debug("MemOS: bridge keepalive failed — %s", err)

        self._bridge_keepalive_thread = threading.Thread(
            target=_run,
            daemon=True,
            name="memos-bridge-keepalive",
        )
        self._bridge_keepalive_thread.start()

    def _turn_start(
        self,
        query: str,
        *,
        session_id: str = "",
        deadline_monotonic: float | None = None,
        deadline_at_ms: int | None = None,
    ) -> str:
        assert self._bridge is not None
        host_runtime = self._host_runtime_context()
        with self._state_lock:
            turn_key = self._active_turn_key
            visible_context_start_ts = self._visible_context_start_ts
        context_hints: dict[str, Any] = {
            "agentIdentity": self._agent_identity,
            "namespace": self._runtime_namespace(),
            **host_runtime,
        }
        if visible_context_start_ts is not None:
            context_hints.update(
                {
                    "visibleContextKnown": True,
                    "visibleContextStartTs": visible_context_start_ts,
                }
            )
        now_ms = int(time.time() * 1000)
        if deadline_monotonic is None:
            deadline_monotonic = time.monotonic() + _PREFETCH_RPC_TIMEOUT
        if deadline_at_ms is None:
            core_budget_seconds = max(
                0.05,
                _PREFETCH_RPC_TIMEOUT - _PREFETCH_RESPONSE_RESERVE_SECONDS,
            )
            deadline_at_ms = now_ms + int(core_budget_seconds * 1000)
        payload: dict[str, Any] = {
            "agent": "hermes",
            "namespace": self._runtime_namespace(),
            "sessionId": session_id or self._session_id,
            "userText": query,
            "contextHints": context_hints,
            "ts": now_ms,
            "deadlineAt": deadline_at_ms,
        }
        if turn_key:
            payload["turnKey"] = turn_key
        resp = self._bridge_request_with_retry(
            "turn.start",
            payload,
            timeout=_remaining_rpc_timeout(
                deadline_monotonic,
                _PREFETCH_RPC_TIMEOUT,
            ),
            deadline_monotonic=deadline_monotonic,
        )
        response_query = (resp or {}).get("query") or {}
        response_session = str(response_query.get("sessionId") or "")
        requested_session = str(payload["sessionId"])
        if response_session and response_session != requested_session:
            raise BridgeError(
                "session_mismatch",
                "turn.start returned a different session "
                f"(requested={requested_session}, returned={response_session})",
            )
        # Stash the real episode id the pipeline auto-created (V7
        # §0.1 may have boundary-cut the previous episode and started
        # a new one). `on_session_end` uses it to close the right
        # episode — see the "Episode tracking" comment block above.
        new_eid = response_query.get("episodeId") or ""
        if new_eid and new_eid != self._episode_id:
            self._episode_id = new_eid
            logger.debug("MemOS: stashed episode %s from turn.start", new_eid)
        context = (resp or {}).get("injectedContext") or ""
        if not context:
            return ""
        return f"## Recalled Memories\n{context}"

    def _turn_end(
        self,
        user_content: str,
        assistant_content: str,
        tool_calls: list[dict[str, Any]],
        ts_ms: int,
        *,
        agent_thinking: str = "",
    ) -> str:
        if not self._bridge:
            return ""
        # Strip private book-keeping fields before sending.
        clean_tool_calls = [
            {k: v for k, v in tc.items() if k not in {"_id", "_ids"}} for tc in tool_calls
        ]
        payload: dict[str, Any] = {
            "agent": "hermes",
            "namespace": self._runtime_namespace(),
            "sessionId": self._session_id,
            "episodeId": self._episode_id,
            "agentText": assistant_content,
            "userText": user_content,
            "toolCalls": clean_tool_calls,
            "contextHints": {
                "agentIdentity": self._agent_identity,
                "namespace": self._runtime_namespace(),
                **self._host_runtime_context(),
            },
            "ts": ts_ms,
        }
        if agent_thinking:
            payload["agentThinking"] = agent_thinking
        result = self._bridge_request("turn.end", payload, timeout=_LONG_RPC_TIMEOUT)
        # Capture the trace ID for feedback submission
        if result and isinstance(result, dict):
            trace_ids = result.get("traceIds", [])
            if trace_ids and len(trace_ids) > 0:
                trace_id = trace_ids[-1]  # Last trace is the current turn
                self._last_trace_id = trace_id
                return trace_id
        return ""

    def _try_submit_verifier_feedback(
        self,
        user_content: str,
        assistant_content: str,
        ts_ms: int,
        *,
        trace_id: str = "",
        fallback: bool = False,
    ) -> bool:
        try:
            submitted = self._submit_verifier_feedback(
                user_content,
                assistant_content,
                ts_ms,
                trace_id=trace_id,
            )
            if submitted and fallback:
                logger.info("MemOS: submitted verifier feedback without trace binding")
            return submitted
        except Exception as err:
            logger.warning("MemOS: verifier feedback submit failed — %s", err)
            return False

    def _submit_verifier_feedback(
        self,
        user_content: str,
        assistant_content: str,
        ts_ms: int,
        *,
        trace_id: str = "",
    ) -> bool:
        if not self._bridge or not self._episode_id:
            return False
        polarity = _feedback_polarity(user_content)
        magnitude = _feedback_magnitude(user_content, polarity)
        raw = {
            "source": "hermes.verifier_feedback",
            "userText": user_content,
            "assistantText": assistant_content,
            "polarity": polarity,
        }
        payload: dict[str, Any] = {
            "episodeId": self._episode_id,
            "channel": "explicit",
            "polarity": polarity,
            "magnitude": magnitude,
            "rationale": user_content,
            "raw": raw,
            "ts": ts_ms,
        }
        if trace_id:
            payload["traceId"] = trace_id
        self._bridge_request("feedback.submit", payload, timeout=75.0)
        return True


# ─── Discovery entry points ───────────────────────────────────────────────


# Pattern 1: `register(ctx)` — preferred by `plugins/memory/__init__.py`.
def register(ctx: Any) -> None:
    """hermes-agent plugin entry point."""
    ctx.register_memory_provider(MemTensorProvider())


# Pattern 2: exported class — fallback via `issubclass(MemoryProvider)`.
__all__ = ["PLUGIN_ID", "PLUGIN_VERSION", "MemTensorProvider", "register"]