linexcel.aidoc

linexcel.aidoc

AI-generated documentation for Excel calculations.

Vendor-neutral by construction: no provider is named in the code and none is chosen for you. There are exactly two ways in:

• base_url= — any OpenAI-compatible endpoint (a local Ollama, vLLM or LM Studio runtime; a hosted gateway such as OpenRouter; OpenAI itself; anything else that speaks /chat/completions)
• provider= — your own callable or :class:LLMProvider object, for an API that speaks something else entirely

Each node is presented with deterministic evidence from the graph and source metadata. Prompts ask the model to cite this evidence; generated claims still require review and are not a proof of calculation correctness.

TokenUsage dataclass

Tokens consumed by one or more documentation requests.

estimated is True as soon as any request in the tally had to be approximated by :func:estimate_tokens instead of being reported by the provider — treat such a total as an order of magnitude, not a bill.

Source code in src/linexcel/aidoc.py

@dataclass
class TokenUsage:
    """Tokens consumed by one or more documentation requests.

    ``estimated`` is ``True`` as soon as any request in the tally had to be
    approximated by :func:`estimate_tokens` instead of being reported by the
    provider — treat such a total as an order of magnitude, not a bill.
    """

    input_tokens: int = 0
    output_tokens: int = 0
    requests: int = 0
    estimated: bool = False
    model: str = ""
    provider: str = ""

    @property
    def total(self) -> int:
        return self.input_tokens + self.output_tokens

    def add(self, other: TokenUsage) -> None:
        """Accumulate ``other`` in place, keeping the model/provider labels."""
        self.input_tokens += other.input_tokens
        self.output_tokens += other.output_tokens
        self.requests += other.requests
        self.estimated = self.estimated or other.estimated
        self.model = self.model or other.model
        self.provider = self.provider or other.provider

    def __str__(self) -> str:
        about = "~" if self.estimated else ""
        where = f" [{self.provider}/{self.model}]" if self.provider else ""
        return (
            f"{about}{self.total:,} tokens "
            f"({about}{self.input_tokens:,} in + {about}{self.output_tokens:,} out) "
            f"over {self.requests} request(s){where}"
        )

add

add(other: TokenUsage) -> None

Accumulate other in place, keeping the model/provider labels.

Source code in src/linexcel/aidoc.py

def add(self, other: TokenUsage) -> None:
    """Accumulate ``other`` in place, keeping the model/provider labels."""
    self.input_tokens += other.input_tokens
    self.output_tokens += other.output_tokens
    self.requests += other.requests
    self.estimated = self.estimated or other.estimated
    self.model = self.model or other.model
    self.provider = self.provider or other.provider

LLMProvider

Bases: Protocol

Minimal protocol: system + user prompt → text response.

Source code in src/linexcel/aidoc.py

@runtime_checkable
class LLMProvider(Protocol):
    """Minimal protocol: system + user prompt → text response."""

    def generate(
        self,
        system_prompt: str,
        user_prompt: str,
        *,
        temperature: float = 0.2,
        max_tokens: int | None = None,
    ) -> str: ...

UsageReportingProvider

Bases: Protocol

A provider that also reports what the call consumed.

Optional: the built-in OpenAI-compatible client implements it so that token counts come from the API rather than from an approximation. Custom providers only need :class:LLMProvider.

Source code in src/linexcel/aidoc.py

@runtime_checkable
class UsageReportingProvider(Protocol):
    """A provider that also reports what the call consumed.

    Optional: the built-in OpenAI-compatible client implements it so that token
    counts come from the API rather than from an approximation. Custom
    providers only need :class:`LLMProvider`.
    """

    def generate_with_usage(
        self,
        system_prompt: str,
        user_prompt: str,
        *,
        temperature: float = 0.2,
        max_tokens: int | None = None,
    ) -> tuple[str, TokenUsage]: ...

VisionProvider

Bases: Protocol

A provider that can be handed an image alongside the prompts.

Optional, and separate from :class:LLMProvider on purpose: most models served behind an OpenAI-compatible endpoint are text-only, and a caller asking for screenshot descriptions should be told so rather than have an image quietly dropped from the request.

Source code in src/linexcel/aidoc.py

