sidebyside/scripts/extract_codex.py

"""Safe Codex data extraction helpers.

This module reads local metadata under ``~/.codex`` and returns a structured,
side-effect-free summary. It avoids secrets, does not touch ``auth.json``, and
is defensive around missing files or tables.
"""

from __future__ import annotations

import json
import sqlite3
from pathlib import Path
from typing import Any

try:  # Python 3.11+
    import tomllib  # type: ignore[attr-defined]
except ImportError:  # pragma: no cover
    tomllib = None  # type: ignore[assignment]

__all__ = ["extract_codex"]

SNIPPET_LIMIT = 400


def extract_codex(home: Path) -> dict:
    """Extract a safe, summary-only Codex profile from ``home/.codex``."""

    codex = home / ".codex"
    status, config, instruction_layers, extensions, usage, discoveries, sources, warnings = (
        {},
        {},
        [],
        {"skills": [], "plugins": [], "mcp_servers": []},
        {},
        [],
        [],
        [],
    )

    config_path = codex / "config.toml"
    version_path = codex / "version.json"
    history_path = codex / "history.jsonl"
    state_db_path = codex / "state_5.sqlite"
    logs_db_path = codex / "logs_2.sqlite"
    marketplace_path = codex / ".tmp" / "plugins" / ".agents" / "plugins" / "marketplace.json"
    skills_dir = codex / "skills"
    hooks_dir = codex / "hooks"
    memories_dir = codex / "memories"
    sessions_dir = codex / "sessions"

    config_data = _load_toml(config_path, warnings, sources)
    version_data = _load_json(version_path, warnings, sources)

    state_summary = _extract_state(state_db_path, warnings, sources)
    logs_summary = _extract_logs(logs_db_path, warnings, sources)
    history_summary = _extract_history(history_path, warnings, sources)
    skill_entries = _extract_skills(skills_dir, warnings, sources)
    plugin_entries, marketplace_entries = _extract_plugins(config_data, codex, marketplace_path, warnings, sources)
    mcp_entries = _extract_mcp_servers(config_data, sources)
    filesystem_summary = _filesystem_counts(
        hooks_dir=hooks_dir,
        memories_dir=memories_dir,
        sessions_dir=sessions_dir,
        warnings=warnings,
        sources=sources,
    )

    config = {
        "model": config_data.get("model"),
        "model_reasoning_effort": config_data.get("model_reasoning_effort"),
        "projects": _summarize_projects(config_data.get("projects", {})),
        "notice": _summarize_notice(config_data.get("notice", {})),
        "plugins_enabled": [entry["name"] for entry in plugin_entries],
        "plugin_count": len(plugin_entries),
        "mcp_server_count": len(mcp_entries),
    }
    if version_data:
        config["version"] = version_data

    instruction_layers.extend(
        [
            {
                "layer": "user config",
                "path": str(config_path),
                "summary": _snippet(
                    f"Model {config_data.get('model') or 'unknown'} with reasoning "
                    f"{config_data.get('model_reasoning_effort') or 'unknown'}; "
                    f"{len(plugin_entries)} enabled plugin(s); {len(mcp_entries)} MCP server(s)."
                ),
                "confidence": "high",
            },
            {
                "layer": "state database",
                "path": str(state_db_path),
                "summary": _snippet(
                    f"{state_summary.get('thread_count', 0)} threads, "
                    f"{state_summary.get('spawn_edge_count', 0)} spawn edges, "
                    f"{state_summary.get('dynamic_tool_count', 0)} dynamic tools, "
                    f"{state_summary.get('agent_job_count', 0)} agent jobs."
                ),
                "confidence": "high",
            },
            {
                "layer": "history",
                "path": str(history_path),
                "summary": _snippet(history_summary.get("summary", "No history data found.")),
                "confidence": history_summary.get("confidence", "low"),
            },
            {
                "layer": "plugin catalog",
                "path": str(marketplace_path),
                "summary": _snippet(
                    f"{len(marketplace_entries)} available marketplace plugin candidates, "
                    f"including {', '.join(marketplace_entries[:6]) if marketplace_entries else 'none'}."
                ),
                "confidence": "medium",
            },
        ]
    )

    extensions["skills"] = skill_entries
    extensions["plugins"] = plugin_entries
    extensions["mcp_servers"] = mcp_entries

    usage = {
        "session_count": filesystem_summary.get("session_file_count", 0),
        "thread_count": state_summary.get("thread_count", 0),
        "spawn_edge_count": state_summary.get("spawn_edge_count", 0),
        "tool_evidence_count": logs_summary.get("log_row_count", 0),
        "history_entry_count": history_summary.get("entry_count", 0),
        "history_session_count": history_summary.get("session_count", 0),
        "log_thread_count": logs_summary.get("thread_count", 0),
        "latest_threads": state_summary.get("latest_threads", []),
        "latest_history": history_summary.get("latest_entries", []),
        "log_target_counts": logs_summary.get("top_targets", []),
    }

    discoveries.extend(_discoveries(
        filesystem_summary=filesystem_summary,
        state_summary=state_summary,
        logs_summary=logs_summary,
        marketplace_entries=marketplace_entries,
    ))

    status = {
        "codex_version": version_data.get("latest_version"),
        "last_checked_at": version_data.get("last_checked_at"),
        "model": config_data.get("model"),
        "reasoning_effort": config_data.get("model_reasoning_effort"),
        "thread_count": state_summary.get("thread_count", 0),
        "session_count": filesystem_summary.get("session_file_count", 0),
        "log_rows": logs_summary.get("log_row_count", 0),
        "log_threads": logs_summary.get("thread_count", 0),
        "history_entries": history_summary.get("entry_count", 0),
        "history_sessions": history_summary.get("session_count", 0),
        "hooks_files": filesystem_summary.get("hooks_file_count", 0),
        "memories_files": filesystem_summary.get("memories_file_count", 0),
        "spawn_edges": state_summary.get("spawn_edge_count", 0),
        "dynamic_tools": state_summary.get("dynamic_tool_count", 0),
        "agent_jobs": state_summary.get("agent_job_count", 0),
    }

    sources.extend(
        [
            _source_entry(config_path, "config", "high"),
            _source_entry(version_path, "version", "high"),
            _source_entry(state_db_path, "sqlite", "high"),
            _source_entry(logs_db_path, "sqlite", "high"),
            _source_entry(history_path, "jsonl", "high"),
            _source_entry(marketplace_path, "json", "high"),
            _source_entry(skills_dir, "directory", "medium"),
            _source_entry(hooks_dir, "directory", "high"),
            _source_entry(memories_dir, "directory", "high"),
            _source_entry(sessions_dir, "directory", "high"),
            _source_entry(codex / ".tmp" / "plugins" / "plugins" / "github" / ".codex-plugin" / "plugin.json", "plugin", "high"),
        ]
    )

    return {
        "id": "codex",
        "name": "Codex",
        "status": status,
        "config": config,
        "instruction_layers": instruction_layers,
        "extensions": extensions,
        "usage": usage,
        "discoveries": discoveries,
        "sources": sources,
        "warnings": warnings,
    }


