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