"""
HermesAdapter — connects to a local Hermes gateway instance.

Hermes (github.com/NousResearch/hermes-agent, MIT) is a self-improving AI agent
framework by Nous Research. This adapter uses the Runs API, which is the right
integration point — it gives us structured SSE events, approval flow, and stop.

API (verified from source, gateway/platforms/api_server.py, Feb 2026):
  Port:    8642 default (API_SERVER_PORT env)
  Auth:    Authorization: Bearer ***  (HERMES_API_KEY env, optional)
  Session: X-Hermes-Session-Id header → persistent memory across runs

  POST /v1/runs
    body: { input, conversation_history?, model?, stream? }
    → 202 { run_id, status: "queued" }

  GET /v1/runs/{run_id}/events        ← SSE, this is what we stream
    event: message.delta              → TextBlock (streaming text chunk)
    event: reasoning.available        → ThinkingBlock
    event: tool.started               → ToolCallBlock(status="running")
    event: tool.completed             → ToolCallBlock(status="done"|"error")
    event: approval.request           → PermissionRequestBlock (run pauses here)
    event: run.completed              → SessionEndBlock(reason="completed")
    event: run.failed                 → ErrorBlock + SessionEndBlock(reason="error")
    event: run.cancelled              → SessionEndBlock(reason="user_stopped")

  POST /v1/runs/{run_id}/approval
    body: { choice: "once"|"session"|"always"|"deny" }
    → called from respond_permission()

  POST /v1/runs/{run_id}/stop         ← called from stop()

  GET /v1/capabilities                → machine-readable API surface

What Hermes provides through Beacon:
  - Messaging proxy: Telegram, Discord, Slack, WhatsApp, Signal, Matrix (19+ platforms)
    configured via `hermes gateway setup` + `hermes gateway start`
  - MCP aggregator: auto-discovers all installed MCP servers, merges tool lists
  - Persistent memory: full-text searchable history across sessions
  - Subagent spawning: parallel sub-agents for multi-workstream tasks
  - Any LLM: routes to OpenAI, Anthropic, OpenRouter, local Ollama, etc.

Optional `model` parameter: when provided (e.g. "deepseek/deepseek-chat"),
it is forwarded to Hermes as the model override for that session. When omitted,
Hermes uses its configured default model.
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import re
import shutil
import uuid
from pathlib import Path
from typing import AsyncIterator

log = logging.getLogger("beacon.hermes")

import httpx

from ..attachments import Attachment, build_openai_content
from ..blocks import (
    Block, AgentInfoBlock, TextBlock, ToolCallBlock, ToolResultBlock,
    ThinkingBlock, AudioHintBlock, SessionEndBlock, ErrorBlock,
    PermissionRequestBlock, ProgressBlock, SessionStatusHintBlock,
)
from .base import (BaseAdapter, DetectedAgent, ExternalSession, split_markdown,
                   tool_label_subject, context_window_for_model)
from ..adapter_logging import setup_adapter_logger

_HERMES_SLASH_COMMANDS = ["/memory", "/status", "/usage"]

_DEFAULT_PORTS = [8643, 8642]


def find_hermes_port() -> int | None:
    """Check for Beacon Gateway (8643) first, then native (8642)."""
    for port in _DEFAULT_PORTS:
        try:
            # Check gateway /status or native /health
            path = "/status" if port == 8643 else "/health"
            resp = httpx.get(f"http://localhost:{port}{path}", timeout=1)
            if resp.status_code == 200:
                return port
        except Exception:
            continue
    return None


_HERMES_DB = Path.home() / ".hermes" / "state.db"
_hermes_run_to_session_cache: dict[str, tuple[str, float]] = {}  # run_id -> (session_id, timestamp)
_CACHE_TTL_SECS = 300  # Cache lookups for 5 minutes

# Internal "max iterations reached" message injected by Hermes — not a real prompt.
_ITERATION_MSG_RE = re.compile(r"^You.ve reached the maximum", re.IGNORECASE)


def _scan_hermes_sessions(limit: int = 20) -> list[ExternalSession]:
    """Read recent sessions from ~/.hermes/state.db.

    CWD comes from assistant messages containing 'Work in directory:'. If that hint
    is missing, fall back to the Beacon adapter logs under
    ~/Library/Logs/Terminull/adapters/hermes/*.log on macOS or
    ~/.local/state/terminull/logs/adapters/hermes/*.log on Linux.
    """
    import sqlite3
    if not _HERMES_DB.exists():
        return []
    try:
        con = sqlite3.connect(f"file:{_HERMES_DB}?mode=ro", uri=True,
                              check_same_thread=False)
        con.row_factory = sqlite3.Row
    except Exception as e:
        log.debug("hermes scan: cannot open DB: %s", e)
        return []
    try:
        rows = con.execute("""
            SELECT s.id,
                   strftime('%Y-%m-%dT%H:%M:%SZ', s.started_at, 'unixepoch') AS started_at,
                   (
                       SELECT MAX(timestamp) FROM messages WHERE session_id = s.id
                   ) AS last_msg_ts,
                   (
                       SELECT substr(content, 1, 200)
                       FROM   messages
                       WHERE  session_id = s.id
                         AND  role = 'user'
                         AND  content NOT LIKE 'You%ve reached the maximum%'
                       ORDER BY timestamp ASC
                       LIMIT 1
                   ) AS first_prompt,
                   (
                       SELECT content
                       FROM   messages
                       WHERE  session_id = s.id
                         AND  role = 'assistant'
                         AND  content LIKE '%Work in directory:%'
                       ORDER BY timestamp ASC
                       LIMIT 1
                   ) AS cwd_hint
            FROM   sessions s
            WHERE  s.message_count > 0
              AND  EXISTS (
                       SELECT 1 FROM messages
                       WHERE  session_id = s.id
                         AND  role = 'user'
                         AND  content NOT LIKE 'You%ve reached the maximum%'
                   )
            ORDER BY s.started_at DESC
            LIMIT ?
        """, (limit,)).fetchall()
    except Exception as e:
        log.debug("hermes scan: query failed: %s", e)
        return []
    finally:
        con.close()

    import time as _t
    now = _t.time()
    hermes_running = find_hermes_port() is not None
    result: list[ExternalSession] = []
    for row in rows:
        first_prompt = row["first_prompt"] or ""
        if not first_prompt:
            continue
        # is_active is per-SESSION, not "the daemon is up": only a session whose
        # newest message landed within the active window is mid-turn. The blanket
        # `is_active=hermes_running` marked every historical session active, so the
        # external-session cap (keep all live + N idle) never trimmed them.
        last_ts = row["last_msg_ts"] or 0
        is_active = hermes_running and (now - last_ts) < _HERMES_ACTIVE_WINDOW_SECS
        cwd = ""
        hint = row["cwd_hint"] or ""
        import re as _re
        m = _re.search(r"Work in directory: ([^\n\r]+)", hint)
        if m:
            cwd = m.group(1).strip()
        # Fallback: try to read from adapter log if DB didn't have a CWD hint.
        if not cwd:
            cwd = _read_adapter_log_cwd(row["id"])

        result.append(ExternalSession(
            session_id=row["id"],
            agent_id="hermes",
            cwd=cwd,
            last_prompt=first_prompt,
            started_at=row["started_at"] or "",
            is_active=is_active,
        ))
    return result


# A Hermes session whose newest message landed within this window is treated as an
# active turn (the daemon is mid-run for it) — Beacon parks rather than co-driving.
_HERMES_ACTIVE_WINDOW_SECS = 25


def _adapter_log_dir() -> Path | None:
    """Return the adapter log directory on macOS or Linux."""
    import sys
    if sys.platform == "darwin":
        d = Path.home() / "Library" / "Logs" / "Beacon" / "adapters" / "hermes"
        return d if d.is_dir() else None
    xdg = Path.home() / ".local" / "state" / "terminull" / "logs" / "adapters" / "hermes"
    return xdg if xdg.is_dir() else None


def _read_adapter_log_cwd(sid: str) -> str:
    """Extract cwd from Hermes adapter log for sid if available."""
    log_dir = _adapter_log_dir()
    if not log_dir:
        return ""
    # Hermes adapters log with Beacon session_id, not Hermes sid; scan for any .log
    # where an entry contains a CWD and a matching session identifier.
    # Heuristic: logs line format includes 'Session: {beacon_sid}' and 'cwd: /path'.
    # For now, this is a best-effort fallback; exact mapping requires richer log keys.
    # We return the first cwd we find as a reasonable hint.
    for log_path in log_dir.glob("*.log"):
        try:
            lines = log_path.read_text(encoding="utf-8", errors="ignore").splitlines()
            for line in lines:
                if "cwd:" in line.lower():
                    parts = line.split("cwd:")
                    if len(parts) == 2:
                        cwd = parts[1].strip()
                        if cwd and (cwd.startswith("/") or cwd.startswith("~")):
                            return cwd
        except Exception:
            continue
    return ""


def _hermes_session_active(sid: str) -> bool:
    """True if the Hermes session `sid` got a message within the active window."""
    import sqlite3, time as _t
    if not _HERMES_DB.exists():
        return False
    try:
        con = sqlite3.connect(f"file:{_HERMES_DB}?mode=ro", uri=True, check_same_thread=False)
    except Exception:
        return False
    try:
        row = con.execute(
            "SELECT MAX(timestamp) FROM messages WHERE session_id = ?", (sid,)).fetchone()
    except Exception:
        return False
    finally:
        con.close()
    last = (row[0] if row else 0) or 0
    return (_t.time() - last) < _HERMES_ACTIVE_WINDOW_SECS


def _hermes_session_id_from_run(run_id: str) -> str | None:
    """Resolve a Hermes session_id from a run_id using the messages table.
    
    The messages table contains the raw JSON of events, including the run_id.
    This provides a fallback path to map run_id → session_id for usage tracking
    when the Hermes gateway does not expose it in /status.
    
    Uses an in-memory cache to avoid repeated slow DB queries.
    """
    import sqlite3
    import time as _t
    
    # Check cache first
    cached = _hermes_run_to_session_cache.get(run_id)
    if cached:
        sid, ts = cached
        if (_t.time() - ts) < _CACHE_TTL_SECS:
            log.debug("_hermes_session_id_from_run: Cache hit for run_id=%s -> session_id=%s", run_id, sid)
            return sid
        else:
            # Expired cache entry
            del _hermes_run_to_session_cache[run_id]
    
    # Periodically cleanup cache to prevent unbounded growth
    if len(_hermes_run_to_session_cache) > _MAX_CACHE_ENTRIES:
        _cleanup_hermes_cache()
    
    if not _HERMES_DB.exists():
        return None
    try:
        con = sqlite3.connect(f"file:{_HERMES_DB}?mode=ro", uri=True, check_same_thread=False)
    except Exception:
        return None
    try:
        # The messages content column contains JSON with tool_call events.
        # We search for entries that mention the run_id in a tool_call or similar context.
        # A more efficient approach would be if Hermes indexed run_id, but we work with
        # the current schema.
        # We look for a pattern like '"run_id": "xyz"' in the content.
        row = con.execute(
            "SELECT session_id FROM messages WHERE content LIKE ? LIMIT 1",
            (f'%"run_id": "{run_id}"%',)
        ).fetchone()
        if row:
            sid = row[0]
            # Cache the result
            _hermes_run_to_session_cache[run_id] = (sid, _t.time())
            log.debug("_hermes_session_id_from_run: DB lookup + cache for run_id=%s -> session_id=%s", run_id, sid)
            return sid
        return None
    except Exception as e:
        log.warning("_hermes_session_id_from_run: DB lookup failed for run_id=%s: %s", run_id, e)
        return None
    finally:
        con.close()


def _delegate_goals(args: dict) -> list[str]:
    """Extract subagent goals from a delegate_task tool_call's args.

    Two observed shapes: a single subagent `{"goal": "…"}`, or parallel
    subagents `{"tasks": [{"goal": "…"}, …]}`. Returns the goals in order.
    """
    tasks = args.get("tasks")
    if isinstance(tasks, list):
        return [str(t.get("goal", "")) for t in tasks if isinstance(t, dict) and t.get("goal")]
    goal = args.get("goal")
    return [str(goal)] if goal else []


def _validate_session_id_sync(current_key: str, new_key: str, context: str) -> bool:
    """Validate that syncing from current_key to new_key is safe.
    
    Returns False if:
    - Both keys are set and different (indicates possible cross-wire)
    - new_key is empty
    - current_key is already the same as new_key (no-op)
    """
    if not new_key:
        log.debug("_validate_session_id_sync: Rejected empty new_key in context=%s", context)
        return False
    if current_key == new_key:
        # No change needed, but it's not an error
        return True
    if current_key and current_key != new_key:
        # Both are set and different - could be cross-wire. 
        # Log a warning but allow it (gateway might have corrected it)
        log.warning("_validate_session_id_sync: Key change in context=%s: %s -> %s (potential cross-wire, proceeding)", 
                   context, current_key, new_key)
    return True


class HermesAdapter(BaseAdapter):
    cap_images = True
    cap_documents = True
    cap_sentinel = True
    cap_subagents = True
    cap_thinking = True
    cap_permissions = True
    cap_subagents = True     # Hermes has subagents; mapped via delegate_task
    cap_interrupt = True     # /v1/runs/{id}/stop is supported
    provides_models = True
    provides_usage = True
    agent_id = "hermes"
    display_name = "Hermes"
    beta = False              # live-verified (in daily use); no Beta badge

    @classmethod
    def detect(cls) -> "DetectedAgent | None":
        # Probe the beacon gateway (8643), then Hermes native (8642).
        gateway_url, api_url = "http://localhost:8643", "http://localhost:8642"
        available, base_url = False, gateway_url
        try:
            if httpx.get(f"{gateway_url}/status", timeout=1).status_code == 200:
                available, base_url = True, gateway_url
        except Exception:
            try:
                if httpx.get(f"{api_url}/health", timeout=1).status_code == 200:
                    available, base_url = True, api_url
            except Exception:
                pass
        binary = shutil.which("hermes")
        if not binary and not available:
            return None
        return DetectedAgent(id="hermes", name="Hermes", binary=binary,
                             extra={"base_url": base_url}, available=available)

    def __init__(
        self,
        base_url: str = "http://localhost:8642",
        session_key: str | None = None,
        cwd: str | None = None,
        model: str | None = None,
        resume_session_id: str | None = None,
    ) -> None:
        self._stop_requested = False  # Flag to break SSE loop when stop() is called
        self._base_url = base_url.rstrip("/")
        self._session_key = session_key          # X-Hermes-Session-Id
        self._resume_session_id = resume_session_id  # pre-existing Hermes session to restore
        self._cwd = cwd
        self._model = model                      # optional model override for /v1/runs
        self._api_key = os.environ.get("HERMES_API_KEY")
        self._history: list[dict] = []
        self._fetch_store: dict[str, str] = {}
        self._send_queue: asyncio.Queue[str] = asyncio.Queue()
        self._tools: list[dict] = []
        self._current_run_id: str | None = None
        self._last_known_run_id: str | None = None  # Persisted across turns for reliable usage/lookup
        self._pending_tool_ids: dict[str, str] = {}  # "{tool_name}:{call_count}" → block id
        self._pending_tool_meta: dict[str, tuple[str, str, dict]] = {}  # tc_id → (label, subject, args)
        self._tool_call_counts: dict[str, int] = {}  # tool_name → invocation count (dedup same-name calls)
        self._subagent_goals: dict[str, list[str]] = {}  # delegate_task tc_id → [goal, …]
        
        # Adapter-specific per-session log
        self._adapter_logger: logging.Logger | None = None
        self._log = self._log_debug = self._log_warn = self._log_err = lambda _: None

    @classmethod
    def scan_external_sessions(cls) -> list[ExternalSession]:
        return _scan_hermes_sessions()

    @classmethod
    def read_session_condensed(cls, ext_sid: str, max_turns: int = 20) -> str:
        """Return last `max_turns` user+assistant exchanges from hermes state.db as condensed text."""
        import sqlite3 as _sqlite3
        if not _HERMES_DB.exists():
            return ""
        try:
            con = _sqlite3.connect(f"file:{_HERMES_DB}?mode=ro", uri=True, check_same_thread=False)
            con.row_factory = _sqlite3.Row
            rows = con.execute(
                "SELECT role, content FROM messages WHERE session_id=? ORDER BY timestamp",
                (ext_sid,)
            ).fetchall()
            con.close()
        except Exception:
            return ""
        turns = [f"{r['role']}: {(r['content'] or '')[:400]}" for r in rows
                 if r['role'] in ('user', 'assistant') and r['content']]
        return "\n\n".join(turns[-(max_turns * 2):])

    @classmethod
    def has_changes_since(cls, ts: float) -> bool:
        """True if hermes state.db has been modified since `ts`."""
        try:
            return _HERMES_DB.exists() and _HERMES_DB.stat().st_mtime > ts
        except OSError:
            return False

    @classmethod
    def is_cwd_externally_owned(cls, cwd: str, sid: str = "") -> bool:
        # Hermes is a shared daemon, not cwd-scoped → key on the session id. "Owned"
        # = an active turn: the daemon is up AND the session got a message very
        # recently. (ended_at is unreliable — mostly NULL — so use recency.) This is
        # the "almost handoff" contention signal. See docs/session-handoff.md.
        if not sid or find_hermes_port() is None:
            return False
        return _hermes_session_active(sid)

    @classmethod
    def make_transcript_mirror(cls, cwd: str, sid: str, *, from_end: bool = False):
        from ..transcript_mirror import make_hermes_mirror
        return make_hermes_mirror(sid, from_end=from_end)

    @classmethod
    def terminal_resume_command(cls, sid: str) -> str | None:
        # Hermes ships a CLI that resumes a session interactively (`hermes --resume <id>`),
        # so "Continue in Terminal" works even though Hermes is daemon-backed (no cwd).
        import shlex
        return f"hermes --resume {shlex.quote(sid)}" if sid else "hermes"

    @classmethod
    async def list_models(cls, extra: dict | None = None) -> list[dict]:
        import httpx
        base_url = (extra or {}).get("base_url", "http://localhost:8643")
        try:
            async with httpx.AsyncClient() as client:
                resp = await client.get(f"{base_url}/available_models", timeout=8)
            if resp.status_code == 200:
                models = resp.json().get("models", [])
                if models:  # Only return if models were actually found
                    return models
            # If API call failed OR no models were returned, fall back to OpenRouter
        except Exception:
            pass  # Fall through to fallback
        
        # Fallback: the shared-provider catalog (hermes is or_policy NONE, so this
        # lands on the hardcoded OpenRouter list — the picker is never empty even
        # when the gateway probe comes up dry).
        return await cls.shared_catalog_models()

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

    async def start(self, session_id: str) -> AgentInfoBlock:
        # Setup session-specific adapter log
        self._adapter_logger = setup_adapter_logger(
            "hermes",
            session_id,
            extra={"model": self._model or "default", "cwd": self._cwd, "base_url": self._base_url}
        )
        self._log = self._adapter_logger.info
        self._log_debug = self._adapter_logger.debug
        self._log_warn = self._adapter_logger.warning
        self._log_err = self._adapter_logger.error
        
        # If resuming an existing Hermes session, use that ID as the session key so
        # the gateway passes it as X-Hermes-Session-Id, preserving conversation history.
        self._session_key = self._resume_session_id or session_id
        self._log(f"Start: session_id={session_id}, session_key={self._session_key}")
        log.debug("HermesAdapter start: session_id=%s, _resume_session_id=%s, chosen _session_key=%s", 
                  session_id, self._resume_session_id, self._session_key)
        self._tools = await self._fetch_tools()
        return self.make_agent_info(
            name="Hermes",
            tools=self._tools,
            slash_commands=_HERMES_SLASH_COMMANDS,
            working_dirs=[self._cwd] if self._cwd else [],
        )

    async def stop(self) -> None:
        """Stop the current run by calling Hermes /v1/runs/{run_id}/stop.
        
        Retries up to 3 times with exponential backoff if run_id is not yet set
        (race condition with run start). This handles the case where stop() is
        called before POST /v1/runs completes and sets _current_run_id.
        """
        import asyncio
        
        max_retries = 3
        base_delay = 0.2
        
        # Set the stop flag to break the SSE loop
        self._stop_requested = True
        
        for attempt in range(max_retries):
            run_id = self._current_run_id
            if run_id:
                log.info("HermesAdapter.stop: Attempting to stop run %s", run_id)
                try:
                    async with httpx.AsyncClient() as client:
                        resp = await client.post(
                            f"{self._base_url}/v1/runs/{run_id}/stop",
                            headers=self._headers(),
                            timeout=5,
                        )
                        if resp.status_code in (200, 201, 202, 204):
                            log.info("HermesAdapter.stop: Successfully stopped run %s (HTTP %s)", 
                                     run_id, resp.status_code)
                        else:
                            log.warning("HermesAdapter.stop: Stop request returned HTTP %s for run %s: %s",
                                       resp.status_code, run_id, resp.text[:200])
                except httpx.TimeoutException:
                    log.warning("HermesAdapter.stop: Timeout while stopping run %s", run_id)
                except httpx.ConnectError as e:
                    log.warning("HermesAdapter.stop: Connection error while stopping run %s: %s", run_id, e)
                except Exception as e:
                    log.error("HermesAdapter.stop: Unexpected error stopping run %s: %s", run_id, e)
                finally:
                    # Always clear history even if stop failed
                    self._history.clear()
                return  # Success or error, exit retry loop
            else:
                # run_id not set yet, wait and retry
                if attempt < max_retries - 1:
                    log.debug(
                        "HermesAdapter.stop: run_id not set yet (attempt %d/%d), waiting %.2fs...",
                        attempt + 1,
                        max_retries,
                        base_delay * (2 ** attempt),
                    )
                    await asyncio.sleep(base_delay * (2 ** attempt))
                else:
                    log.warning(
                        "HermesAdapter.stop: no run_id set after %d retries - run may not have started yet",
                        max_retries,
                    )
                    self._history.clear()

    def set_model(self, model: str) -> None:
        self._model = model or None
        log.info("HermesAdapter model switched to %r", self._model)

    # ── I/O ───────────────────────────────────────────────────────────────────

    async def send(self, text: str, attachments: list[Attachment] | None = None) -> None:
        content = build_openai_content(text, attachments or [])
        self._history.append({"role": "user", "content": content})
        await self._send_queue.put(text)

    async def respond_permission(self, permission_id: str, approved: bool) -> None:
        # permission_id is run_id (set as PermissionRequestBlock.id)
        run_id = permission_id or self._current_run_id
        if not run_id:
            return
        choice = "once" if approved else "deny"
        try:
            async with httpx.AsyncClient() as client:
                await client.post(
                    f"{self._base_url}/v1/runs/{run_id}/approval",
                    json={"choice": choice},
                    headers=self._headers(),
                    timeout=5,
                )
        except Exception:
            pass

    async def get_usage(self) -> Dict[str, Any]:
        # Offload: the body does a blocking httpx GET to the gateway (+ DB
        # fallback). Kept off the event loop so a slow gateway can't stall it.
        return await asyncio.to_thread(self._get_usage_blocking)

    def _get_usage_blocking(self) -> Dict[str, Any]:
        """Extract current session statistics from the Hermes gateway.

        The adapter first tries to match by the latest run_id (most precise),
        then by session_id (via both direct lookup and DB fallback if needed).
        This works both mid-turn and post-turn because _last_known_run_id is
        preserved across turns.

        The DB fallback path resolves the session_id from a run_id when the
        gateway's active_runs_stats may not contain it (e.g., older runs or
        when the session key diverges from the Hermes internal session_id).
        """
        import httpx
        try:
            response = httpx.get(f"{self._base_url}/status", headers=self._headers(), timeout=5)
            if response.status_code != 200:
                return {
                    "session": self._session_key,
                    "error": f"gateway_http_{response.status_code}",
                    "model": self._model or "unknown",
                }

            data = response.json()
            active_runs = data.get("active_runs_stats", {})

            # Prefer run_id match (most precise); works both mid-turn and post-turn
            run_id = self._current_run_id or self._last_known_run_id
            if run_id and run_id in active_runs:
                run_data = active_runs[run_id]
                db_session_id = run_data.get("session_id")
                # If Hermes reports a different session_id, sync it for future consistency
                if db_session_id and _validate_session_id_sync(self._session_key, db_session_id, "get_usage.run_match"):
                    self._session_key = db_session_id
                return {
                    "session": db_session_id or self._session_key,
                    "match_method": "by_run_id",
                    "model": run_data.get("model", self._model or "unknown"),
                    "tokens": {
                        "input": run_data.get("input_tokens", 0),
                        "output": run_data.get("output_tokens", 0),
                        "cache_read": run_data.get("cache_read_tokens", 0),
                        "cache_write": run_data.get("cache_write_tokens", 0),
                    },
                    "active_seconds": run_data.get("active_seconds", 0),
                }

            # If we have a known run_id but it's not in active_runs, try DB fallback
            # (useful for post-turn queries when the run has completed)
            if run_id:
                db_session_id = _hermes_session_id_from_run(run_id)
                if db_session_id:
                    self._log_debug(f"get_usage: Resolved session_id {db_session_id} from run_id {run_id} via DB")
                    # Sync the session_id for consistency
                    if self._session_key != db_session_id:
                        self._session_key = db_session_id
                    # Try to find stats in active_runs by the resolved session_id
                    for rid, run_data in active_runs.items():
                        if run_data.get("session_id") == db_session_id:
                            return {
                                "session": db_session_id,
                                "match_method": "by_run_id_db_fallback",
                                "model": run_data.get("model", self._model or "unknown"),
                                "tokens": {
                                    "input": run_data.get("input_tokens", 0),
                                    "output": run_data.get("output_tokens", 0),
                                    "cache_read": run_data.get("cache_read_tokens", 0),
                                    "cache_write": run_data.get("cache_write_tokens", 0),
                                },
                                "active_seconds": run_data.get("active_seconds", 0),
                            }

            # Fallback: match by session_id directly from _session_key
            for rid, run_data in active_runs.items():
                if run_data.get("session_id") == self._session_key:
                    return {
                        "session": self._session_key,
                        "match_method": "by_session_id",
                        "model": run_data.get("model", self._model or "unknown"),
                        "tokens": {
                            "input": run_data.get("input_tokens", 0),
                            "output": run_data.get("output_tokens", 0),
                            "cache_read": run_data.get("cache_read_tokens", 0),
                            "cache_write": run_data.get("cache_write_tokens", 0),
                        },
                        "active_seconds": run_data.get("active_seconds", 0),
                    }

            # No matching active run: return empty stats with a warning
            log.debug("get_usage: No matching active run for session=%s run_id=%s last_known_run_id=%s", 
                     self._session_key, self._current_run_id, self._last_known_run_id)
            return {
                "session": self._session_key,
                "warning": "no_active_runs_found",
                "model": self._model or "unknown",
                "match_method": "none",
            }
        except Exception as e:
            log.exception("get_usage: Exception fetching usage for session=%s: %s", self._session_key, e)
            return {
                "session": self._session_key,
                "error": str(e),
                "model": self._model or "unknown",
                "match_method": "error",
            }

    async def get_status_hint(self) -> SessionStatusHintBlock | None:
        try:
            async with httpx.AsyncClient() as client:
                resp = await client.get(f"{self._base_url}/status", timeout=3)
            if resp.status_code != 200:
                return None
            data = resp.json()
            active_runs = data.get("active_runs_stats", {})
            stats: dict = {}
            if self._current_run_id and self._current_run_id in active_runs:
                stats = active_runs[self._current_run_id]
                db_session_id = stats.get("session_id")
                # Sync session_id if Hermes reports a different one
                if db_session_id and _validate_session_id_sync(self._session_key, db_session_id, "get_status_hint.run_match"):
                    self._session_key = db_session_id
            else:
                # Fallback: DB lookup if we have a last known run_id
                run_id = self._last_known_run_id
                if run_id:
                    db_session_id = _hermes_session_id_from_run(run_id)
                    if db_session_id:
                        self._log_debug(f"get_status_hint: Resolved session_id {db_session_id} from run_id {run_id} via DB")
                        if self._session_key != db_session_id:
                            self._session_key = db_session_id
                        # Try to find by resolved session_id
                        for rdata in active_runs.values():
                            if rdata.get("session_id") == db_session_id:
                                stats = rdata
                                break
                # Final fallback: match by session_id directly
                if not stats:
                    for rdata in active_runs.values():
                        if rdata.get("session_id") == self._session_key:
                            stats = rdata
                            break
            model = stats.get("model") or data.get("hermes_model") or ""
            if not model:
                return None
            return SessionStatusHintBlock(
                model=model,
                output_tokens=stats.get("output_tokens", 0),
                last_input_tokens=stats.get("input_tokens", 0),
                cache_creation_tokens=stats.get("cache_write_tokens", 0),
                context_window=context_window_for_model(model),
            )
        except Exception:
            return None

    def clear_history(self) -> None:
        self._history.clear()

    async def handle_slash_command(self, cmd: str, args: str):
        if cmd == "/status":
            return self._handle_status()
        if cmd == "/usage":
            return self._handle_usage()
        if cmd != "/memory":
            return None
        query = args.strip() or "recent"
        try:
            async with httpx.AsyncClient() as client:
                resp = await client.post(
                    f"{self._base_url}/v1/memory/search",
                    json={"query": query, "limit": 10},
                    headers=self._headers(),
                    timeout=5,
                )
                if resp.status_code == 200:
                    results = resp.json().get("results", [])
                    if results:
                        lines = [
                            f"• {r.get('content', r.get('text', ''))[:120]}"
                            for r in results
                        ]
                        msg = f"Memory ({len(results)} result(s) for \"{query}\"):\n" + "\n".join(lines)
                    else:
                        msg = f"No memory results for \"{query}\"."
                else:
                    msg = f"Memory search failed (HTTP {resp.status_code})."
        except Exception as e:
            msg = f"Memory search error: {e}"

        async def _gen():
            yield TextBlock(text=msg)
            yield SessionEndBlock(reason="completed")

        return _gen()

    def _handle_status(self):
        async def _gen():
            try:
                async with httpx.AsyncClient() as client:
                    resp = await client.get(f"{self._base_url}/status", timeout=5)
                    if resp.status_code == 200:
                        data = resp.json()
                        
                        # Find stats for current session if available
                        stats_line = ""
                        active_runs = data.get("active_runs_stats", {})
                        current_stats = None

                        # Try matching by run_id first (works mid-turn and post-turn)
                        run_id = self._current_run_id or self._last_known_run_id
                        if run_id and run_id in active_runs:
                            current_stats = active_runs[run_id]
                            db_session_id = current_stats.get("session_id")
                            # Sync session_id if Hermes reports a different one
                            if db_session_id and self._session_key != db_session_id:
                                self._log_debug(f"_handle_status: Syncing _session_key {self._session_key} -> {db_session_id} from active_runs")
                                self._session_key = db_session_id
                        else:
                            # Fallback: DB lookup if we have a run_id
                            if run_id:
                                db_session_id = _hermes_session_id_from_run(run_id)
                                if db_session_id:
                                    self._log_debug(f"_handle_status: Resolved session_id {db_session_id} from run_id {run_id} via DB")
                                    if self._session_key != db_session_id:
                                        self._session_key = db_session_id
                                    # Try to find by resolved session_id
                                    for rid, rdata in active_runs.items():
                                        if rdata.get("session_id") == db_session_id:
                                            current_stats = rdata
                                            break
                            # Final fallback: look for session_id match directly
                            if not current_stats:
                                for rid, rdata in active_runs.items():
                                    if rdata.get("session_id") == self._session_key:
                                        current_stats = rdata
                                        break
                        
                        if current_stats:
                            in_t = current_stats.get("input_tokens", 0)
                            out_t = current_stats.get("output_tokens", 0)
                            stats_line = (
                                f"• **Session Tokens:** {in_t + out_t} (↑{in_t} ↓{out_t})\n"
                                f"• **Model:** `{current_stats.get('model', 'unknown')}`\n"
                            )

                        hermes_sessions = data.get("hermes_active_sessions")
                        hermes_model = data.get("hermes_model", "")
                        total_sessions = data.get("hermes_total_sessions")
                        total_messages = data.get("hermes_total_messages")
                        db_mb = data.get("hermes_db_size_mb")

                        hermes_line = ""
                        if hermes_sessions is not None:
                            hermes_line += f"• **Hermes Sessions:** {hermes_sessions} active"
                            if total_sessions is not None:
                                hermes_line += f" / {total_sessions} total"
                            hermes_line += "\n"
                        if total_messages is not None:
                            size_str = f" ({db_mb} MB)" if db_mb is not None else ""
                            hermes_line += f"• **Hermes Messages:** {total_messages}{size_str}\n"

                        model_line = ""
                        model_val = (current_stats or {}).get("model") or hermes_model
                        if model_val:
                            model_line = f"• **Model:** `{model_val}`\n"

                        status_text = (
                            f"**Hermes Gateway**\n"
                            f"• **Session:** `{self._session_key}`\n"
                            f"• **Version:** `{data.get('version', 'unknown')}`\n"
                            f"• **Working Dir:** `{self._cwd or 'N/A'}`\n"
                            f"{model_line}"
                            f"{stats_line}"
                            f"{hermes_line}"
                            f"• **Tools:** {data.get('tools_count', 0)}\n"
                        )
                    else:
                        status_text = f"Failed to fetch status (HTTP {resp.status_code})"
            except Exception as e:
                status_text = f"Status error: {e}"

            yield TextBlock(text=status_text)
            yield SessionEndBlock(reason="completed")
        return _gen()

    def _handle_usage(self):
        async def _gen():
            # Try to fetch current usage from the gateway status
            try:
                import httpx
                async with httpx.AsyncClient() as client:
                    resp = await client.get(f"{self._base_url}/status")
                    if resp.status_code == 200:
                        data = resp.json()
                        # Find the stats for our current session in active_runs_stats
                        # Note: self._session_key is the session ID we passed
                        stats = {}
                        active_runs = data.get("active_runs_stats", {})

                        # Try matching by run_id first (works mid-turn and post-turn)
                        run_id = self._current_run_id or self._last_known_run_id
                        if run_id and run_id in active_runs:
                            stats = active_runs[run_id]
                            db_session_id = stats.get("session_id")
                            # Sync session_id if Hermes reports a different one
                            if db_session_id and self._session_key != db_session_id:
                                self._log_debug(f"_handle_usage: Syncing _session_key {self._session_key} -> {db_session_id} from active_runs")
                                self._session_key = db_session_id
                        else:
                            # Fallback: DB lookup if we have a run_id
                            if run_id:
                                db_session_id = _hermes_session_id_from_run(run_id)
                                if db_session_id:
                                    self._log_debug(f"_handle_usage: Resolved session_id {db_session_id} from run_id {run_id} via DB")
                                    if self._session_key != db_session_id:
                                        self._session_key = db_session_id
                                    # Try to find by resolved session_id
                                    for rid, rdata in active_runs.items():
                                        if rdata.get("session_id") == db_session_id:
                                            stats = rdata
                                            break
                            # Final fallback: look for session_id match directly
                            if not stats:
                                for rid, rdata in active_runs.items():
                                    if rdata.get("session_id") == self._session_key:
                                        stats = rdata
                                        break
                        
                        usage_text = (
                            f"**Session Usage**\n"
                            f"• **Session ID:** `{self._session_key}`\n"
                            f"• **Model:** `{stats.get('model', self._model or 'Default')}`\n"
                            f"• **Input:** {stats.get('input_tokens', 0)} tokens\n"
                            f"• **Output:** {stats.get('output_tokens', 0)} tokens\n"
                            f"• **Cache Read:** {stats.get('cache_read_tokens', 0)} tokens\n"
                            f"• **Cache Write:** {stats.get('cache_write_tokens', 0)} tokens\n"
                        )
                        yield TextBlock(text=usage_text)
                    else:
                        yield TextBlock(text=f"**Usage**\n• *Gateway returned status {resp.status_code}*")
            except Exception as e:
                yield TextBlock(text=f"**Usage**\n• *Local usage tracking error: {str(e)}*")
            
            yield SessionEndBlock(reason="completed")
        return _gen()

    async def stream(self) -> AsyncIterator[Block]:
        while True:
            await self._send_queue.get()
            async for block in self._run_turn():
                yield block

    # ── Run loop ──────────────────────────────────────────────────────────────

    async def _run_turn(self) -> AsyncIterator[Block]:
        self._stop_requested = False  # Reset stop flag for new turn
        yield AudioHintBlock(state="agent_speaking")

        user_message = self._last_user_text()
        # Pass the full history (including current message with any images/attachments)
        # as conversation_history so multimodal content isn't dropped.
        # input= is the text-only version Hermes uses for routing/logging.
        conv_history = self._history
        
        log.debug("HermesAdapter _run_turn: Current session key='%s', history length=%d", 
                  self._session_key, len(conv_history))
        
        # Log the last few messages in the history for debugging
        if conv_history:
            recent_count = min(5, len(conv_history))
            log.debug("HermesAdapter _run_turn: Recent history: %s", 
                      [(msg.get('role'), str(msg.get('content', ''))[:100]) for msg in conv_history[-recent_count:]])

        # Before making the API request, we should update session_key with the real one sent in response
        # if we have a response from an earlier call with it
        run_body: dict = {
            "input": user_message,
            "conversation_history": conv_history,
        }
        # Forward setup_dir if present
        if self._cwd:
            run_body["setup_dir"] = self._cwd

        # Optional model override — when provided by iOS, use it instead of
        # Hermes' configured default. Only valid model IDs are sent through.
        if self._model:
            run_body["model"] = self._model
        # Build instructions: always include beacon session routing hint so the
        # globally-registered sentinel MCP tools can route blocks back correctly.
        instructions_parts = []
        if self._cwd:
            instructions_parts.append(f"Work in directory: {self._cwd}")
        
        # Pass the current run_id to the agent via instructions so it can be 
        # included in tool calls or telemetry if needed.
        if self._current_run_id:
            instructions_parts.append(f"Current Run ID: {self._current_run_id}")

        if self._session_key:
            instructions_parts.append(
                f"Beacon session ID: {self._session_key}. "
                f"When calling any beacon tool (request_user_input, suggest_next_steps, "
                f"open_url, request_file_download), always include "
                f'"_beacon_session": "{self._session_key}" in the arguments.'
            )
        instructions_parts.append(
            "BEHAVIOR GUIDELINE: You are running on an iOS device via the Beacon app. "
            "When asking the user a question with options or a simple confirmation, "
            "PREFER using the `request_user_input` tool to provide a native UI experience "
            "instead of just typing text. Only use plain text for open-ended or complex questions. "
            "IMPORTANT: This guideline only applies to your own (main session) responses. "
            "When spawning subagents via delegate_task, do NOT instruct them to use request_user_input; "
            "they must remain fully autonomous and non-interactive."
        )
        # Operator's append-only custom prompt (per-machine [prompts] toggle). Hermes builds
        # its own instructions and never calls build_append_system_prompt, so append here too
        # for parity with every other adapter. Same shared helper → same gate + wording.
        from .base import _custom_prompt_lines
        _custom = "\n".join(_custom_prompt_lines()).strip()
        if _custom:
            instructions_parts.append(_custom)
        if instructions_parts:
            run_body["instructions"] = "\n\n".join(instructions_parts)

        # Start the run
        headers = self._headers()
        log.debug("HermesAdapter POST /v1/runs: headers=%s, session_key=%s", 
                  {k: ("***" if k.lower() == "authorization" else v) for k, v in headers.items()},
                  self._session_key)
        try:
            async with httpx.AsyncClient() as client:
                resp = await client.post(
                    f"{self._base_url}/v1/runs",
                    json=run_body,
                    headers=headers,
                    timeout=10,
                )
        except Exception as e:
            yield ErrorBlock(message=f"Hermes connection failed: {e}", recoverable=True)
            yield SessionEndBlock(reason="error")
            return

        if resp.status_code not in (200, 201, 202):
            yield ErrorBlock(message=f"Hermes /v1/runs HTTP {resp.status_code}", recoverable=True)
            yield SessionEndBlock(reason="error")
            return

        run_resp = resp.json()
        run_id = run_resp.get("run_id", "")
        # The Hermes native API may also return the resolved session_id (which can differ
        # from our _session_key if a new session was created). Store this if present.
        run_session_id = run_resp.get("session_id") or run_resp.get("session")
        if run_session_id and _validate_session_id_sync(self._session_key, run_session_id, "post_v1_runs"):
            self._log_debug(f"Hermes POST /v1/runs: updating _session_key {self._session_key} -> {run_session_id}")
            self._session_key = run_session_id
        self._current_run_id = run_id
        self._last_known_run_id = run_id  # Remember for post-turn usage/lookup

        # Stream text live: each message.delta becomes a TextBlock(done=False) chunk.
        # Use run_id as the stable block id so iOS merges them in place.
        stream_text = ""
        stream_id   = run_id
        # Track whether we've received a 'text' event from gateway - if so, stop yielding message.delta
        # to prevent duplication. Gateway sends 'text' with full accumulated text; we should use that.
        using_gateway_text = False

        # Stream SSE events
        session_ended = False
        try:
            async with httpx.AsyncClient(timeout=httpx.Timeout(120, connect=10)) as client:
                async with client.stream(
                    "GET",
                    f"{self._base_url}/v1/runs/{run_id}/events",
                    headers={**self._headers(), "Accept": "text/event-stream"},
                ) as response:
                    async for line in response.aiter_lines():
                        if self._stop_requested:
                            log.info("SSE loop: stop_requested=True, breaking stream")
                            break
                        if not line.startswith("data: "):
                            continue
                        raw = line[6:].strip()
                        if not raw or raw == "[DONE]":
                            break
                        try:
                            event = json.loads(raw)
                        except Exception:
                            log.warning("SSE parse error for line: %r", line)
                            continue
                        etype = event.get("event") or event.get("type", "")
                        log.debug("SSE event: type=%r keys=%s", etype, list(event.keys()))
                        if etype == "message.delta":
                            # Skip delta events if we're using gateway 'text' events to prevent duplication
                            if using_gateway_text:
                                log.debug("Skipping message.delta (already using gateway 'text' events)")
                                continue
                            delta = event.get("delta", "")
                            # If the delta starts with the current buffer, the gateway is
                            # erroneously sending full text in the delta field.
                            if delta.startswith(stream_text) and len(delta) > len(stream_text):
                                stream_text = delta
                            else:
                                stream_text += delta
                            yield TextBlock(text=stream_text, done=False, id=stream_id)
                        elif etype == "text":
                            # Beacon gateway format: full accumulated text with done flag
                            # Mark that we're using gateway text events to prevent duplicating message.delta
                            using_gateway_text = True
                            text_content = event.get("text", "")
                            text_id = event.get("id") or stream_id
                            text_done = event.get("done", True)
                            log.debug("SSE text: done=%s len=%d", text_done, len(text_content))
                            
                            # Deduplicate: if we just split blocks and the text block is marked done,
                            # ensure we don't append to history twice.
                            yield TextBlock(text=text_content, done=text_done, id=text_id)
                            if text_done and text_content and text_content != stream_text:
                                self._history.append({"role": "assistant", "content": text_content})
                                stream_text = ""
                        elif etype == "session_status":
                            # Full stats payload from beacon gateway
                            yield SessionStatusHintBlock(
                                model=event.get("model", ""),
                                output_tokens=event.get("output_tokens", 0),
                                last_input_tokens=event.get("input_tokens", 0),
                                cache_creation_tokens=event.get("cache_write_tokens", 0),
                                cost_usd=event.get("cost_usd", 0.0),
                                turns=event.get("turns", 0),
                            )
                        else:
                            if etype in ("run.completed", "run.failed", "run.cancelled", "session_end") and stream_text:
                                log.info("Turn end event: %s (stop_requested=%s)", etype, self._stop_requested)
                                # Close the streaming block, then split for code fences
                                yield TextBlock(text=stream_text, done=True, id=stream_id)
                                self._history.append({"role": "assistant", "content": stream_text})
                                stream_text = ""
                            async for block in self._parse_event(event):
                                log.debug("SSE→block: type=%s", type(block).__name__)
                                yield block
                                if isinstance(block, SessionEndBlock):
                                    session_ended = True
        except Exception as e:
            log.exception("SSE stream error for run %s: %s", run_id, e)
            yield ErrorBlock(message=str(e), recoverable=True)
            yield SessionEndBlock(reason="error")
            session_ended = True
        finally:
            self._current_run_id = None
            self._pending_tool_ids.clear()
            self._pending_tool_meta.clear()
            self._tool_call_counts.clear()
            self._subagent_goals.clear()
            # Keep _last_known_run_id for post-turn usage/status queries

        # Flush any remaining text (run ended without run.completed event)
        if stream_text:
            yield TextBlock(text=stream_text, done=True, id=stream_id)
            self._history.append({"role": "assistant", "content": stream_text})

        # Guarantee a turn-end block — the gateway may end the stream with [DONE]
        # without emitting run.completed, leaving the session stuck in "generating".
        if not session_ended:
            log.warning("[hermes] SSE ended without run.completed for run %s — emitting SessionEndBlock", run_id)
            yield SessionEndBlock(reason="completed")

    # ── Event parser ─────────────────────────────────────────────────────────

    async def _parse_event(self, event: dict) -> AsyncIterator[Block]:
        etype = event.get("event") or event.get("type", "")
        
        # Extract session information from events if available
        session_id_from_event = event.get("session_id", event.get("session", event.get("conversation_id")))
        if session_id_from_event and _validate_session_id_sync(self._session_key, session_id_from_event, "parse_event"):
            self._session_key = session_id_from_event

        if etype in ("reasoning.available", "thinking"):
            text = event.get("thinking") or event.get("text", "")
            label = event.get("label", "thinking")
            if text:
                yield ThinkingBlock(thinking=text, label=label)

        elif etype in ("tool.started", "tool_call"):
            tool_name = event.get("tool") or event.get("name", "unknown")
            status = event.get("status", "running")
            log.debug("[hermes] event=%s name=%s status=%s id=%s args_keys=%s",
                      etype, tool_name, status, event.get("id"), list((event.get("args") or {}).keys()))

            # Support for the live plan / todo tool. This ensures the plan
            # is rendered as a native persistent card in the app.
            if tool_name == "todo":
                todos = (event.get("args") or {}).get("todos", [])
                yield self.todo_block(
                    self._session_key or "hermes",
                    todos,
                    id_prefix="plan",
                )
                return

            # Subagents (delegate_task) → a live "subagents" checkbox card instead
            # of a raw tool-call block. args is either {goal: "…"} (one subagent)
            # or {tasks: [{goal: "…"}, …]} (parallel); the same tc_id is reused for
            # running→done. cap_subagents stays False (no native Agent block).
            if tool_name == "delegate_task":
                tc_id = event.get("id") or "unknown"
                goals = _delegate_goals(event.get("args") or {})
                if goals:
                    self._subagent_goals[tc_id] = goals
                else:
                    goals = self._subagent_goals.get(tc_id, [])
                done = status in ("done", "error")
                yield self.todo_block(
                    self._session_key or "",
                    [{"content": g, "status": "completed" if done else "in_progress"}
                     for g in goals],
                    id_prefix="subagents",
                )
                yield AudioHintBlock(state="agent_speaking" if done else "agent_silent")
                return   # suppress the generic ToolCallBlock for delegate_task

            # Beacon gateway sends tool_call with status=done/error — route to done logic
            if status in ("done", "error"):
                tc_id = event.get("id") or "unknown"
                lbl, subj, saved_args = self._pending_tool_meta.pop(tc_id, ("", "", {}))
                is_error = (status == "error")
                block = ToolCallBlock(
                    id=tc_id, name=tool_name, args=saved_args,
                    status=status, label=lbl, subject=subj, preview="",
                )
                log.debug("[hermes] → ToolCallBlock status=%s name=%s args=%s label=%r subject=%r",
                          status, tool_name, saved_args, lbl, subj)
                yield block
                yield AudioHintBlock(state="agent_speaking")
            else:
                preview = event.get("preview", "")
                # Rich adapter provides full args
                args = event.get("args") or ({"preview": preview} if preview else {})

                count = self._tool_call_counts.get(tool_name, 0)
                self._tool_call_counts[tool_name] = count + 1
                key = f"{tool_name}:{count}"

                # Use provided ID if available (rich gateway), else generate
                tc_id = event.get("id") or str(uuid.uuid4())
                self._pending_tool_ids[key] = tc_id
                lbl, subj = tool_label_subject(tool_name, args)
                self._pending_tool_meta[tc_id] = (lbl, subj, args)

                block = ToolCallBlock(
                    id=tc_id,
                    name=tool_name,
                    args=args,
                    status="running",
                    label=lbl,
                    subject=subj,
                )
                log.debug("[hermes] → ToolCallBlock status=running name=%s args=%s label=%r subject=%r",
                          tool_name, args, lbl, subj)
                yield AudioHintBlock(state="agent_silent")
                yield block

        elif etype == "tool_result":
            tc_id = event.get("tool_call_id") or "unknown"
            # delegate_task result is already surfaced as the subagents card; the
            # raw payload is a JSON blob of per-subagent stats — drop it and mark
            # the card complete.
            if tc_id in self._subagent_goals:
                goals = self._subagent_goals.pop(tc_id, [])
                yield self.todo_block(
                    self._session_key or "",
                    [{"content": g, "status": "completed"} for g in goals],
                    id_prefix="subagents",
                )
                return
            content = event.get("content", "")
            is_error = bool(event.get("is_error"))
            log.debug("[hermes] event=tool_result tc_id=%s is_error=%s content_len=%d content_preview=%r",
                      tc_id, is_error, len(str(content)), str(content)[:120])
            yield ToolResultBlock(
                tool_call_id=tc_id,
                content=str(content),
                is_error=is_error,
            )

        elif etype == "tool.completed":
            tool_name = event.get("tool", "")
            log.debug("[hermes] event=tool.completed name=%s error=%s result_keys=%s",
                      tool_name, event.get("error"), list(event.keys()))
            # Find the oldest pending call for this tool name
            key = next(
                (k for k in self._pending_tool_ids if k.startswith(f"{tool_name}:")),
                None,
            )
            tc_id = self._pending_tool_ids.pop(key, str(uuid.uuid4())) if key else str(uuid.uuid4())
            lbl, subj, saved_args = self._pending_tool_meta.pop(tc_id, ("", "", {}))
            is_error = bool(event.get("error"))
            result_content = event.get("result") or event.get("output") or ""
            preview = "" if is_error else str(result_content).split("\n")[0][:120]
            block = ToolCallBlock(
                id=tc_id, name=tool_name, args=saved_args,
                status="error" if is_error else "done",
                label=lbl, subject=subj, preview=preview,
            )
            log.debug("[hermes] → ToolCallBlock status=%s name=%s args=%s label=%r subject=%r preview=%r",
                      block.status, tool_name, saved_args, lbl, subj, preview)
            yield block
            if result_content:
                log.debug("[hermes] → ToolResultBlock tc_id=%s is_error=%s content_len=%d",
                          tc_id, is_error, len(str(result_content)))
                yield ToolResultBlock(
                    tool_call_id=tc_id,
                    content=str(result_content),
                    is_error=is_error,
                )
            yield AudioHintBlock(state="agent_speaking")

        elif etype == "approval.request":
            # Run pauses here until respond_permission() posts to /approval
            run_id = event.get("run_id") or self._current_run_id or ""
            action = (
                event.get("action")
                or event.get("tool")
                or event.get("description")
                or "Unknown action"
            )
            risk_raw = event.get("risk", "medium")
            risk = risk_raw if risk_raw in ("low", "medium", "high") else "medium"
            details = event.get("details") or event.get("preview") or ""
            yield PermissionRequestBlock(id=run_id, action=action, risk=risk, details=details)

        elif etype == "progress":
            yield ProgressBlock(
                id=event.get("id") or str(uuid.uuid4()),
                label=event.get("label", "progress"),
                value=float(event.get("value", 0.0)),
                status=event.get("status", ""),
            )

        elif etype == "error":
            msg = event.get("message") or event.get("error") or "Unknown error"
            retryable = bool(event.get("retryable", False))
            log.warning("[hermes] error event: %s retryable=%s", msg, retryable)
            yield ErrorBlock(message=msg, recoverable=retryable)

        elif etype in ("run.completed", "session_end"):
            yield AudioHintBlock(state="agent_silent")
            yield SessionEndBlock(
                reason="completed",
                input_tokens=event.get("input_tokens", 0),
                output_tokens=event.get("output_tokens", 0),
            )

        elif etype == "run.failed":
            err = event.get("error") or "Hermes run failed"
            yield ErrorBlock(message=err, recoverable=True)
            yield AudioHintBlock(state="agent_silent")
            yield SessionEndBlock(reason="error")

        elif etype == "run.cancelled":
            yield AudioHintBlock(state="agent_silent")
            yield SessionEndBlock(reason="user_stopped")

    # ── Helpers ───────────────────────────────────────────────────────────────

    def _headers(self) -> dict:
        h: dict[str, str] = {}
        if self._api_key:
            h["Authorization"] = f"Bearer {self._api_key}"
        if self._session_key:
            h["X-Hermes-Session-Id"] = self._session_key
            log.debug("HermesAdapter setting X-Hermes-Session-Id header: %s", self._session_key)
        else:
            log.warning("HermesAdapter no session ID to set in X-Hermes-Session-Id header")
        return h

    async def _fetch_tools(self) -> list[dict]:
        try:
            async with httpx.AsyncClient() as client:
                resp = await client.get(f"{self._base_url}/mcp", timeout=5)
                if resp.status_code == 200:
                    return resp.json().get("tools", [])
        except Exception:
            pass
        return []

    def _last_user_text(self) -> str:
        for msg in reversed(self._history):
            if msg.get("role") == "user":
                content = msg.get("content")
                if isinstance(content, str):
                    return content
                if isinstance(content, list):
                    texts = [c["text"] for c in content if c.get("type") == "text"]
                    return "\n".join(texts)
        return ""
