| """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: |
| import tomllib |
| except ImportError: |
| tomllib = None |
|
|
| __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: |
| 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: |
| 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: |
| 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: |
| 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: |
| 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: |
| 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: |
| 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: |
| 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} |
|
|