@runtime_checkable
class VisionProvider(Protocol):
    """A provider that can be handed an image alongside the prompts.

    Optional, and separate from :class:`LLMProvider` on purpose: most models
    served behind an OpenAI-compatible endpoint are text-only, and a caller
    asking for screenshot descriptions should be told so rather than have an
    image quietly dropped from the request.
    """

    def generate_with_image(
        self,
        system_prompt: str,
        user_prompt: str,
        image: bytes,
        *,
        media_type: str = "image/png",
        temperature: float = 0.2,
        max_tokens: int | None = None,
    ) -> tuple[str, TokenUsage]: ...

estimate_tokens

estimate_tokens(text: str) -> int

Approximate the token count of text.

Only a fallback: :class:TokenUsage prefers the counts the provider reports. Latin script is counted as words × 4/3 (the usual 1 token ≈ 0.75 words ratio); CJK characters are counted individually, because a Japanese or Chinese sentence carries no spaces and would otherwise register as a single word.

estimate_tokens("the quick brown fox jumps") 6 estimate_tokens("") 0

Source code in src/linexcel/aidoc.py

def estimate_tokens(text: str) -> int:
    """Approximate the token count of ``text``.

    Only a fallback: :class:`TokenUsage` prefers the counts the provider
    reports. Latin script is counted as words × 4/3 (the usual 1 token ≈ 0.75
    words ratio); CJK characters are counted individually, because a Japanese
    or Chinese sentence carries no spaces and would otherwise register as a
    single word.

    >>> estimate_tokens("the quick brown fox jumps")
    6
    >>> estimate_tokens("")
    0
    """
    if not text:
        return 0
    cjk = len(_CJK_RE.findall(text))
    latin_words = len(_WORD_RE.findall(_CJK_RE.sub(" ", text)))
    return cjk + latin_words * 4 // 3

build_dossier

build_dossier(
    graph: dict[str, Any], node_id: str
) -> dict[str, Any] | None

Deterministic dossier for a node: everything the AI is allowed to use.

Source code in src/linexcel/aidoc.py

def build_dossier(graph: dict[str, Any], node_id: str) -> dict[str, Any] | None:
    """
    Deterministic dossier for a node: everything the AI is allowed to use.
    """
    return _build_dossier(_index_dossiers(graph), node_id)

build_workbook_dossier

build_workbook_dossier(
    graph: dict[str, Any],
    *,
    context: dict[str, Any] | None = None,
) -> dict[str, Any]

Return a compact, deterministic dossier for a whole-workbook overview.

context is a :attr:linexcel.LineageResult.workbook_context mapping. The graph alone describes how a workbook computes; it says nothing about what a reader sees on opening it — titles sitting above a table, the labels in the first column, cell comments, hidden columns, frozen panes. Those cues are exactly what the sheet screenshots show, and merging them into each sheet entry lets a text-only model describe the file as it looks without any image ever leaving the machine.

Both parts stay deterministic: every value is read from the workbook, so the "cite only the dossier" rule of the system prompt still holds.

Source code in src/linexcel/aidoc.py

