sidebyside/scripts/extract_hermes.py

"""Safe Hermes data extraction.

This module exposes `extract_hermes(home: Path) -> dict` and performs no work
at import time. It reads a constrained set of files under `~/.hermes` and
`~/.local/state/hermes`, summarizes what it finds, and avoids embedding raw
secrets or large prompt/request payloads.
"""

from __future__ import annotations

import json
import os
import re
import sqlite3
from pathlib import Path
from typing import Any, Iterable


MAX_SAMPLE_ITEMS = 8
MAX_TEXT_PREVIEW = 180
LOW_USAGE_SKILL_THRESHOLD = 3
SKIP_DIR_NAMES = {
    ".git",
    ".mypy_cache",
    ".pytest_cache",
    ".ruff_cache",
    ".uv-cache",
    ".venv",
    "__pycache__",
    "build",
    "dist",
    "node_modules",
    "site-packages",
    "venv",
}


def extract_hermes(home: Path) -> dict:
    home = Path(home).expanduser()
    hermes_dir = home / ".hermes"
    state_dir = home / ".local" / "state" / "hermes"

    sources: list[dict[str, Any]] = []
    warnings: list[str] = []
    discoveries: list[dict[str, Any]] = []

    config = _extract_config(hermes_dir, state_dir, sources, warnings)
    status = _extract_status(hermes_dir, state_dir, config, sources, warnings)
    extensions = _extract_extensions(hermes_dir, state_dir, sources, warnings)
    usage = _extract_usage(hermes_dir, state_dir, sources, warnings)
    instruction_layers = _extract_instruction_layers(hermes_dir, state_dir, sources, warnings)

    _add_discoveries(discoveries, status, config, extensions, usage)

    return {
        "id": "hermes",
        "name": "Hermes",
        "status": status,
        "config": config,
        "instruction_layers": instruction_layers,
        "extensions": extensions,
        "usage": usage,
        "discoveries": discoveries,
        "sources": _dedupe_records(sources, key_fields=("path", "kind")),
        "warnings": _dedupe_preserve_order(warnings),
    }


def _extract_config(
    hermes_dir: Path,
    state_dir: Path,
    sources: list[dict[str, Any]],
    warnings: list[str],
) -> dict[str, Any]:
    config: dict[str, Any] = {}
    yaml_files = [
        path
        for path in (
            hermes_dir / "config.yaml",
            hermes_dir / "dashboard" / "alerts.yaml",
        )
        if path.exists()
    ]

    json_candidates = [
        path
        for path in (
            hermes_dir / "gateway_state.json",
            hermes_dir / "channel_directory.json",
            hermes_dir / "cron" / "jobs.json",
        )
        if path.exists()
    ]

    for path in yaml_files:
        if _looks_sensitive(path):
            continue
        parsed = _parse_yaml_summary(path)
        if parsed:
            _merge_nested(config, parsed)
            _record_source(sources, path, kind="yaml-config", details=parsed.get("_summary", {}))

    # Prefer a few known configuration JSON files, but do not ingest secrets.
    for path in json_candidates:
        if _looks_sensitive(path):
            continue
        if path.name in {
            "config.json",
            "settings.json",
            "state.json",
            "runtime.json",
        } or "config" in path.name.lower() or "settings" in path.name.lower():
            parsed = _safe_json_load(path)
            if isinstance(parsed, dict):
                summary = _summarize_json_dict(parsed)
                _merge_nested(config, summary)
                _record_source(sources, path, kind="json-config", details={"keys": sorted(summary.keys())[:12]})
            elif parsed is not None:
                warnings.append(f"unexpected JSON config shape at {path}")

    config = _normalize_config(config)
    if not config:
        config = {}
    return config


def _extract_status(
    hermes_dir: Path,
    state_dir: Path,
    config: dict[str, Any],
    sources: list[dict[str, Any]],
    warnings: list[str],
) -> dict[str, Any]:
    state_db = _find_first_file((state_dir, hermes_dir), {"state.db", "hermes.db", "db.sqlite", "state.sqlite"})
    db_stats = _read_state_db(state_db, sources, warnings)
    session_file_count = len(list(_collect_session_files(hermes_dir, state_dir)))
    channel_state = _merge_channel_state(db_stats.get("channels", {}), hermes_dir / "channel_directory.json", sources)
    gateway_state = _merge_gateway_state(db_stats.get("gateway", {}), hermes_dir / "gateway_state.json", sources)

    status = {
        "exists": {"hermes_dir": hermes_dir.exists(), "state_dir": state_dir.exists(), "state_db": bool(state_db and state_db.exists())},
        "gateway": gateway_state,
        "channels": channel_state,
        "cron": db_stats.get("cron", {}),
        "counts": {
            "sessions": db_stats.get("session_count", 0),
            "session_files": session_file_count,
            "tools": db_stats.get("tool_count", 0),
            "models": db_stats.get("model_count", 0),
            "sources": db_stats.get("source_count", 0),
            "channels": channel_state.get("count", 0),
            "profiles": db_stats.get("profile_count", 0),
            "skills": db_stats.get("skill_count", 0),
        },
        "flags": {
            "checkpoints": _get_bool_flag(config, "checkpoints.enabled"),
            "smart_model_routing": _get_bool_flag(config, "smart_model_routing.enabled"),
            "browser_record_sessions": _get_bool_flag(config, "browser.record_sessions"),
            "gateway_startup_failed": bool(gateway_state.get("startup_failed")),
        },
    }
    return status


