Source code for tools.read_own_docs

"""Read the bot's own documentation, architecture graphs, and recent history."""

import asyncio
import os
import subprocess

import aiofiles

_PROJECT_ROOT = os.path.normpath(
    os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."),
)

# (relative_path, one-line description)
_SECTIONS: dict[str, tuple[str, str]] = {
    # Dataflow graphs
    "system_initialization": (
        "docs/graphs/system_initialization.md",
        "Startup and shutdown sequence diagram",
    ),
    "platform_adapters": (
        "docs/graphs/platform_adapters.md",
        "Platform adapter class hierarchy, message flow, and lifecycle",
    ),
    "message_pipeline": (
        "docs/graphs/message_pipeline.md",
        "End-to-end message processing pipeline with parallel pre-inference",
    ),
    "tool_calling_loop": (
        "docs/graphs/tool_calling_loop.md",
        "Iterative LLM tool-use loop and safety limits",
    ),
    "background_tasks": (
        "docs/graphs/background_tasks.md",
        "Background scheduler task schedule and on-demand agents",
    ),
    "memory_and_rag": (
        "docs/graphs/memory_and_rag.md",
        "Knowledge graph retrieval tiers and RAG auto-search flow",
    ),
    # Reference docs
    "features": (
        "docs/FEATURES.md",
        "Complete feature and tool catalog",
    ),
    "setup": (
        "docs/SETUP.md",
        "Installation, configuration, and first-run guide",
    ),
    "latency": (
        "docs/LATENCY_OPTIMIZATIONS.md",
        "Latency analysis and optimization recommendations",
    ),
    "ncm": (
        "docs/LIMBIC_SYSTEM_AND_NCM.md",
        "Limbic system and neurochemical model technical reference",
    ),
    "changelog": (
        "docs/CHANGELOG_V3.md",
        "What changed from the old codebase to v3",
    ),
    "migration": (
        "docs/MIGRATION_GAPS.md",
        "Functionality not yet ported from the old codebase",
    ),
}

TOOL_NAME = "read_own_docs"
TOOL_DESCRIPTION = (
    "Read Stargazer's own documentation. With no arguments returns the main "
    "self-doc overview. Pass a section name to read a specific architecture "
    "graph, reference doc, or the latest git commits. Use section='list' to "
    "see all available sections."
)
TOOL_PARAMETERS = {
    "type": "object",
    "properties": {
        "section": {
            "type": "string",
            "description": (
                "Which section to read. Omit for the main overview "
                "(stargazer-docs.md). Use 'list' to see available sections, "
                "or 'commits' for the latest 5 git commits."
            ),
            "enum": [
                "list",
                "commits",
                *sorted(_SECTIONS),
            ],
        },
    },
}


async def _read_file(path: str) -> str:
    if not await asyncio.to_thread(os.path.exists, path):
        return f"Error: file not found at {path}"
    async with aiofiles.open(path, "r", encoding="utf-8") as f:
        return await f.read()


[docs] async def run(section: str = "", **_kwargs) -> str: """Return documentation content for the requested section.""" section = (section or "").strip().lower() if not section: return await _read_file(os.path.join(_PROJECT_ROOT, "stargazer-docs.md")) if section == "list": lines = ["Available sections for read_own_docs:\n"] lines.append(" (no section) — Main self-doc overview (stargazer-docs.md)") lines.append(" commits — Latest 5 git commits") for name, (_, desc) in sorted(_SECTIONS.items()): lines.append(f" {name:24s}{desc}") return "\n".join(lines) if section == "commits": try: result = await asyncio.to_thread( subprocess.run, ["git", "log", "--oneline", "-5"], cwd=_PROJECT_ROOT, capture_output=True, text=True, timeout=5, ) if result.returncode == 0 and result.stdout.strip(): return f"Latest 5 commits:\n{result.stdout.strip()}" return "Could not retrieve git log." except Exception as exc: return f"Error reading git log: {exc}" entry = _SECTIONS.get(section) if entry is None: return ( f"Unknown section '{section}'. " "Use section='list' to see available options." ) rel_path, _ = entry return await _read_file(os.path.join(_PROJECT_ROOT, rel_path))