def build_workbook_dossier(
    graph: dict[str, Any], *, context: dict[str, Any] | None = None
) -> dict[str, Any]:
    """Return a compact, deterministic dossier for a whole-workbook overview.

    ``context`` is a :attr:`linexcel.LineageResult.workbook_context` mapping.
    The graph alone describes how a workbook *computes*; it says nothing about
    what a reader sees on opening it — titles sitting above a table, the labels
    in the first column, cell comments, hidden columns, frozen panes. Those cues
    are exactly what the sheet screenshots show, and merging them into each
    sheet entry lets a text-only model describe the file as it looks without any
    image ever leaving the machine.

    Both parts stay deterministic: every value is read from the workbook, so the
    "cite only the dossier" rule of the system prompt still holds.
    """
    nodes = graph.get("nodes", [])
    meta = graph.get("meta", {})
    stats = meta.get("stats", {})
    sheet_stats = stats.get("sheets", [])
    nodes_by_sheet: dict[str, dict[str, int]] = {}
    for node in nodes:
        sheet = node.get("sheet")
        if not sheet:
            continue
        kinds = nodes_by_sheet.setdefault(sheet, {})
        kind = node.get("kind", "unknown")
        kinds[kind] = kinds.get(kind, 0) + 1

    sheets = [
        {
            "name": sheet.get("name"),
            "dimensions": {"rows": sheet.get("rows"), "columns": sheet.get("cols")},
            "formula_cells": sheet.get("formulaCells"),
            "lineage_nodes": nodes_by_sheet.get(sheet.get("name"), {}),
        }
        for sheet in sheet_stats
    ]
    if context:
        sheets = _merge_presentation(sheets, context)
    formula_patterns = sorted(
        (
            {
                "node_id": node.get("id"),
                "sheet": node.get("sheet"),
                "address": node.get("addr"),
                "formula": node.get("formula"),
                "cells": node.get("count", 1),
                "extent": node.get("bbox"),
                **_value_evidence(node),
            }
            for node in nodes
            if node.get("kind") in {"cell", "group"}
        ),
        key=lambda item: item["cells"],
        reverse=True,
    )
    pattern_total = len(formula_patterns)
    formula_patterns = formula_patterns[:20]
    defined_names = [
        {"name": node.get("label"), "targets": node.get("targets", [])}
        for node in nodes
        if node.get("kind") == "name"
    ]
    vba = [
        {
            "module": node.get("module"),
            "procedure": node.get("proc"),
            "type": node.get("procKind"),
        }
        for node in nodes
        if node.get("kind") == "vba"
    ]
    opaque_references = [
        node.get("label") for node in nodes if node.get("kind") == "opaque"
    ]
    ext_total = stats.get("externalWorkbooks", 0)
    ext_read = stats.get("externalWorkbooksRead", 0)
    external_workbooks = {
        "workbooks_referenced": ext_total,
        "workbooks_read_from_disk": ext_read,
        "unread_workbooks": max(0, ext_total - ext_read),
        "interpretation": (
            "Workbooks not read from disk were not opened; dependent cells use "
            "embedded file caches if present, or have no value. "
            "A known cached value does not prove an external file was opened or read."
        ),
    }
    return {
        "filename": meta.get("filename"),
        # Live native handles are intentionally not returned from an isolated
        # worker. Their API availability says nothing about recalculation.
        "execution": {
            key: value
            for key, value in meta.get("execution", {"status": "unknown"}).items()
            if key in {"status", "phase", "completedPhases", "elapsedSeconds"}
        },
        "recalculation_engine": meta.get("engine", "unspecified"),
        "value_coverage": {
            "scope": meta.get("coverage", {}).get("scope", "not_provided"),
            "total_nodes": meta.get("coverage", {}).get("totalNodes"),
            "counts": {
                key: value.get("count")
                for key, value in meta.get("coverage", {}).get("categories", {}).items()
            },
            "interpretation": (
                "Counts describe graph nodes, not workbook cells. Engine formula "
                "values were recalculated in the worker; they remain independently "
                "unverified. Cache values are separate. A group represents several "
                "cells but counts once; its displayed value belongs to one member."
            ),
        },
        "coverage": meta.get("analysisCoverage", {"status": "not_inspected"}),
        "analysis": {
            "formula_cells": stats.get("totalFormulas"),
            "count_scope": "extracted_formula_cells_not_workbook_total",
            "lineage_nodes": stats.get("totalNodes", 0),
            "lineage_edges": stats.get("totalEdges", 0),
            "grouped_patterns": stats.get("groupedPatterns", 0),
        },
        "sheets": sheets,
        "formula_patterns": formula_patterns,
        "formula_pattern_coverage": {
            "scope": "graph_formula_nodes",
            "total": pattern_total,
            "shown": len(formula_patterns),
            "omitted": pattern_total - len(formula_patterns),
        },
        "graph_connections": _document_connections(graph),
        "defined_names": defined_names,
        "source_defined_names": _document_name_evidence(
            meta.get("definedNameEvidence", {"status": "not_inspected"})
        ),
        "defined_names_in_graph_are_not_source_inventory": True,
        "vba_procedures": vba,
        "external_workbooks": external_workbooks,
        "external_or_unresolved_references": opaque_references,
        "warnings": [
            *meta.get("warnings", []),
            *(context.get("warnings", []) if context else []),
        ],
    }

render_markdown_table

render_markdown_table(
    columns: list[Any],
    rows: list[list[Any]],
    *,
    caption: str | None = None,
) -> str

Render a Markdown pipe table from structured data — the only place in linexcel where table syntax is written.

A column whose cells all parse as numbers is right-aligned (---:), so amounts line up on their decimal separator. Rows shorter than the header are padded, longer ones truncated: a ragged model row can no longer shift every column after it.