def _extract_extensions(
    hermes_dir: Path,
    state_dir: Path,
    sources: list[dict[str, Any]],
    warnings: list[str],
) -> dict[str, list[dict[str, Any]]]:
    skills: list[dict[str, Any]] = []
    toolsets: list[dict[str, Any]] = []
    channels: list[dict[str, Any]] = []
    profiles: list[dict[str, Any]] = []

    skills_snapshot = _find_first_file((hermes_dir, state_dir), {".skills_prompt_snapshot.json"})
    if not skills_snapshot:
        # Fallback to any obvious snapshot file.
        for base in (hermes_dir, state_dir):
            candidate = _find_first_file((base,), {"skills_prompt_snapshot.json", ".skills_prompt_snapshot.json"})
            if candidate:
                skills_snapshot = candidate
                break

    if skills_snapshot and skills_snapshot.exists():
        payload = _safe_json_load(skills_snapshot)
        if isinstance(payload, dict):
            skills.extend(_summarize_skill_snapshot(payload))
            _record_source(sources, skills_snapshot, kind="skills-snapshot", details={"skills": len(skills)})
        else:
            warnings.append(f"unable to parse skills snapshot at {skills_snapshot}")

    skills_dir = _find_first_dir((hermes_dir, state_dir), {"skills", ".skills"})
    if skills_dir:
        skills.extend(_summarize_skills_dir(skills_dir))
        _record_source(sources, skills_dir, kind="skills-dir", details={"items": len(skills)})

    state_db = _find_first_file((state_dir, hermes_dir), {"state.db", "hermes.db", "db.sqlite", "state.sqlite"})
    db_stats = _read_state_db(state_db, sources, warnings)
    toolsets.extend(db_stats.get("toolsets", []))
    channels.extend(db_stats.get("channels_list", []))
    profiles.extend(db_stats.get("profiles", []))

    return {
        "skills": _dedupe_records(skills, key_fields=("name", "path", "id")),
        "toolsets": _dedupe_records(toolsets, key_fields=("name", "path", "id")),
        "channels": _dedupe_records(channels, key_fields=("name", "state", "path")),
        "profiles": _dedupe_records(profiles, key_fields=("name", "path", "id")),
    }


def _extract_usage(
    hermes_dir: Path,
    state_dir: Path,
    sources: list[dict[str, Any]],
    warnings: list[str],
) -> dict[str, Any]:
    state_db = _find_first_file((state_dir, hermes_dir), {"state.db", "hermes.db", "db.sqlite", "state.sqlite"})
    db_stats = _read_state_db(state_db, sources, warnings)
    session_files = list(_collect_session_files(hermes_dir, state_dir))

    session_summaries = []
    for path in session_files[:50]:
        summary = _summarize_session_file(path)
        if summary:
            session_summaries.append(summary)
            _record_source(sources, path, kind="session-json", details={"events": summary.get("events"), "messages": summary.get("messages")})

    prompt_counts = _count_prompt_like_artifacts(hermes_dir, state_dir)
    usage = {
        "sessions": {
            "db_count": db_stats.get("session_count", 0),
            "file_count": len(session_files),
            "samples": session_summaries[:MAX_SAMPLE_ITEMS],
        },
        "tools": {
            "db_count": db_stats.get("tool_count", 0),
            "top": db_stats.get("tool_usage", [])[:MAX_SAMPLE_ITEMS],
        },
        "models": {
            "db_count": db_stats.get("model_count", 0),
            "top": db_stats.get("model_usage", [])[:MAX_SAMPLE_ITEMS],
        },
        "sources": {
            "db_count": db_stats.get("source_count", 0),
            "top": db_stats.get("source_usage", [])[:MAX_SAMPLE_ITEMS],
        },
        "cron": db_stats.get("cron", {}),
        "prompts": prompt_counts,
        "observed_skills": db_stats.get("skill_usage", [])[:MAX_SAMPLE_ITEMS],
    }
    return usage