def _load_toml(path: Path, warnings: list[str], sources: list[dict[str, Any]]) -> dict[str, Any]:
    if not path.exists():
        warnings.append(f"Missing config: {path}")
        return {}
    try:
        if tomllib is None:
            warnings.append(f"tomllib unavailable; skipped TOML parse for {path}")
            return {}
        with path.open("rb") as fh:
            data = tomllib.load(fh)
        _append_source(sources, path, "config", "high")
        return data
    except Exception as exc:  # noqa: BLE001
        warnings.append(f"Failed to parse TOML {path}: {exc}")
        return {}


def _load_json(path: Path, warnings: list[str], sources: list[dict[str, Any]]) -> dict[str, Any]:
    if not path.exists():
        warnings.append(f"Missing JSON: {path}")
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        _append_source(sources, path, "json", "high")
        return data if isinstance(data, dict) else {}
    except Exception as exc:  # noqa: BLE001
        warnings.append(f"Failed to parse JSON {path}: {exc}")
        return {}


def _connect_sqlite(path: Path, warnings: list[str]) -> sqlite3.Connection | None:
    if not path.exists():
        warnings.append(f"Missing sqlite database: {path}")
        return None
    try:
        conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
        conn.row_factory = sqlite3.Row
        return conn
    except Exception as exc:  # noqa: BLE001
        warnings.append(f"Failed to open sqlite database {path}: {exc}")
        return None