Source code in src/linexcel/aidoc.py

def render_markdown_table(
    columns: list[Any],
    rows: list[list[Any]],
    *,
    caption: str | None = None,
) -> str:
    """Render a Markdown pipe table from structured data — the only place in
    linexcel where table syntax is written.

    A column whose cells all parse as numbers is right-aligned (``---:``),
    so amounts line up on their decimal separator. Rows shorter than the
    header are padded, longer ones truncated: a ragged model row can no
    longer shift every column after it.
    """
    cols = [_md_cell(c) for c in columns]
    width = len(cols)
    body = [([_md_cell(v) for v in row] + [""] * width)[:width] for row in rows]
    aligns = []
    for j in range(width):
        values = [r[j] for r in body if r[j]]
        numeric = bool(values) and all(_NUMERIC_CELL_RE.fullmatch(v) for v in values)
        aligns.append("---:" if numeric else "---")
    lines: list[str] = []
    if caption:
        lines += [f"*{caption}*", ""]
    lines.append("| " + " | ".join(cols) + " |")
    lines.append("|" + "|".join(aligns) + "|")
    lines += ["| " + " | ".join(r) + " |" for r in body]
    return "\n".join(lines)

document_workbook

document_workbook(
    graph: dict[str, Any],
    *,
    model: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    provider: ProviderLike | None = None,
    language: str = "en",
    usage: TokenUsage | None = None,
    max_tokens: int | None = None,
    token_budget: int | None = None,
    context: dict[str, Any] | None = None,
    validation_results: dict[str, Any] | None = None,
) -> str

Generate a Markdown overview grounded in the workbook dossier.

Provider resolution (first match wins; no implicit default): 1. provider — custom LLMProvider instance or callable 2. base_url + model (or LINEXCEL_AI_BASE_URL + LINEXCEL_AI_MODEL) — any OpenAI-compatible endpoint

context is the workbook presentation context — the sheet previews, comments, merged cells, frozen panes and hidden columns a reader sees when opening the file. Pass it to describe the workbook as it looks, not only as it computes; see :func:build_workbook_dossier.

If a :class:TokenUsage is passed as usage, what the call consumed is accumulated into it. token_budget caps cumulative spend across that accumulator: an already-exhausted budget raises before anything is sent.

validation_results, when supplied, receives the bounded quotation-check report under workbook, including the raw response. Unsupported comparable quotations and processing limits add a visible qualification; ellipses and numerical illustrations remain explicitly unverified without alerts.

Source code in src/linexcel/aidoc.py

def document_workbook(
    graph: dict[str, Any],
    *,
    model: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    provider: ProviderLike | None = None,
    language: str = "en",
    usage: TokenUsage | None = None,
    max_tokens: int | None = None,
    token_budget: int | None = None,
    context: dict[str, Any] | None = None,
    validation_results: dict[str, Any] | None = None,
) -> str:
    """Generate a Markdown overview grounded in the workbook dossier.

    Provider resolution (first match wins; no implicit default):
    1. `provider` — custom LLMProvider instance or callable
    2. `base_url` + `model` (or `LINEXCEL_AI_BASE_URL` + `LINEXCEL_AI_MODEL`) —
       any OpenAI-compatible endpoint

    ``context`` is the workbook presentation context — the sheet previews,
    comments, merged cells, frozen panes and hidden columns a reader sees when
    opening the file. Pass it to describe the workbook as it looks, not only as
    it computes; see :func:`build_workbook_dossier`.

    If a :class:`TokenUsage` is passed as ``usage``, what the call consumed is
    accumulated into it. ``token_budget`` caps cumulative spend across that
    accumulator: an already-exhausted budget raises before anything is sent.

    ``validation_results``, when supplied, receives the bounded quotation-check
    report under ``workbook``, including the raw response. Unsupported comparable
    quotations and processing limits add a visible qualification; ellipses and
    numerical illustrations remain explicitly unverified without alerts.
    """
    if language not in _LANGUAGES:
        raise ValueError(f"Unsupported language: {language!r}. Use one of {_LANGUAGES}")
    _check_budget(token_budget, usage)
    dossier = build_workbook_dossier(graph, context=context)
    blob = _fit_workbook_dossier(dossier)
    llm = _resolve_provider(
        provider=provider, model=model, api_key=api_key, base_url=base_url
    )
    system = _WORKBOOK_SYSTEM[language]
    user = "Workbook dossier (deterministic, extracted from workbook):\n" + blob
    try:
        text, call_usage = _generate(llm, system, user, max_tokens=max_tokens)
    except AiDocError as exc:
        if usage is not None and exc.usage is not None:
            usage.add(exc.usage)
        raise
    except Exception as exc:
        raise AiDocError(f"AI documentation failed: {exc}") from exc
    if usage is not None:
        usage.add(call_usage)
    rendered = _insert_tables(text) if text else ""
    if not rendered.strip():
        raise AiDocError("AI returned empty response")
    report = validate_documentation(text, json.loads(blob))
    if validation_results is not None:
        validation_results["workbook"] = report
    return _qualify_documentation(rendered, report, language)