def _extract_instruction_layers(
    hermes_dir: Path,
    state_dir: Path,
    sources: list[dict[str, Any]],
    warnings: list[str],
) -> list[dict[str, Any]]:
    layers: list[dict[str, Any]] = []
    candidates = [
        hermes_dir / "SOUL.md",
        hermes_dir / "ADHS_LEARNING_PROFILE.md",
        hermes_dir / "hermes-agent" / "AGENTS.md",
        hermes_dir / "hermes-agent" / "README.md",
    ]
    profiles_dir = hermes_dir / "profiles"
    if profiles_dir.exists():
        for profile in sorted(profiles_dir.iterdir()):
            if not profile.is_dir() or _skip_dir_name(profile.name):
                continue
            for name in ("SOUL.md", "AGENTS.md", "README.md"):
                candidate = profile / name
                if candidate.exists():
                    candidates.append(candidate)
    for path in candidates:
        if not path.exists():
            continue
        if _looks_sensitive(path):
            continue
        layer = _summarize_instruction_file(path)
        if layer:
            layers.append(layer)
            _record_source(sources, path, kind="instruction-layer", details={"kind": layer.get("kind")})
    return _dedupe_records(layers, key_fields=("path", "kind", "name"))


def _add_discoveries(
    discoveries: list[dict[str, Any]],
    status: dict[str, Any],
    config: dict[str, Any],
    extensions: dict[str, list[dict[str, Any]]],
    usage: dict[str, Any],
) -> None:
    channels = status.get("channels", {}) or {}
    gateway = status.get("gateway", {}) or {}
    flags = status.get("flags", {}) or {}
    counts = status.get("counts", {}) or {}

    if channels.get("count", 0) == 0:
        discoveries.append({"kind": "empty_channels", "message": "No channels were discovered in Hermes state."})

    if gateway.get("startup_failed"):
        discoveries.append({"kind": "gateway_startup_failed", "message": "Hermes gateway reports startup failure.", "state": gateway})

    for flag_name in ("checkpoints", "smart_model_routing", "browser_record_sessions"):
        if flags.get(flag_name) is False:
            discoveries.append({"kind": f"{flag_name}_disabled", "message": f"{flag_name} is disabled."})

    skill_names = [item.get("name") for item in extensions.get("skills", []) if item.get("name")]
    observed_skills = {item.get("name"): item.get("count", 0) for item in usage.get("observed_skills", []) if item.get("name")}
    low_usage = [name for name in skill_names if observed_skills.get(name, 0) <= LOW_USAGE_SKILL_THRESHOLD]
    if len(skill_names) >= 8 and low_usage:
        discoveries.append(
            {
                "kind": "many_skills_low_usage",
                "message": "Many known skills have low observed usage.",
                "known_skills": len(skill_names),
                "low_usage_skills": low_usage[:MAX_SAMPLE_ITEMS],
            }
        )

    if counts.get("sessions", 0) and counts.get("tools", 0) == 0:
        discoveries.append({"kind": "no_tool_usage", "message": "Sessions exist but no tool usage was observed."})


def _merge_gateway_state(
    db_gateway: dict[str, Any],
    gateway_path: Path,
    sources: list[dict[str, Any]],
) -> dict[str, Any]:
    gateway = dict(db_gateway or {})
    payload = _safe_json_load(gateway_path)
    if isinstance(payload, dict):
        status = payload.get("status") or payload.get("state") or payload.get("gateway_state")
        if status and "status" not in gateway:
            gateway["status"] = _scrub_text(status)
        error = payload.get("last_error") or payload.get("exit_reason")
        if error and "last_error" not in gateway:
            gateway["last_error"] = _scrub_text(error)
        lowered = " ".join(
            str(payload.get(key, ""))
            for key in ("status", "state", "gateway_state", "last_error", "exit_reason")
        ).lower()
        platforms = payload.get("platforms")
        if isinstance(platforms, dict):
            for item in platforms.values():
                if isinstance(item, dict):
                    lowered += " " + " ".join(str(item.get(key, "")) for key in ("state", "error_message", "error_code")).lower()
        if "failed" in lowered or "error" in lowered:
            gateway["startup_failed"] = True
        _record_source(sources, gateway_path, kind="gateway-state", details={"keys": sorted(payload.keys())[:12]})
    gateway.setdefault("startup_failed", False)
    return gateway


def _merge_channel_state(
    db_channels: dict[str, Any],
    channels_path: Path,
    sources: list[dict[str, Any]],
) -> dict[str, Any]:
    channels = dict(db_channels or {})
    payload = _safe_json_load(channels_path)
    if isinstance(payload, dict):
        platform_counts = {}
        total = 0
        for name, entries in payload.items():
            if isinstance(entries, list):
                platform_counts[name] = len(entries)
                total += len(entries)
        channels["platform_counts"] = platform_counts
        channels["count"] = max(int(channels.get("count") or 0), total)
        channels["empty_platforms"] = [name for name, count in platform_counts.items() if count == 0]
        _record_source(sources, channels_path, kind="channel-directory", details={"platforms": len(platform_counts)})
    channels.setdefault("count", 0)
    return channels