def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
    row = conn.execute(
        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1",
        (table,),
    ).fetchone()
    return row is not None


def _extract_state(path: Path, warnings: list[str], sources: list[dict[str, Any]]) -> dict[str, Any]:
    conn = _connect_sqlite(path, warnings)
    if conn is None:
        return {}
    try:
        if not _table_exists(conn, "threads"):
            warnings.append(f"Missing table threads in {path}")
            return {}

        thread_count = conn.execute("SELECT COUNT(*) AS n FROM threads").fetchone()["n"]
        spawn_edge_count = _count_if_table(conn, "thread_spawn_edges")
        dynamic_tool_count = _count_if_table(conn, "thread_dynamic_tools")
        agent_job_count = _count_if_table(conn, "agent_jobs")
        agent_job_item_count = _count_if_table(conn, "agent_job_items")
        job_count = _count_if_table(conn, "jobs")

        latest_threads = _fetch_rows(
            conn,
            """
            SELECT id, title, cwd, model, reasoning_effort, updated_at
            FROM threads
            ORDER BY updated_at DESC
            LIMIT 5
            """,
        )

        _append_source(sources, path, "sqlite", "high")
        return {
            "thread_count": thread_count,
            "spawn_edge_count": spawn_edge_count,
            "dynamic_tool_count": dynamic_tool_count,
            "agent_job_count": agent_job_count,
            "agent_job_item_count": agent_job_item_count,
            "job_count": job_count,
            "latest_threads": [
                {
                    "id": row["id"],
                    "title": _snippet(row["title"]),
                    "cwd": row["cwd"],
                    "model": row["model"],
                    "reasoning_effort": row["reasoning_effort"],
                    "updated_at": row["updated_at"],
                }
                for row in latest_threads
            ],
        }
    finally:
        conn.close()


def _extract_logs(path: Path, warnings: list[str], sources: list[dict[str, Any]]) -> dict[str, Any]:
    conn = _connect_sqlite(path, warnings)
    if conn is None:
        return {}
    try:
        if not _table_exists(conn, "logs"):
            warnings.append(f"Missing table logs in {path}")
            return {}

        log_row_count = conn.execute("SELECT COUNT(*) AS n FROM logs").fetchone()["n"]
        thread_count = conn.execute("SELECT COUNT(DISTINCT thread_id) AS n FROM logs").fetchone()["n"]
        top_targets = _fetch_rows(
            conn,
            """
            SELECT target, COUNT(*) AS n
            FROM logs
            GROUP BY target
            ORDER BY n DESC, target
            LIMIT 10
            """,
        )

        _append_source(sources, path, "sqlite", "high")
        return {
            "log_row_count": log_row_count,
            "thread_count": thread_count,
            "top_targets": [
                {"target": row["target"], "count": row["n"]} for row in top_targets
            ],
        }
    finally:
        conn.close()


