"""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:
"""Asynchronously read a documentation file's full UTF-8 text.
Performs a non-blocking existence check by offloading
:func:`os.path.exists` to a worker thread via :func:`asyncio.to_thread`,
returning a human-readable ``"Error: file not found ..."`` string (rather
than raising) when the path is absent. When the file exists it is opened
with :mod:`aiofiles` and read in full as UTF-8 so the event loop is never
blocked on disk I/O. This performs read-only filesystem access and has no
other side effects (no Redis, knowledge-graph, LLM, or network calls).
This is called by :func:`run` in this module to serve the main overview
and the per-section docs. Note that an unrelated, identically named
``_read_file`` exists in ``tools/security_tools.py``; the calls there
refer to that module's function, not this one.
Args:
path: Absolute path to the documentation file to read.
Returns:
str: The file's decoded contents, or an ``"Error: file not found at
{path}"`` message when the path does not exist.
Raises:
OSError: If the file exists but cannot be opened or read (e.g. a
permissions error or a decode failure surfaced by the open).
"""
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 Stargazer's own documentation for the requested section.
Entry point for the ``read_own_docs`` tool, letting the bot read its own
architecture graphs, reference docs, and recent git history at runtime.
With no *section* it returns the top-level self-doc overview
(``stargazer-docs.md``); ``"list"`` enumerates the available sections;
``"commits"`` shows the latest git log; any other value is looked up in the
module-level ``_SECTIONS`` map and the corresponding doc file is read.
Section docs are read via :func:`_read_file` (async, ``aiofiles``-backed),
and ``"commits"`` shells out to ``git log --oneline -5`` through
``subprocess.run`` offloaded with ``asyncio.to_thread`` so the event loop is
not blocked. All paths are resolved relative to the repository root
(``_PROJECT_ROOT``). Read-only: no Redis, knowledge-graph, LLM, or network
access. This module-level ``run`` is discovered and dispatched by the tool
runtime via ``tool_loader.py`` (``getattr(module, "run")``); no other
in-repo callers.
Args:
section (str): Section selector — empty for the overview, ``"list"`` for
the index, ``"commits"`` for git history, or a key from
``_SECTIONS``. Case-insensitive and whitespace-trimmed.
Returns:
str: The requested document text, the section index, the recent commit
list, or a human-readable error/"unknown section" message. Errors are
returned as strings rather than raised.
"""
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))