def _read_state_db(
    path: Path | None,
    sources: list[dict[str, Any]],
    warnings: list[str],
) -> dict[str, Any]:
    if not path or not path.exists():
        return {}

    try:
        conn = sqlite3.connect(f"file:{path}?mode=ro&immutable=1", uri=True)
    except sqlite3.Error as exc:
        warnings.append(f"unable to open state db {path}: {exc}")
        return {}

    conn.row_factory = sqlite3.Row
    stats: dict[str, Any] = {}
    try:
        stats["session_count"] = _count_rows(conn, ("sessions", "session"))
        stats["tool_count"] = _count_rows(conn, ("tools", "tool_usage", "session_tools"))
        stats["model_count"] = _count_rows(conn, ("models", "model_usage", "session_models"))
        stats["source_count"] = _count_rows(conn, ("sources", "source_usage"))
        stats["profile_count"] = _count_rows(conn, ("profiles", "profile"))
        stats["skill_count"] = _count_rows(conn, ("skills", "skill"))
        stats["cron"] = _read_cron_state(conn)
        stats["gateway"] = _read_gateway_state(conn)
        stats["channels"] = _read_channel_state(conn)
        stats["channels_list"] = _read_channels_list(conn)
        stats["toolsets"] = _read_named_inventory(conn, ("toolsets", "toolset"))
        stats["profiles"] = _read_named_inventory(conn, ("profiles", "profile"))
        stats["tool_usage"] = _read_ranked_usage(conn, ("tool_usage", "tools", "session_tools"), label="tool")
        stats["model_usage"] = _read_ranked_usage(conn, ("model_usage", "models", "session_models"), label="model")
        stats["source_usage"] = _read_ranked_usage(conn, ("source_usage", "sources"), label="source")
        stats["skill_usage"] = _read_ranked_usage(conn, ("skill_usage", "skills"), label="skill")
        for key, value in _read_core_session_stats(conn).items():
            if value not in (None, "", [], {}, 0) or stats.get(key) in (None, "", [], {}, 0):
                stats[key] = value
        _record_source(sources, path, kind="state-db", details={"tables": _table_names(conn)[:12]})
    except sqlite3.Error as exc:
        warnings.append(f"unable to query state db {path}: {exc}")
    finally:
        conn.close()
    return stats


def _read_core_session_stats(conn: sqlite3.Connection) -> dict[str, Any]:
    stats: dict[str, Any] = {}
    tables = set(_table_names(conn))
    if "sessions" in tables:
        stats["session_count"] = _scalar_int(conn, "SELECT COUNT(*) FROM sessions")
        stats["tool_count"] = _scalar_int(conn, "SELECT COALESCE(SUM(tool_call_count), 0) FROM sessions")
        stats["model_count"] = _scalar_int(conn, "SELECT COUNT(DISTINCT model) FROM sessions WHERE model IS NOT NULL")
        stats["source_count"] = _scalar_int(conn, "SELECT COUNT(DISTINCT source) FROM sessions WHERE source IS NOT NULL")
        stats["cron"] = {"count": _scalar_int(conn, "SELECT COUNT(*) FROM sessions WHERE source = 'cron'")}
        stats["model_usage"] = _query_name_counts(conn, "SELECT model, COUNT(*) FROM sessions WHERE model IS NOT NULL GROUP BY model ORDER BY COUNT(*) DESC, model LIMIT 20")
        stats["source_usage"] = _query_name_counts(conn, "SELECT source, COUNT(*) FROM sessions WHERE source IS NOT NULL GROUP BY source ORDER BY COUNT(*) DESC, source LIMIT 20")
    if "messages" in tables:
        message_tool_count = _scalar_int(conn, "SELECT COUNT(*) FROM messages WHERE tool_name IS NOT NULL AND tool_name != ''")
        if message_tool_count:
            stats["tool_count"] = max(int(stats.get("tool_count") or 0), message_tool_count)
            stats["tool_usage"] = _query_name_counts(conn, "SELECT tool_name, COUNT(*) FROM messages WHERE tool_name IS NOT NULL AND tool_name != '' GROUP BY tool_name ORDER BY COUNT(*) DESC, tool_name LIMIT 30")
    return stats


def _scalar_int(conn: sqlite3.Connection, query: str) -> int:
    try:
        row = conn.execute(query).fetchone()
        return int(row[0] or 0) if row else 0
    except sqlite3.Error:
        return 0


def _query_name_counts(conn: sqlite3.Connection, query: str) -> list[dict[str, Any]]:
    try:
        rows = conn.execute(query).fetchall()
    except sqlite3.Error:
        return []
    return [{"name": _scrub_text(row[0]), "count": int(row[1] or 0)} for row in rows if row and row[0]]