def _extract_history(path: Path, warnings: list[str], sources: list[dict[str, Any]]) -> dict[str, Any]:
    if not path.exists():
        warnings.append(f"Missing history: {path}")
        return {}

    entry_count = 0
    sessions: set[str] = set()
    latest_entries: list[dict[str, Any]] = []
    try:
        from collections import deque

        latest_buffer: deque[dict[str, Any]] = deque(maxlen=5)
        with path.open("r", encoding="utf-8") as fh:
            for line in fh:
                text = line.strip()
                if not text:
                    continue
                entry_count += 1
                parsed = _safe_json_line(text)
                if not parsed:
                    continue
                session_id = parsed.get("session_id")
                if isinstance(session_id, str):
                    sessions.add(session_id)
                latest_buffer.append(
                    {
                        "session_id": session_id,
                        "ts": parsed.get("ts"),
                        "summary": _snippet(_history_summary_text(parsed)),
                    }
                )
        latest_entries = list(latest_buffer)
        _append_source(sources, path, "jsonl", "high")
        return {
            "entry_count": entry_count,
            "session_count": len(sessions),
            "latest_entries": latest_entries,
            "summary": f"{entry_count} history entries across {len(sessions)} sessions.",
            "confidence": "high" if entry_count else "low",
        }
    except Exception as exc:  # noqa: BLE001
        warnings.append(f"Failed to read history {path}: {exc}")
        return {}


def _extract_skills(skills_dir: Path, warnings: list[str], sources: list[dict[str, Any]]) -> list[dict[str, Any]]:
    if not skills_dir.exists():
        warnings.append(f"Missing skills directory: {skills_dir}")
        return []

    entries: list[dict[str, Any]] = []
    for skill_md in sorted(skills_dir.rglob("SKILL.md")):
        try:
            text = skill_md.read_text(encoding="utf-8")
            meta = _parse_front_matter(text)
            description = meta.get("description") or _first_nonempty_line_after_front_matter(text)
            entries.append(
                {
                    "name": meta.get("name") or skill_md.parent.name,
                    "path": str(skill_md),
                    "summary": _snippet(description),
                    "source": "local-skill",
                }
            )
            _append_source(sources, skill_md, "skill", "high")
        except Exception as exc:  # noqa: BLE001
            warnings.append(f"Failed to read skill {skill_md}: {exc}")
    return entries


def _extract_plugins(
    config_data: dict[str, Any],
    codex: Path,
    marketplace_path: Path,
    warnings: list[str],
    sources: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[str]]:
    plugins_cfg = config_data.get("plugins", {})
    enabled = []
    if isinstance(plugins_cfg, dict):
        for name, item in sorted(plugins_cfg.items()):
            if isinstance(item, dict) and item.get("enabled"):
                enabled.append(name)

    plugin_entries: list[dict[str, Any]] = []
    for name in enabled:
        plugin_slug = name.split("@", 1)[0]
        plugin_path = codex / ".tmp" / "plugins" / "plugins" / plugin_slug / ".codex-plugin" / "plugin.json"
        manifest = _load_json(plugin_path, warnings, sources)
        description = manifest.get("description") if isinstance(manifest, dict) else ""
        interface = manifest.get("interface", {}) if isinstance(manifest, dict) else {}
        plugin_entries.append(
            {
                "name": plugin_slug,
                "config_key": name,
                "enabled": True,
                "path": str(plugin_path),
                "display_name": interface.get("displayName") if isinstance(interface, dict) else None,
                "summary": _snippet(description),
            }
        )

    marketplace_entries = _marketplace_plugins(marketplace_path, warnings, sources)
    return plugin_entries, marketplace_entries


def _marketplace_plugins(path: Path, warnings: list[str], sources: list[dict[str, Any]]) -> list[str]:
    if not path.exists():
        warnings.append(f"Missing marketplace catalog: {path}")
        return []
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        _append_source(sources, path, "json", "high")
        plugins = []
        for item in data.get("plugins", []):
            if isinstance(item, dict) and isinstance(item.get("name"), str):
                plugins.append(item["name"])
        return plugins
    except Exception as exc:  # noqa: BLE001
        warnings.append(f"Failed to parse marketplace catalog {path}: {exc}")
        return []