describe_images

describe_images(
    images: Mapping[str, bytes | str | Path],
    *,
    model: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    provider: ProviderLike | None = None,
    language: str = "en",
    usage: TokenUsage | None = None,
    max_tokens: int | None = None,
    token_budget: int | None = None,
) -> dict[str, str]

Describe rendered images with a multimodal model, {name: markdown}.

images maps a name — a sheet name, in practice — to a PNG, either as bytes or as a path to read. Each one is sent on its own, so a description is grounded in a single picture and nothing else.

This is the one part of linexcel whose evidence is not the deterministic dossier: a screenshot shows what no extraction reaches — colour conventions, conditional formatting, charts, the shape of a layout — and the prompt confines the model to what is visible rather than letting it reason about the calculation.

The provider must accept an image (:class:VisionProvider); a text-only one raises rather than having the picture dropped from the request. Model resolution is otherwise :func:document_workbook's, so model= here is where a vision model is named when it differs from the writing one.

Images are sent one at a time: they are large, and the local runtimes this is most used against serialize them anyway. An image that fails is skipped with a :class:UserWarning; :class:AiDocError is raised only when every one failed. token_budget is checked before each call.

Source code in src/linexcel/aidoc.py

def describe_images(
    images: Mapping[str, bytes | str | Path],
    *,
    model: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    provider: ProviderLike | None = None,
    language: str = "en",
    usage: TokenUsage | None = None,
    max_tokens: int | None = None,
    token_budget: int | None = None,
) -> dict[str, str]:
    """Describe rendered images with a multimodal model, ``{name: markdown}``.

    ``images`` maps a name — a sheet name, in practice — to a PNG, either as
    bytes or as a path to read. Each one is sent on its own, so a description
    is grounded in a single picture and nothing else.

    This is the one part of linexcel whose evidence is not the deterministic
    dossier: a screenshot shows what no extraction reaches — colour
    conventions, conditional formatting, charts, the shape of a layout — and
    the prompt confines the model to what is visible rather than letting it
    reason about the calculation.

    The provider must accept an image (:class:`VisionProvider`); a text-only
    one raises rather than having the picture dropped from the request. Model
    resolution is otherwise :func:`document_workbook`'s, so ``model=`` here is
    where a vision model is named when it differs from the writing one.

    Images are sent one at a time: they are large, and the local runtimes this
    is most used against serialize them anyway. An image that fails is skipped
    with a :class:`UserWarning`; :class:`AiDocError` is raised only when every
    one failed. ``token_budget`` is checked before each call.
    """
    if language not in _LANGUAGES:
        raise ValueError(f"Unsupported language: {language!r}. Use one of {_LANGUAGES}")
    _check_budget(token_budget, usage)
    if not images:
        return {}
    llm = _resolve_provider(
        provider=provider, model=model, api_key=api_key, base_url=base_url
    )
    if not isinstance(llm, VisionProvider):
        raise AiDocError(
            f"{type(llm).__name__} cannot be handed an image: describing "
            "screenshots needs a provider exposing generate_with_image("
            "system_prompt, user_prompt, image, *, media_type). The built-in "
            "OpenAI-compatible client does; point it at a multimodal model."
        )
    system = _VISION_SYSTEM[language]
    described: dict[str, str] = {}
    failed: list[str] = []
    for name, image in images.items():
        try:
            _check_budget(token_budget, usage)
        except AiDocError:
            warnings.warn(
                f"Token budget reached: {len(images) - len(described)} "
                "screenshot(s) left undescribed.",
                UserWarning,
                stacklevel=2,
            )
            break
        try:
            payload, media_type = _image_payload(name, image)
            text, call_usage = llm.generate_with_image(
                system,
                f"Sheet: {name}",
                payload,
                media_type=media_type,
                temperature=0.2,
                max_tokens=max_tokens,
            )
        except (AiDocError, OSError) as exc:
            if (
                isinstance(exc, AiDocError)
                and exc.usage is not None
                and usage is not None
            ):
                usage.add(exc.usage)
            failed.append(f"{name} ({exc})")
            continue
        if usage is not None:
            usage.add(call_usage)
        rendered = _unwrap_markdown(text)
        if rendered.strip():
            described[name] = rendered
        else:
            failed.append(f"{name} (AI returned empty response)")
    if failed and not described:
        raise AiDocError("No screenshot could be described: " + "; ".join(failed))
    if failed:
        warnings.warn(
            f"{len(failed)} screenshot(s) not described: " + "; ".join(failed),
            UserWarning,
            stacklevel=2,
        )
    return described