def _read_cron_state(conn: sqlite3.Connection) -> dict[str, Any]:
    tables = _table_names(conn)
    result: dict[str, Any] = {"count": 0}
    for table in tables:
        if "cron" not in table.lower():
            continue
        rows = _safe_select_all(conn, table, limit=50)
        result["count"] += len(rows)
        result.setdefault("tables", []).append({"name": table, "rows": len(rows)})
    return result


def _read_gateway_state(conn: sqlite3.Connection) -> dict[str, Any]:
    tables = _table_names(conn)
    gateway: dict[str, Any] = {}
    for table in tables:
        if "gateway" not in table.lower():
            continue
        rows = _safe_select_all(conn, table, limit=10)
        if rows:
            gateway["tables"] = gateway.get("tables", []) + [{"name": table, "rows": len(rows)}]
            first = rows[0]
            if isinstance(first, dict):
                for key in ("state", "status", "startup_failed", "last_error", "provider", "model"):
                    if key in first and key not in gateway:
                        gateway[key] = _scrub_value(first.get(key))
    if "startup_failed" not in gateway:
        for table in tables:
            if "gateway" not in table.lower():
                continue
            rows = _safe_select_all(conn, table, limit=10)
            for row in rows:
                if isinstance(row, dict):
                    status = str(row.get("status", row.get("state", ""))).lower()
                    if "failed" in status:
                        gateway["startup_failed"] = True
                        break
    gateway.setdefault("startup_failed", False)
    return gateway


def _read_channel_state(conn: sqlite3.Connection) -> dict[str, Any]:
    channels: dict[str, Any] = {"count": 0}
    tables = _table_names(conn)
    for table in tables:
        if "channel" not in table.lower():
            continue
        rows = _safe_select_all(conn, table, limit=100)
        channels["count"] += len(rows)
        channels.setdefault("tables", []).append({"name": table, "rows": len(rows)})
    return channels


def _read_channels_list(conn: sqlite3.Connection) -> list[dict[str, Any]]:
    results: list[dict[str, Any]] = []
    for table in _table_names(conn):
        if "channel" not in table.lower():
            continue
        for row in _safe_select_all(conn, table, limit=200):
            if isinstance(row, dict):
                results.append(
                    {
                        "name": _scrub_text(row.get("name") or row.get("channel") or row.get("id")),
                        "state": _scrub_text(row.get("state") or row.get("status")),
                        "path": _scrub_text(row.get("path") or row.get("directory")),
                    }
                )
    return [item for item in results if item.get("name")]


def _read_named_inventory(conn: sqlite3.Connection, table_names: Iterable[str]) -> list[dict[str, Any]]:
    inventory: list[dict[str, Any]] = []
    tables = _table_names(conn)
    for table in tables:
        if not any(candidate in table.lower() for candidate in table_names):
            continue
        for row in _safe_select_all(conn, table, limit=200):
            if isinstance(row, dict):
                name = row.get("name") or row.get("id") or row.get("toolset") or row.get("profile")
                inventory.append(
                    {
                        "name": _scrub_text(name),
                        "path": _scrub_text(row.get("path") or row.get("directory")),
                        "state": _scrub_text(row.get("state") or row.get("status")),
                    }
                )
    return [item for item in inventory if item.get("name")]


def _read_ranked_usage(conn: sqlite3.Connection, table_names: Iterable[str], label: str) -> list[dict[str, Any]]:
    ranked: dict[str, int] = {}
    tables = _table_names(conn)
    for table in tables:
        if not any(candidate in table.lower() for candidate in table_names):
            continue
        rows = _safe_select_all(conn, table, limit=500)
        for row in rows:
            if not isinstance(row, dict):
                continue
            name = (
                row.get("name")
                or row.get(f"{label}_name")
                or row.get(label)
                or row.get("id")
                or row.get("source")
            )
            if not name:
                continue
            count = row.get("count") or row.get("uses") or row.get("usage_count") or row.get("n") or 1
            try:
                ranked[str(name)] = ranked.get(str(name), 0) + int(count)
            except (TypeError, ValueError):
                ranked[str(name)] = ranked.get(str(name), 0) + 1
    return [
        {"name": name, "count": count}
        for name, count in sorted(ranked.items(), key=lambda item: (-item[1], item[0].lower()))
    ]


def _safe_select_all(conn: sqlite3.Connection, table: str, limit: int = 50) -> list[dict[str, Any]]:
    try:
        cursor = conn.execute(f'SELECT * FROM "{table}" LIMIT ?', (limit,))
    except sqlite3.Error:
        return []
    rows: list[dict[str, Any]] = []
    for row in cursor.fetchall():
        if isinstance(row, sqlite3.Row):
            rows.append({key: _scrub_value(row[key]) for key in row.keys()})
    return rows