def _extract_mcp_servers(config_data: dict[str, Any], sources: list[dict[str, Any]]) -> list[dict[str, Any]]:
    mcp_servers = config_data.get("mcp_servers", {})
    if not isinstance(mcp_servers, dict):
        return []

    entries: list[dict[str, Any]] = []
    for name, data in sorted(mcp_servers.items()):
        if not isinstance(data, dict):
            continue
        entry: dict[str, Any] = {"name": name}
        if "command" in data:
            entry["type"] = "local"
            entry["command"] = data.get("command")
            entry["args"] = data.get("args", [])
            entry["cwd"] = data.get("cwd")
        elif "url" in data:
            entry["type"] = "remote"
            entry["url"] = data.get("url")
        entry["summary"] = _snippet(
            f"{entry.get('type', 'unknown')} MCP server {name}."
        )
        entries.append(entry)
    return entries


def _filesystem_counts(
    *,
    hooks_dir: Path,
    memories_dir: Path,
    sessions_dir: Path,
    warnings: list[str],
    sources: list[dict[str, Any]],
) -> dict[str, Any]:
    hooks_file_count = _count_files(hooks_dir)
    memories_file_count = _count_files(memories_dir)
    session_file_count = _count_files(sessions_dir)

    if hooks_dir.exists():
        _append_source(sources, hooks_dir, "directory", "high")
    else:
        warnings.append(f"Missing hooks directory: {hooks_dir}")
    if memories_dir.exists():
        _append_source(sources, memories_dir, "directory", "high")
    else:
        warnings.append(f"Missing memories directory: {memories_dir}")
    if sessions_dir.exists():
        _append_source(sources, sessions_dir, "directory", "high")
    else:
        warnings.append(f"Missing sessions directory: {sessions_dir}")

    return {
        "hooks_file_count": hooks_file_count,
        "memories_file_count": memories_file_count,
        "session_file_count": session_file_count,
    }


def _discoveries(
    *,
    filesystem_summary: dict[str, Any],
    state_summary: dict[str, Any],
    logs_summary: dict[str, Any],
    marketplace_entries: list[str],
) -> list[dict[str, Any]]:
    discoveries: list[dict[str, Any]] = []

    if filesystem_summary.get("memories_file_count", 0) == 0:
        discoveries.append(
            {
                "kind": "memory-store",
                "path": "~/.codex/memories",
                "status": "empty",
                "candidate": "Add or populate Codex memories for long-lived project facts.",
                "confidence": "high",
            }
        )
    if filesystem_summary.get("hooks_file_count", 0) == 0:
        discoveries.append(
            {
                "kind": "hooks",
                "path": "~/.codex/hooks",
                "status": "empty",
                "candidate": "Add Codex hook configuration or hook scripts.",
                "confidence": "high",
            }
        )
    if filesystem_summary.get("session_file_count", 0) == 0:
        discoveries.append(
            {
                "kind": "sessions",
                "path": "~/.codex/sessions",
                "status": "empty",
                "candidate": "Start a session to populate the Codex session store.",
                "confidence": "high",
            }
        )

    discoveries.append(
        {
            "kind": "plugin",
            "name": "github",
            "status": "enabled",
            "candidate": "GitHub plugin is enabled in local config.",
            "confidence": "high",
        }
    )

    if marketplace_entries:
        discoveries.append(
            {
                "kind": "plugin",
                "name": "Codex plugin marketplace",
                "status": "available",
                "candidate": (
                    f"{len(marketplace_entries)} marketplace plugin(s) are available but not enabled. "
                    f"Examples: {', '.join(marketplace_entries[:12])}."
                ),
                "items": marketplace_entries[:40],
                "count": len(marketplace_entries),
                "confidence": "medium",
            }
        )

    if state_summary.get("dynamic_tool_count", 0) == 0:
        discoveries.append(
            {
                "kind": "database",
                "table": "thread_dynamic_tools",
                "status": "empty",
                "candidate": "No dynamic tools are recorded; this table is a candidate for future tool evidence.",
                "confidence": "high",
            }
        )
    if state_summary.get("agent_job_count", 0) == 0:
        discoveries.append(
            {
                "kind": "database",
                "table": "agent_jobs",
                "status": "empty",
                "candidate": "No agent jobs are recorded; job scaffolding is present but unused.",
                "confidence": "high",
            }
        )
    if logs_summary.get("log_row_count", 0) == 0:
        discoveries.append(
            {
                "kind": "logs",
                "path": "~/.codex/logs_2.sqlite",
                "status": "empty",
                "candidate": "No log evidence found in logs_2.sqlite.",
                "confidence": "medium",
            }
        )
    return discoveries