document_nodes

document_nodes(
    graph: dict[str, Any],
    node_ids: list[str],
    *,
    model: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    provider: ProviderLike | None = None,
    language: str = "en",
    max_workers: int = 4,
    usage: TokenUsage | None = None,
    max_tokens: int | None = None,
    token_budget: int | None = None,
    validation_results: dict[str, Any] | None = None,
) -> dict[str, str]

Document the requested nodes, returns {node_id: markdown}.

Provider resolution is the same as :func:document_workbook (no implicit default; see :func:_resolve_provider).

Nodes are documented concurrently (max_workers in-flight requests; raise it if the provider's rate limits allow). Documenting a large workbook is a long, often billed operation, so a node that fails does not discard the ones that succeeded: the successful cards are returned and a :class:UserWarning reports how many nodes were dropped. :class:AiDocError is raised only when every node failed.

If a :class:TokenUsage is passed as usage, consumed tokens are accumulated into it, including usage reported for rejected responses and calls in a run that later fails. Tokens already spent are still billed.

token_budget is a ceiling on the total tokens the run may spend, input and output together, counted against usage so several calls sharing one accumulator share one ceiling. It is enforced between requests, the only point at which a cost is known: nodes still queued when the tally reaches the budget are never sent, and a :class:UserWarning reports how many were left undocumented. Requests already in flight are allowed to finish, so the final tally can exceed the budget by up to max_workers responses — set it as an order of magnitude, not to the token. Use max_tokens to bound each individual response instead.

validation_results, when supplied, receives a bounded quotation-check report keyed by node ID, including raw responses. Formula quotations without source support are visibly qualified, not silently corrected. General prose remains unverified even when every inspected quotation has source support.

Source code in src/linexcel/aidoc.py

def document_nodes(
    graph: dict[str, Any],
    node_ids: list[str],
    *,
    model: str | None = None,
    api_key: str | None = None,
    base_url: str | None = None,
    provider: ProviderLike | None = None,
    language: str = "en",
    max_workers: int = 4,
    usage: TokenUsage | None = None,
    max_tokens: int | None = None,
    token_budget: int | None = None,
    validation_results: dict[str, Any] | None = None,
) -> dict[str, str]:
    """Document the requested nodes, returns {node_id: markdown}.

    Provider resolution is the same as :func:`document_workbook` (no implicit
    default; see :func:`_resolve_provider`).

    Nodes are documented concurrently (``max_workers`` in-flight requests;
    raise it if the provider's rate limits allow). Documenting a large
    workbook is a long, often billed operation, so a node that fails does not
    discard the ones that succeeded: the successful cards are returned and a
    :class:`UserWarning` reports how many nodes were dropped.
    :class:`AiDocError` is raised only when *every* node failed.

    If a :class:`TokenUsage` is passed as ``usage``, consumed tokens are
    accumulated into it, including usage reported for rejected responses and
    calls in a run that later fails. Tokens already spent are still billed.

    ``token_budget`` is a ceiling on the **total** tokens the run may spend,
    input and output together, counted against ``usage`` so several calls
    sharing one accumulator share one ceiling. It is enforced between requests,
    the only point at which a cost is known: nodes still queued when the tally
    reaches the budget are never sent, and a :class:`UserWarning` reports how
    many were left undocumented. Requests already in flight are allowed to
    finish, so the final tally can exceed the budget by up to ``max_workers``
    responses — set it as an order of magnitude, not to the token. Use
    ``max_tokens`` to bound each individual response instead.

    ``validation_results``, when supplied, receives a bounded quotation-check
    report keyed by node ID, including raw responses. Formula quotations without
    source support are visibly qualified, not silently corrected. General prose
    remains unverified even when every inspected quotation has source support.
    """
    if language not in _LANGUAGES:
        raise ValueError(f"Unsupported language: {language!r}. Use one of {_LANGUAGES}")
    if max_workers < 1:
        raise ValueError("max_workers must be >= 1")
    _check_budget(token_budget, usage)
    llm = _resolve_provider(
        provider=provider, model=model, api_key=api_key, base_url=base_url
    )
    system = _SYSTEM[language]
    docs: dict[str, str] = {}
    dossiers = []
    index = _index_dossiers(graph)
    for nid in node_ids:
        d = _build_dossier(index, nid)
        if d is not None:
            blob = _fit_node_dossier(d)
            dossiers.append((nid, blob))
    if not dossiers:
        return docs

    def _doc_one(nid_blob: tuple[str, str]) -> tuple[str, str, TokenUsage, dict]:
        nid, blob = nid_blob
        user = "Lineage dossier (deterministic, extracted from workbook):\n" + blob
        text, call_usage = _generate(llm, system, user, max_tokens=max_tokens)
        rendered = _insert_tables(text) if text else ""
        if not rendered.strip():
            raise AiDocError("AI returned empty response", usage=call_usage)
        report = validate_documentation(text, json.loads(blob))
        return (
            nid,
            _qualify_documentation(rendered, report, language),
            call_usage,
            report,
        )

    # The tally drives the budget, so it must exist even when the caller wants
    # no accumulator of their own; when they do pass one, it *is* the tally.
    tally = usage if usage is not None else TokenUsage()
    failures: list[tuple[str, Exception]] = []
    queue = iter(dossiers)
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures: dict[Future[tuple[str, str, TokenUsage, dict]], str] = {}

        def _submit_next() -> bool:
            """Queue one more node unless the budget is spent or none is left."""
            if token_budget is not None and tally.total >= token_budget:
                return False
            item = next(queue, None)
            if item is None:
                return False
            futures[pool.submit(_doc_one, item)] = item[0]
            return True

        # Nodes are submitted a few at a time rather than all at once: a budget
        # can only stop work that has not been handed to the pool yet.
        for _ in range(max_workers):
            if not _submit_next():
                break
        # Accumulating here rather than inside the workers keeps `tally` free of
        # races: this loop is the single consumer of the pool's results.
        while futures:
            done, _pending = wait(list(futures), return_when=FIRST_COMPLETED)
            for fut in done:
                node_id = futures.pop(fut)
                try:
                    nid, text, call_usage, report = fut.result()
                except Exception as exc:
                    if isinstance(exc, AiDocError) and exc.usage is not None:
                        tally.add(exc.usage)
                    failures.append((node_id, exc))
                    continue
                docs[nid] = text
                if validation_results is not None:
                    validation_results[nid] = report
                tally.add(call_usage)
            for _ in range(len(done)):
                if not _submit_next():
                    break

    # Only an exhausted budget can leave the queue undrained.
    unsent = sum(1 for _ in queue) if token_budget is not None else 0
    if unsent:
        warnings.warn(
            f"Token budget of {token_budget:,} tokens reached after "
            f"{tally.total:,}: {len(docs)} of {len(dossiers)} nodes documented, "
            f"{unsent} never sent. Raise token_budget to document the rest.",
            UserWarning,
            stacklevel=2,
        )

    if failures and not docs:
        node_id, exc = failures[0]
        raise AiDocError(
            f"AI documentation failed for all {len(failures)} nodes "
            f"(first error, node {node_id}): {exc}"
        ) from exc
    if failures:
        node_id, exc = failures[0]
        warnings.warn(
            f"AI documentation failed for {len(failures)} of {len(dossiers)} "
            f"nodes; {len(docs)} cards returned. First error (node {node_id}): "
            f"{exc}",
            UserWarning,
            stacklevel=2,
        )
    return docs