def _count_rows(conn: sqlite3.Connection, candidates: Iterable[str]) -> int:
    for table in _table_names(conn):
        if table.lower() not in {candidate.lower() for candidate in candidates} and not any(candidate.lower() == table.lower() for candidate in candidates):
            if not any(candidate in table.lower() for candidate in candidates):
                continue
        try:
            cursor = conn.execute(f'SELECT COUNT(*) FROM "{table}"')
            row = cursor.fetchone()
            if row:
                return int(row[0])
        except sqlite3.Error:
            continue
    return 0


def _table_names(conn: sqlite3.Connection) -> list[str]:
    try:
        cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")
    except sqlite3.Error:
        return []
    return [str(row[0]) for row in cursor.fetchall() if row and row[0]]


def _collect_session_files(hermes_dir: Path, state_dir: Path) -> Iterable[Path]:
    seen: set[Path] = set()
    for base in (hermes_dir, state_dir):
        if not base.exists():
            continue
        for path in _walk_files(base):
            if path.suffix != ".json" or _looks_sensitive(path):
                continue
            if "session" in path.name.lower() or "conversation" in path.name.lower() or "run" in path.name.lower():
                try:
                    resolved = path.resolve()
                except OSError:
                    continue
                if resolved not in seen:
                    seen.add(resolved)
                    yield path


def _summarize_session_file(path: Path) -> dict[str, Any] | None:
    data = _safe_json_load(path)
    if not isinstance(data, dict):
        return None
    summary: dict[str, Any] = {
        "path": str(path),
        "keys": sorted(k for k in data.keys() if not _is_secret_key(k))[:20],
    }
    if "id" in data:
        summary["id"] = _scrub_value(data.get("id"))
    if "model" in data:
        summary["model"] = _scrub_value(data.get("model"))
    if "provider" in data:
        summary["provider"] = _scrub_value(data.get("provider"))
    if isinstance(data.get("messages"), list):
        summary["messages"] = len(data["messages"])
    if isinstance(data.get("events"), list):
        summary["events"] = len(data["events"])
    if isinstance(data.get("turns"), list):
        summary["turns"] = len(data["turns"])
    summary["preview"] = _summarize_json_shape(data)
    return summary


def _count_prompt_like_artifacts(hermes_dir: Path, state_dir: Path) -> dict[str, Any]:
    counts = {"files": 0, "bytes": 0}
    for base in (hermes_dir, state_dir):
        if not base.exists():
            continue
        for path in _walk_files(base):
            if _looks_sensitive(path):
                continue
            if any(token in path.name.lower() for token in ("prompt", "instruction", "snapshot")):
                counts["files"] += 1
                try:
                    counts["bytes"] += path.stat().st_size
                except OSError:
                    continue
    return counts


def _summarize_skill_snapshot(payload: dict[str, Any]) -> list[dict[str, Any]]:
    skills: list[dict[str, Any]] = []
    for key in ("skills", "items", "entries"):
        value = payload.get(key)
        if isinstance(value, list):
            for item in value:
                if not isinstance(item, dict):
                    continue
                name = item.get("name") or item.get("skill_name") or item.get("skill") or item.get("id")
                skills.append(
                    {
                        "name": _scrub_text(name),
                        "path": _scrub_text(item.get("path")),
                        "category": _scrub_text(item.get("category")),
                        "summary": _clip_text(item.get("summary") or item.get("description")),
                        "version": _scrub_text(item.get("version")),
                    }
                )
    if not skills and isinstance(payload.get("snapshot"), list):
        for item in payload["snapshot"]:
            if isinstance(item, dict):
                skills.append({"name": _scrub_text(item.get("name")), "summary": _clip_text(item.get("summary"))})
    return [item for item in skills if item.get("name")]


def _summarize_skills_dir(skills_dir: Path) -> list[dict[str, Any]]:
    entries: list[dict[str, Any]] = []
    try:
        for path in skills_dir.iterdir():
            if _looks_sensitive(path):
                continue
            if path.is_dir():
                entries.append({"name": path.name, "path": str(path), "kind": "dir"})
            elif path.is_file():
                entries.append({"name": path.stem, "path": str(path), "kind": "file", "size": _safe_size(path)})
    except OSError:
        return entries
    return entries


def _summarize_instruction_file(path: Path) -> dict[str, Any] | None:
    text = _safe_read_text(path)
    if text is None:
        return None
    summary = {
        "path": str(path),
        "kind": _infer_instruction_kind(path, text),
        "name": path.name,
        "lines": len(text.splitlines()),
        "preview": _clip_text(_first_meaningful_lines(text)),
    }
    return summary