def _summarize_projects(projects: dict[str, Any]) -> list[dict[str, Any]]:
    if not isinstance(projects, dict):
        return []
    entries = []
    for path, meta in sorted(projects.items()):
        if not isinstance(meta, dict):
            continue
        entries.append(
            {
                "path": path,
                "trust_level": meta.get("trust_level"),
            }
        )
    return entries[:30]


def _summarize_notice(notice: dict[str, Any]) -> dict[str, Any]:
    if not isinstance(notice, dict):
        return {}
    return {
        "model_migrations": notice.get("model_migrations", {}),
    }


def _fetch_rows(conn: sqlite3.Connection, query: str, params: tuple[Any, ...] = ()) -> list[sqlite3.Row]:
    try:
        return list(conn.execute(query, params).fetchall())
    except sqlite3.Error:
        return []


def _count_if_table(conn: sqlite3.Connection, table: str) -> int:
    if not _table_exists(conn, table):
        return 0
    try:
        return int(conn.execute(f"SELECT COUNT(*) AS n FROM {table}").fetchone()["n"])
    except sqlite3.Error:
        return 0


def _count_files(path: Path) -> int:
    if not path.exists():
        return 0
    try:
        return sum(1 for item in path.rglob("*") if item.is_file())
    except Exception:  # noqa: BLE001
        return 0


def _safe_json_line(line: str) -> dict[str, Any] | None:
    try:
        data = json.loads(line)
        return data if isinstance(data, dict) else None
    except Exception:  # noqa: BLE001
        return None


def _history_summary_text(item: dict[str, Any]) -> str:
    text = item.get("text")
    if isinstance(text, str) and text.strip():
        return text
    for key in ("prompt", "message", "content"):
        value = item.get(key)
        if isinstance(value, str) and value.strip():
            return value
    return json.dumps({k: v for k, v in item.items() if k != "text"}, ensure_ascii=True)[:SNIPPET_LIMIT]


def _parse_front_matter(text: str) -> dict[str, str]:
    if not text.startswith("---\n"):
        return {}
    parts = text.split("---\n", 2)
    if len(parts) < 3:
        return {}
    meta_text = parts[1]
    meta: dict[str, str] = {}
    for line in meta_text.splitlines():
        if ":" not in line:
            continue
        key, value = line.split(":", 1)
        meta[key.strip()] = value.strip().strip('"')
    return meta


def _first_nonempty_line_after_front_matter(text: str) -> str:
    body = text
    if text.startswith("---\n"):
        parts = text.split("---\n", 2)
        if len(parts) >= 3:
            body = parts[2]
    for line in body.splitlines():
        stripped = line.strip()
        if stripped:
            return stripped
    return ""


def _snippet(text: Any, limit: int = SNIPPET_LIMIT) -> str:
    if text is None:
        return ""
    if not isinstance(text, str):
        text = str(text)
    compact = " ".join(text.split())
    if len(compact) <= limit:
        return compact
    return compact[: max(0, limit - 1)].rstrip() + "…"


def _append_source(sources: list[dict[str, Any]], path: Path, kind: str, confidence: str) -> None:
    entry = {
        "path": str(path),
        "kind": kind,
        "confidence": confidence,
    }
    if entry not in sources:
        sources.append(entry)


def _source_entry(path: Path, kind: str, confidence: str) -> dict[str, Any]:
    return {"path": str(path), "kind": kind, "confidence": confidence}