def _parse_yaml_summary(path: Path) -> dict[str, Any]:
    text = _safe_read_text(path)
    if text is None:
        return {}
    root: dict[str, Any] = {}
    stack: list[tuple[int, dict[str, Any]]] = [(-1, root)]
    for raw_line in text.splitlines():
        indent = len(raw_line) - len(raw_line.lstrip(" "))
        stripped = raw_line.strip()
        if not stripped or stripped.startswith("#"):
            continue
        if stripped.startswith("- "):
            # The parser only builds line-based summaries, not full YAML arrays.
            continue
        match = re.match(r"^([A-Za-z0-9_.-]+)\s*:\s*(.*)$", stripped)
        if not match:
            continue
        key, value = match.group(1), match.group(2).strip()
        while len(stack) > 1 and indent <= stack[-1][0]:
            stack.pop()
        current = stack[-1][1]
        if value == "":
            current[key] = {}
            stack.append((indent, current[key]))
            continue
        if value.lower() in {"true", "false"}:
            current[key] = value.lower() == "true"
        elif re.fullmatch(r"-?\d+", value):
            current[key] = int(value)
        elif re.fullmatch(r"-?\d+\.\d+", value):
            current[key] = float(value)
        else:
            current[key] = _scrub_text(_strip_yaml_quotes(value))
    summary = {k: v for k, v in root.items() if not _is_secret_key(k)}
    summary["_summary"] = {
        "keys": sorted(k for k in summary.keys() if k != "_summary")[:20],
        "path": str(path),
        "lines": len(text.splitlines()),
    }
    return summary


def _normalize_config(config: dict[str, Any]) -> dict[str, Any]:
    normalized = {}
    for key, value in config.items():
        if key == "_summary":
            continue
        normalized[key] = value
    flat = {}
    for key in ("checkpoints", "smart_model_routing", "browser", "model", "provider", "fallbacks"):
        if key in normalized:
            flat[key] = normalized[key]
    for key in ("default_model", "default_provider"):
        if key in normalized:
            flat[key] = normalized[key]
    if "browser" in normalized and isinstance(normalized["browser"], dict):
        browser = normalized["browser"]
        flat["browser"] = {
            "record_sessions": browser.get("record_sessions"),
        }
    for key in normalized:
        if key not in flat:
            flat[key] = normalized[key]
    return flat


def _get_bool_flag(config: dict[str, Any], dotted_key: str) -> bool | None:
    parts = dotted_key.split(".")
    current: Any = config
    for part in parts:
        if isinstance(current, dict) and part in current:
            current = current[part]
        else:
            return None
    if isinstance(current, bool):
        return current
    if isinstance(current, str):
        lowered = current.lower()
        if lowered in {"true", "yes", "on", "1"}:
            return True
        if lowered in {"false", "no", "off", "0"}:
            return False
    return None


def _find_files(base: Path, suffixes: set[str]) -> list[Path]:
    if not base.exists():
        return []
    result: list[Path] = []
    for path in _walk_files(base):
        if path.suffix in suffixes or path.name in suffixes:
            result.append(path)
    return result


def _find_first_file(bases: Iterable[Path], names: set[str]) -> Path | None:
    for base in bases:
        if not base.exists():
            continue
        for name in names:
            candidate = base / name
            if candidate.is_file():
                return candidate
        for path in _walk_files(base):
            if path.name in names:
                return path
    return None


def _find_first_dir(bases: Iterable[Path], names: set[str]) -> Path | None:
    for base in bases:
        if not base.exists():
            continue
        for name in names:
            candidate = base / name
            if candidate.is_dir():
                return candidate
        for root, dirs, _files in os.walk(base):
            dirs[:] = [name for name in dirs if not _skip_dir_name(name)]
            for dirname in dirs:
                if dirname in names:
                    return Path(root) / dirname
    return None


def _walk_files(base: Path) -> Iterable[Path]:
    try:
        for root, dirs, files in os.walk(base):
            dirs[:] = [name for name in dirs if not _skip_dir_name(name)]
            for filename in files:
                path = Path(root) / filename
                if not _skip_path(path):
                    yield path
    except OSError:
        return


def _skip_dir_name(name: str) -> bool:
    return name in SKIP_DIR_NAMES or name.endswith(".egg-info")


def _skip_path(path: Path) -> bool:
    return any(_skip_dir_name(part) for part in path.parts)


def _safe_json_load(path: Path) -> Any | None:
    text = _safe_read_text(path)
    if text is None:
        return None
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return None


def _safe_read_text(path: Path) -> str | None:
    try:
        return path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return None


def _safe_size(path: Path) -> int | None:
    try:
        return path.stat().st_size
    except OSError:
        return None


def _record_source(
    sources: list[dict[str, Any]],
    path: Path,
    *,
    kind: str,
    details: dict[str, Any] | None = None,
) -> None:
    sources.append(
        {
            "path": str(path),
            "kind": kind,
            "size": _safe_size(path),
            "details": details or {},
        }
    )


def _looks_sensitive(path: Path) -> bool:
    name = path.name.lower()
    return any(token in name for token in (".env", "token", "secret", "auth", "cookie", "credential", "password"))


def _is_secret_key(key: str) -> bool:
    lowered = key.lower()
    return any(token in lowered for token in ("secret", "token", "password", "cookie", "auth", "apikey", "api_key"))


def _scrub_value(value: Any) -> Any:
    if isinstance(value, dict):
        return {k: _scrub_value(v) for k, v in value.items() if not _is_secret_key(str(k))}
    if isinstance(value, list):
        return [_scrub_value(v) for v in value[:MAX_SAMPLE_ITEMS]]
    if isinstance(value, str):
        return _scrub_text(value)
    return value


def _scrub_text(value: Any) -> Any:
    if value is None:
        return None
    text = str(value)
    if _is_secret_key(text) or _looks_like_secret_value(text):
        return "[redacted]"
    return _clip_text(text)


def _clip_text(text: str) -> str:
    text = text.strip()
    if len(text) <= MAX_TEXT_PREVIEW:
        return text
    return text[: MAX_TEXT_PREVIEW - 3].rstrip() + "..."


def _first_meaningful_lines(text: str, limit: int = 8) -> str:
    lines = []
    for raw in text.splitlines():
        stripped = raw.strip()
        if not stripped:
            continue
        if stripped.startswith("#"):
            continue
        if _is_secret_key(stripped):
            continue
        lines.append(stripped)
        if len(lines) >= limit:
            break
    return "\n".join(lines)


def _infer_instruction_kind(path: Path, text: str) -> str:
    name = path.name.lower()
    if "system" in name:
        return "system"
    if "instruction" in name:
        return "instruction"
    if "prompt" in name:
        return "prompt"
    if path.suffix in {".yaml", ".yml"}:
        return "config"
    if "skill" in name:
        return "skill"
    return "text"


def _summarize_json_dict(data: dict[str, Any]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in data.items():
        if _is_secret_key(str(key)):
            continue
        if key in {"instruction", "prompt", "system_prompt", "request", "messages", "events", "conversation"}:
            continue
        if isinstance(value, dict):
            result[key] = _summarize_json_dict(value)
        elif isinstance(value, list):
            result[key] = _summarize_json_list(value)
        else:
            result[key] = _scrub_value(value)
    return result


def _summarize_json_list(values: list[Any]) -> list[Any]:
    summarized = []
    for item in values[:MAX_SAMPLE_ITEMS]:
        if isinstance(item, dict):
            summarized.append(_summarize_json_dict(item))
        elif isinstance(item, list):
            summarized.append(_summarize_json_list(item))
        else:
            summarized.append(_scrub_value(item))
    return summarized


def _summarize_json_shape(data: dict[str, Any]) -> dict[str, Any]:
    return {
        "keys": sorted(k for k in data.keys() if not _is_secret_key(k))[:20],
        "lists": sorted(k for k, v in data.items() if isinstance(v, list) and not _is_secret_key(k))[:20],
        "dicts": sorted(k for k, v in data.items() if isinstance(v, dict) and not _is_secret_key(k))[:20],
    }


def _dedupe_records(items: list[dict[str, Any]], *, key_fields: tuple[str, ...]) -> list[dict[str, Any]]:
    seen: set[tuple[Any, ...]] = set()
    deduped: list[dict[str, Any]] = []
    for item in items:
        key = tuple(item.get(field) for field in key_fields)
        if key in seen:
            continue
        seen.add(key)
        deduped.append(item)
    return deduped


def _merge_nested(target: dict[str, Any], source: dict[str, Any]) -> None:
    for key, value in source.items():
        if key == "_summary":
            continue
        if key not in target:
            target[key] = value
            continue
        if isinstance(target[key], dict) and isinstance(value, dict):
            _merge_nested(target[key], value)
        elif isinstance(target[key], list) and isinstance(value, list):
            target[key].extend(v for v in value if v not in target[key])


def _normalize_value(value: Any) -> Any:
    if isinstance(value, dict):
        return {k: _normalize_value(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_normalize_value(v) for v in value]
    return value


def _strip_yaml_quotes(value: str) -> str:
    if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}:
        return value[1:-1]
    return value


def _looks_like_secret_value(text: str) -> bool:
    lowered = text.lower().strip()
    if not lowered:
        return False
    if any(token in lowered for token in ("bearer ", "api_key", "access_token", "refresh_token", "secret", "password")):
        return True
    if "private key" in lowered or "BEGIN PRIVATE KEY" in text:
        return True
    if len(text) >= 24 and " " not in text and any(ch.isdigit() for ch in text) and any(ch.isalpha() for ch in text):
        if re.fullmatch(r"[A-Za-z0-9._\-+/=]+", text):
            return True
    return False


def _dedupe_preserve_order(values: list[str]) -> list[str]:
    seen = set()
    result = []
    for value in values:
        if value in seen:
            continue
        seen.add(value)
        result.append(value)
    return result