"""The ONLY place that talks to an LLM — through a coding-agent CLI, not an API key.

Ingest shells out to a CLI (``claude``, ``copilot``, or ``agy`` — Google's Antigravity CLI) in
**agentic** mode:
the CLI is pointed at the workspace (``cwd`` = the workspace root) with autonomous file tools,
reads the raw source and the existing wiki itself, and **edits the wiki page files directly**. We
pass only a short instruction that references files BY PATH — never file content — so the argv
stays tiny (this is what kills the old Windows ``WinError 206`` argv-length limit) and the
agent can handle an arbitrarily large raw file. Everything the agent must KNOW lives in the
rules tree (``citadel/rules/``, overridable per file from the workspace ``rules/``): the prompt
is only the code-invariant frame — the rules read list ``kind`` maps onto, the session variables,
and the operational invariants (see :func:`_build_instruction`).

- Pick the backend with ``CITADEL_LLM_CLI`` (``claude`` | ``copilot`` | ``agy``; default
  ``claude``), read via ``config.LLM_CLI`` and normalized by :func:`resolve_cli_name` (which is
  also where the RETIRED ``gemini`` backend is refused with a migration hint).
- Override the binary path with ``CLAUDE_CODE_PATH`` / ``COPILOT_CLI_PATH`` / ``AGY_CLI_PATH``.
- ``config.INGEST_MODEL`` is passed as ``--model`` to EVERY backend (all three honor the flag),
  and to none when it is unset — the knob has no default, so an unconfigured workspace runs each
  CLI on its own default model instead of being handed a foreign model id.
- Each backend also REPORTS which model actually served the session (claude's ``modelUsage`` map,
  copilot's JSONL ``model``/``modelCacheState`` events, agy's ``init`` event). That reported id —
  not the configured request — is what ingest stamps into the manifest, so the wiki never claims
  a model that never ran.
- Sessions run **hermetically** by default (``CITADEL_HERMETIC``, default on): when the
  installed CLI advertises a session-isolation flag (claude ``--bare`` — skip user
  hooks/CLAUDE.md/MCP discovery), it is appended so the user's personal agent configuration
  never leaks into ingest. Feature-probed per binary; an older CLI runs exactly as before.

One function does real work: ``run_ingest_session(rel_key)`` runs the chosen CLI once against
the workspace. The RESULT is whatever the agent wrote under ``wiki/``, which ``ingest``
discovers via a filesystem diff; the return value is only the session's best-effort
:class:`SessionUsage` (what the run cost, as the backend itself reports it — claude's
``--output-format json`` envelope, copilot's ``--output-format json`` JSONL stream, agy's
``--output-format stream-json`` events; None when the backend reports nothing). It raises
``RuntimeError`` on a missing/
unusable CLI, a non-zero exit, a claude ``is_error`` envelope, or a timeout (the failure
surface ``ingest``'s per-source ``try/except`` already expects).
"""

from __future__ import annotations

import contextlib
import json
import os
import re
import shutil
import subprocess
import sys
import threading
import time
from dataclasses import dataclass
from pathlib import Path

from . import config
from .llm_usage import AIC_USD as AIC_USD

# --- Facade re-exports (the session-usage accounting layer, split into llm_usage.py) ------------
# Everything spend-shaped lives there; every name stays addressable through this module
# (`llm.SessionUsage`, `llm.combine_usage`, the per-backend envelope parsers the tests drive).
# Names this module itself consumes are imported plainly; the pure re-exports use the
# redundant-alias form so the linter keeps them.
from .llm_usage import (
    SessionUsage,
    _error_from_copilot_jsonl,
    _last_result_envelope,
    _usage_from_claude_envelope,
    _usage_from_stream,
)
from .llm_usage import _iter_jsonl as _iter_jsonl
from .llm_usage import _model_from_claude_envelope as _model_from_claude_envelope
from .llm_usage import _usage_from_agy_stream as _usage_from_agy_stream
from .llm_usage import _usage_from_copilot_jsonl as _usage_from_copilot_jsonl
from .llm_usage import combine_usage as combine_usage
from .llm_usage import format_aic as format_aic
from .llm_usage import format_cost as format_cost


# CLI binary resolution (env override name -> default binary name).
_CLI_PATH_ENV = {"claude": "CLAUDE_CODE_PATH", "copilot": "COPILOT_CLI_PATH", "agy": "AGY_CLI_PATH"}
_CLI_DEFAULT_BIN = {"claude": "claude", "copilot": "copilot", "agy": "agy"}

# Backends that were supported once and are not any more -> the migration hint. The Gemini CLI was
# renamed and reworked into Google's Antigravity CLI (`agy`), whose flags, output format and model
# ids all differ; silently treating `gemini` as an unknown CLI would spawn a binary with a
# best-effort argv it does not understand, so the removal is named out loud instead.
_RETIRED_CLIS = {
    "gemini": (
        "the 'gemini' backend was removed: the Gemini CLI is now Google's Antigravity CLI. "
        "Set CITADEL_LLM_CLI=agy (binary 'agy', override with AGY_CLI_PATH) and pick a model "
        "with CITADEL_INGEST_MODEL (see `agy models`)."
    )
}


def resolve_cli_name(cli: str | None) -> str:
    """The configured backend name, normalized (lower-cased, trimmed, empty -> ``claude``) — and
    the ONE place a RETIRED backend is refused, with the migration hint instead of a confusing
    downstream failure. Every entry point that reads ``config.LLM_CLI`` goes through this."""
    name = (cli or "claude").strip().lower()
    if name in _RETIRED_CLIS:
        raise RuntimeError(_RETIRED_CLIS[name])
    return name


def _resolve_cli(cli: str) -> str:
    """Return an executable path for the chosen CLI or raise a clear RuntimeError.

    A set-but-unresolvable ``*_CLI_PATH`` override gets its OWN error naming the variable and its
    value: the generic "not found on PATH" reads as a PATH problem, which sends the user
    debugging the wrong thing entirely — the classic case being a workspace ``.env`` on a SHARED
    drive that carries one machine's absolute CLI path onto every other machine that mounts it."""
    env_name = _CLI_PATH_ENV.get(cli, "")
    override = os.environ.get(env_name, "").strip()
    binary = override or _CLI_DEFAULT_BIN.get(cli, cli)
    path = shutil.which(binary)
    if path:
        return path
    if os.path.isabs(binary) and os.access(binary, os.X_OK):
        return binary
    if override:
        raise RuntimeError(
            f"{env_name} is set to {override!r}, but that is not an executable file on this "
            f"machine, so the {cli!r} CLI cannot be launched. Fix or remove the override "
            f"(mind that a workspace .env on a shared drive applies to EVERY machine that "
            f"mounts it — keep machine-specific paths in the shell environment instead)."
        )
    raise RuntimeError(
        f"the {cli!r} CLI was not found on PATH. Install it and log in "
        f"(for claude: run `claude` once and `/login`), or set CITADEL_LLM_CLI to a "
        f"CLI you have, or point {env_name or 'the path env var'} at "
        f"the binary."
    )


def _agent_path(path: Path) -> str:
    """The path string to name a configured directory (wiki/raw) in the agent prompt. The agentic
    CLI runs with ``cwd`` = ``config.WORKSPACE_ROOT``, so a directory UNDER the workspace is named relative
    to it (short, cwd-relative), while one OUTSIDE the workspace — e.g. a wiki/raw tree on a mounted
    network drive — is named by its ABSOLUTE path so the agent (granted access via ``--add-dir``)
    can find it. Honors ``CITADEL_WIKI_DIR`` / ``CITADEL_RAW_DIR`` and never collapses an
    out-of-workspace dir to a bare name (the old bug that made the agent edit a non-existent
    ``./wiki``). Single source of truth: ``config.rel_or_abs_posix``."""
    return config.rel_or_abs_posix(path)


def _external_dirs(rel_key: str, read_path: str | None = None) -> list[str]:
    """OS-native paths of the directories the agent must read/write that live OUTSIDE the
    workspace, so the CLI can be granted access to them (claude ``--add-dir``, agy ``--add-dir``
    — both grant RECURSIVELY, so one grant of a rules root covers its
    tasks/formats/genres subdirs; copilot needs nothing — it runs with ``--allow-all-paths``). The
    agent's ``cwd`` is the workspace root, so this returns — de-duplicated and sorted — only the
    out-of-workspace members of {wiki dir (written), every raw source root
    (``config.source_roots()``), docs dir, every directory that can
    hold a referenced rules file (the packaged ``config.PACKAGED_RULES_DIR`` — under site-packages
    for a pip install, hence never inside a user workspace and ALWAYS granted there — plus the
    workspace ``rules/`` overlay, which lives under cwd and therefore filters out), the source
    file's own parent, and — for an Office source — the temp dir holding its extracted text}.
    Empty for the all-under-workspace dev-checkout layout (cwd already covers the rules), so that
    invocation is byte-for-byte unchanged.

    Each granted directory is named in its CHILD-friendly spelling (``config.native_form``): the
    grant has to match the paths the agent actually walks from its ``cwd``, and on a Windows mapped
    drive that is the drive letter, not the UNC form ``resolve()`` produces. Outside that case the
    two spellings are identical."""
    candidates = [
        config.wiki_dir(),
        *config.source_roots(),  # every raw source root (multi-root: an out-of-workspace root needs a grant)
        config.DOCS_DIR,
        config.PACKAGED_RULES_DIR,
        config.source_path_for_key(rel_key).parent,
    ]
    ws_rules = config.workspace_rules_dir()
    if ws_rules is not None:
        candidates.append(ws_rules)
    if read_path:
        # The extracted-text file lives in a system temp dir (outside the workspace); the agent
        # must be granted access so it can read what to ingest.
        candidates.append(Path(read_path).parent)
    out: dict[str, None] = {}
    for d in candidates:
        if config.is_outside_workspace(d):
            out[str(config.native_form(Path(d).resolve()))] = None
    return sorted(out)


@dataclass(frozen=True)
class _KindSpec:
    """The per-kind prompt-composition spec — the ONE table that replaces the scattered ``kind``
    checks. The EXTERNAL kind strings are the stable API (ingest.py, the manifest, and the tests
    keep them); only their mapping onto prompt SHAPE lives here. Columns:

    - ``task_file`` — the tree-relative ``tasks/*.md`` brief this kind reads.
    - ``reads_source`` — whether the session reads a raw SOURCE for content. It gates the two
      content-only frame parts: the genre enumeration (the agent judges genre FROM the source's
      text) and the fallback-date bullet (the source file's own date). ``delete`` (the file is
      gone) and ``curate`` (it improves EXISTING pages, not a source) read no source.
    - ``format_policy`` — how the format brief is chosen: ``"source"`` (an Office extract arrives
      via ``read_path`` → ``formats/office.md``; a large-source slice is covered by the task brief;
      else a PDF is magic-sniffed → ``formats/pdf.md`` — the ingest/reconcile axis), ``"repo"``,
      ``"image"``, ``"audio"`` (a whisper transcript arrives via ``read_path`` →
      ``formats/transcripts.md``, kept on EVERY segment — its cite-the-original and locator rules
      bind per slice), ``"pdf"`` (a pypdf text-layer extraction arrives via ``read_path`` →
      ``formats/pdf.md``, kept on every segment for the same reason), or ``"none"``. ``"none"``
      attaches NO format brief EVEN when a ``read_path`` is present — curate's findings file
      arrives via ``read_path`` and must never pull in ``formats/office.md``.
    - ``subject_prefix`` — the ``- <prefix>: <key>`` session bullet naming what the session acts on.
    """

    task_file: str
    reads_source: bool
    format_policy: str
    subject_prefix: str


# kind -> its composition spec. ingest/image/repo are all "fold a NEW source in" lifecycles (the
# format policy carries what differs); the *-reconcile kinds re-fold a changed/forced source;
# delete strips a removed source's provenance; curate (PR6) re-reads the run's findings file and
# improves EXISTING pages. An unknown kind is a FAIL-LOUD programming error (see _spec_for_kind).
_KIND_SPECS: dict[str, _KindSpec] = {
    "ingest": _KindSpec("tasks/ingest.md", True, "source", "Source (the source of record)"),
    "image": _KindSpec("tasks/ingest.md", True, "image", "Source (the source of record)"),
    "audio": _KindSpec("tasks/ingest.md", True, "audio", "Source (the source of record)"),
    "pdf": _KindSpec("tasks/ingest.md", True, "pdf", "Source (the source of record)"),
    "repo": _KindSpec("tasks/ingest.md", True, "repo", "Source (the source of record)"),
    "reconcile": _KindSpec("tasks/reconcile.md", True, "source", "Source (the source of record)"),
    "image-reconcile": _KindSpec("tasks/reconcile.md", True, "image", "Source (the source of record)"),
    "audio-reconcile": _KindSpec("tasks/reconcile.md", True, "audio", "Source (the source of record)"),
    "pdf-reconcile": _KindSpec("tasks/reconcile.md", True, "pdf", "Source (the source of record)"),
    "repo-reconcile": _KindSpec("tasks/reconcile.md", True, "repo", "Source (the source of record)"),
    "delete": _KindSpec("tasks/delete.md", False, "none", "Source (REMOVED from disk — do not open it)"),
    "curate": _KindSpec("tasks/curate.md", False, "none", "Page to curate (the cluster anchor)"),
}


def _spec_for_kind(kind: str) -> _KindSpec:
    """The composition spec for ``kind`` — FAIL LOUD (``ValueError``) on an unknown kind (a typo,
    or a new lifecycle that forgot to register here) instead of silently defaulting to an ingest
    brief and folding a source under the wrong task."""
    try:
        return _KIND_SPECS[kind]
    except KeyError:
        raise ValueError(f"unknown ingest kind {kind!r} (known: {', '.join(sorted(_KIND_SPECS))})") from None


def _is_pdf_source(rel_key: str) -> bool:
    """True when the source is a PDF, flagged exactly like ingest flags them — by the ``%PDF-``
    magic header (``ingest._is_ingestible`` / ``_read_source_text``) — with the ``.pdf`` suffix as
    the fallback when the file cannot be read (a vanished file, a unit test's phantom key)."""
    try:
        with open(config.source_path_for_key(rel_key), "rb") as fh:
            return fh.read(5) == b"%PDF-"
    except OSError:
        return rel_key.lower().endswith(".pdf")


def _format_brief(rel_key: str, kind: str, read_path: str | None, segment: tuple[int, int] | None) -> str | None:
    """The tree-relative FORMAT brief for this session, or None when no format applies — decided by
    the kind's ``format_policy`` (the ONE spec table). Formats are the structurally
    Python-detectable axis (repo markers, image magic, Office extraction, PDF magic); code selects
    the brief, genres stay the agent's content judgment.

    - ``repo``/``image``/``audio``/``pdf`` policies carry their format in the kind itself.
      ``audio`` and ``pdf`` keep their brief on every SEGMENT too (unlike the Office exemption
      below): the brief's cite-the-original-file and lines-locator rules bind for each slice of a
      long recording / large extraction.
    - ``none`` (delete, curate) attaches NO brief — even with a ``read_path`` present (curate's
      findings file must not pull in ``formats/office.md``).
    - ``source`` (ingest/reconcile): a ``read_path`` WITHOUT a segment is a pre-extracted Office
      source (``formats/office.md``); WITH a segment it is a large-source slice covered by the task
      brief's Large-sources rules (a slice is not an Office extract, even when the source was one);
      otherwise a direct read, and a PDF (magic-sniffed) reads ``formats/pdf.md``."""
    policy = _spec_for_kind(kind).format_policy
    if policy == "repo":
        return "formats/repo.md"
    if policy == "image":
        return "formats/image.md"
    if policy == "audio":
        return "formats/transcripts.md"
    if policy == "pdf":
        return "formats/pdf.md"
    if policy == "none":
        return None
    if read_path:
        return None if segment is not None else "formats/office.md"
    if _is_pdf_source(rel_key):
        return "formats/pdf.md"
    return None


def _referenced_rules(
    rel_key: str, kind: str = "ingest", read_path: str | None = None, segment: tuple[int, int] | None = None
) -> list[tuple[str, Path]]:
    """Every RESOLVED rules file the prompt for this session references, as ``(role, path)``
    entries in the prompt's TRUE READ ORDER — the CANONICAL order, which
    :func:`_build_instruction` renders its rules lines from (single composition owner, so the
    two can never drift): ``schema`` and ``core`` (every session), the ``task`` brief for
    ``kind``, the ``format`` brief when one applies, the workspace ``local`` house rules when
    present, and finally one ``genre`` entry per effective genre brief (agent-judged; only when
    the kind READS a source — ``delete``/``curate`` have no source content to judge). The single
    source of truth the prompt-validation tests check against (every path must exist and be
    reachable)."""
    spec = _spec_for_kind(kind)
    files: list[tuple[str, Path]] = [
        ("schema", config.effective_rules_file("schema.md")),
        ("core", config.effective_rules_file("core.md")),
        ("task", config.effective_rules_file(spec.task_file)),
    ]
    fmt = _format_brief(rel_key, kind, read_path, segment)
    if fmt:
        files.append(("format", config.effective_rules_file(fmt)))
    local = config.local_rules_file()
    if local is not None:
        files.append(("local", local))
    if spec.reads_source:
        files.extend(("genre", g) for g in config.effective_genres())
    return files


# role -> prompt line for the non-genre rules entries; genres render as ONE enumerating line.
_RULES_LINE = {
    "schema": "- Format contract: {}",
    "core": "- How you work: {}",
    "task": "- Task brief (what THIS session does): {}",
    "format": "- Format brief (how to read THIS source): {}",
    "local": "- Workspace house rules: {}",
}


def _build_instruction(
    rel_key: str,
    kind: str = "ingest",
    read_path: str | None = None,
    segment: tuple[int, int] | None = None,
    line_range: tuple[int, int] | None = None,
) -> str:
    """The short, paths-only agent prompt: ONLY the code-invariant frame. Everything the agent
    must KNOW lives in the rules tree (``citadel/rules/``, see its README) and is referenced BY
    PATH — the prompt never embeds file content, so it stays at most a couple thousand chars
    regardless of raw-file size (the WinError 206 fix). The frame is:

    1. the rules read list — rendered from :func:`_referenced_rules` (the single composition
       owner) in its canonical order: schema.md + core.md (every session), the task brief
       ``kind`` maps to, the format brief when one applies, the workspace ``rules/local.md``
       when present, and ONE line enumerating the effective genre briefs for the agent to judge
       from the source's CONTENT (none for ``delete`` — it never reads the source);
    2. the session VARIABLES as bullets — the source key (verbatim), the configured wiki/raw
       directories, the prepared read path (an Office extract / segment slice / repo digest),
       the extracted-images ``media/`` folder with its file names (only an Office source that
       carried embedded images), the segment position, the source file's own date as the
       content-date fallback, the target wiki language (``CITADEL_WIKI_LANG``), the PDF mode
       (``CITADEL_PDF_MODE``, PDF sources only), the style-profiling switch
       (``CITADEL_STYLE_PROFILES``, only when ON), and the related-source budget
       (``CITADEL_RELATED_SOURCES``, source-reading kinds only — how many OTHER raw files this
       session may consult, per ``core.md`` § Related sources);
    3. the operational invariants ingest enforces mechanically: the off-limits generated files,
       and the run-``citadel check``-before-finishing gate.

    Rules are named by their RESOLVED effective locations (workspace ``rules/`` overlay >
    packaged ``citadel/rules/``), rendered through the same :func:`config.rel_or_abs_posix`
    discipline as every other path: workspace-relative in the dev checkout, ABSOLUTE
    site-packages paths for a pip install (which :func:`_external_dirs` grants to the CLI).
    ``rel_key`` is the source key (workspace-relative posix for an in-workspace source, ABSOLUTE
    posix for one on a mounted drive); every directory name is read from config at CALL time so a
    custom layout (``CITADEL_WIKI_DIR=wikiET``, a ``T:\\team-wiki\\wiki`` network path) is named
    correctly instead of a hardcoded ``wiki/``.

    The external ``kind`` strings (``ingest`` / ``reconcile`` / ``delete`` / ``repo`` /
    ``repo-reconcile`` / ``image`` / ``image-reconcile`` / ``audio`` / ``audio-reconcile`` /
    ``pdf`` / ``pdf-reconcile``) are the stable API; they map internally
    onto (task brief, format brief). ``read_path`` is the prepared file for a pre-extracted
    Office source, one segment's slice of a large source (with ``segment=(part, total)``), or a
    repo digest — the agent reads it for content while citing ``rel_key`` as the source of
    record (per the task/format briefs)."""
    wiki_rel = _agent_path(config.wiki_dir())
    # The raw-dir bullet names the root that COVERS this source (in a multi-root corpus a
    # second-root source must not be pointed at the primary); an out-of-root explicit path
    # falls back to the primary RAW_DIR. Lexical lookup — no disk access, delete-safe.
    raw_root = config.root_covering(config.source_path_for_key(rel_key)) or config.RAW_DIR
    raw_rel = _agent_path(raw_root)
    spec = _spec_for_kind(kind)  # the ONE table: subject bullet + whether a source is read
    fmt = _format_brief(rel_key, kind, read_path, segment)  # for the PDF-mode bullet below

    lines = [
        "You are the ingest engine for a self-structuring wiki in Google's Open Knowledge Format.",
        "Read these rules files FIRST and follow them exactly:",
    ]
    genres: list[Path] = []
    for role, path in _referenced_rules(rel_key, kind, read_path, segment):
        if role == "genre":
            genres.append(path)  # last in the canonical order — collected onto one line below
        else:
            lines.append(_RULES_LINE[role].format(_agent_path(path)))
    if genres:
        lines.append(
            "Judge the source's genre from its CONTENT; if it reads like one of these, ALSO read "
            "and follow the matching file: " + ", ".join(_agent_path(g) for g in genres) + "."
        )

    lines += ["", "Session variables (use these paths verbatim):"]
    lines.append(f"- {spec.subject_prefix}: {rel_key}")
    lines.append(f"- Wiki directory: {wiki_rel}/")
    lines.append(f"- Raw directory: {raw_rel}/")
    if read_path:
        lines.append(f"- Prepared file — read THIS for the source's content: {read_path}")
        # An Office source's embedded images (when it carried any) sit in a media/ folder beside
        # the extract. Name the folder AND its files: an agent told only by the format brief that
        # such a folder MAY exist has (measurably) concluded it doesn't and skipped the images.
        if fmt == "formats/office.md":
            media_dir = Path(read_path).parent / "media"
            images: list[str] = []
            with contextlib.suppress(OSError):
                if media_dir.is_dir():
                    images = sorted(entry.name for entry in media_dir.iterdir())
            if images:
                shown = ", ".join(images[:8]) + (", …" if len(images) > 8 else "")
                lines.append(
                    "- Extracted images — VIEW each of these with your file reader and ingest "
                    f"what they show (see the format brief): {media_dir.as_posix()}/ ({shown})"
                )
    if segment is not None:
        lines.append(f"- Segment: part {segment[0]} of {segment[1]}")
        if line_range is not None:
            # A chunked AUDIO/PDF-extract pass: the prepared file is the WHOLE transcript /
            # extraction (its line numbers are the verification cache's), and this bullet bounds
            # the pass — no rebased slice, so every `lines A-B` locator the agent writes is
            # correct by construction.
            noun = "transcript" if fmt == "formats/transcripts.md" else "extracted text"
            lines.append(
                f"- {noun.capitalize()} window for THIS pass: lines {line_range[0]}-{line_range[1]} — the "
                f"prepared file holds the WHOLE {noun}; read ONLY this window (use a "
                "ranged/offset read, the file may be too large to read whole) and fold in its "
                "facts; the file's own line numbers ARE the locator line numbers"
            )
    # Fallback date for time-anchored sources: the raw file's own modification date, used only
    # when the source's CONTENT states no date (genres/meeting-minutes.md § Dates). Only for kinds
    # that READ a source (delete/curate have none), and guarded so a missing file (a repo folder,
    # a unit test's phantom key) yields no bullet.
    fallback_date = ""
    if spec.reads_source:
        with contextlib.suppress(OSError):
            src = config.source_path_for_key(rel_key)
            if src.is_file():
                fallback_date = time.strftime("%Y-%m-%d", time.gmtime(src.stat().st_mtime))
    if fallback_date:
        lines.append(
            "- Fallback date — the source file's own date, used only when the source's content "
            f"states no date: {fallback_date}"
        )
    lines.append(f"- Wiki language: {config.WIKI_LANG}")
    if fmt == "formats/pdf.md":
        lines.append(f"- PDF mode: {'images' if config.PDF_MODE == 'images' else 'text'}")
    if config.STYLE_PROFILES:
        lines.append("- Style profiling: ON")
    # Related-source budget (CITADEL_RELATED_SOURCES): how many OTHER raw files this session may
    # open when its own source uses something it never explains — a term/code/reference that only
    # resolves next to a file the wiki already holds. The NUMBER is the code-invariant part and
    # belongs in the frame; WHEN a lookup is warranted, how to find the right file, and the duty to
    # cite the consulted file live in the rules (core.md § Related sources). Only for kinds that
    # read a source, and only when the budget is non-zero — a 0 budget names nothing, so the rules
    # section's "without that bullet, read only your own source" leaves the session unchanged.
    if config.RELATED_SOURCES > 0 and spec.reads_source:
        lines.append(
            f"- Related-source budget: {config.RELATED_SOURCES} — at most this many OTHER raw "
            "files may be consulted in THIS session to resolve something the source above does "
            "not explain, each cited for whatever is taken from it (see `core.md` § Related "
            "sources); the source above stays the source of record"
        )
    # Operator guidance (--guidance / CITADEL_INGEST_GUIDANCE): a per-run steer from the human
    # running ingest — unlike text inside a source, it comes from the run instruction itself, so
    # following it is legitimate. Only for kinds that read a source (a delete cleanup has no source
    # to steer; curate's own steer, `citadel curate --guidance`, travels in the cluster's FINDINGS
    # file — curate's designated instruction channel — so it lands as a listed finding rather than
    # a second instruction competing with the improve-or-NOOP rule). Always bounded by the rules:
    # guidance routes and emphasizes, it never overrides citation/grounding/off-limits rules.
    if config.INGEST_GUIDANCE and spec.reads_source:
        lines.append(
            "- Operator guidance for THIS run (steer routing and emphasis with it, WITHIN the "
            "rules above — it never overrides the citation, grounding, or off-limits rules): " + config.INGEST_GUIDANCE
        )

    lines += [
        "",
        f"Do what the task brief says by EDITING FILES DIRECTLY under {wiki_rel}/, using your "
        "built-in file tools (read/search/edit) — not shell commands — to read and search. "
        f"Your finished pages STAY under {wiki_rel}/ — that IS the deliverable. There is no "
        "publish step for you to perform: the orchestrator validates and imports your edits "
        f"from {wiki_rel}/ after this session ends, so never move, copy, or re-create them "
        "anywhere else (a page placed outside it is lost, not published), and never delete, "
        "rename, or move that wiki directory itself. "
        f"The source and {raw_rel}/ are READ-ONLY inputs: read them, but never write, create, "
        f"move, or delete anything there. Never create or edit {wiki_rel}/index.md, "
        f"{wiki_rel}/log.md, any */index.md, or any dotfile, and make no changes outside "
        f"{wiki_rel}/. When your edits "
        "are complete, run `citadel check` (or `uv run python -m "
        "citadel check`) ONCE; only if it reports errors, fix them and run it again to confirm.",
    ]
    return "\n".join(lines)


def _build_invocation(
    cli: str, cli_path: str, prompt: str, extra_dirs: list[str] | None = None
) -> tuple[list[str], str | None]:
    """Return ``(argv, stdin_text)`` for the chosen CLI in agentic, non-interactive mode.

    Each CLI runs with autonomous file tools and ``cwd`` = the workspace root (set by
    ``_run_session``). ``extra_dirs`` are directories the agent must reach that live OUTSIDE the
    workspace (an out-of-workspace wiki/raw on a mounted network drive, and the packaged rules dir
    for a pip install — computed by :func:`_external_dirs`) — empty for the all-under-workspace
    dev-checkout layout. For **claude** the prompt goes on STDIN (argv carries only flags); for
    copilot/agy it is a ``-p`` argument — now safe because the prompt is tiny.

    Every backend is passed ``--model`` when :data:`config.INGEST_MODEL` is set, and none when it
    is not (the knob has no default, so an unset workspace runs each CLI on its own default model
    rather than being handed a foreign model id).

    All three CLIs run in their fully autonomous mode (each backend's headless YOLO equivalent);
    which files the agent may touch and which tools it should use is governed by the run
    instruction (:func:`_build_instruction`) — ``raw/`` is read-only, edits go only under the
    wiki, and the shell is reserved for the ``citadel check`` self-check and page deletes/renames —
    not by divergent per-CLI permission flags."""
    extra_dirs = extra_dirs or []
    if cli == "claude":
        # acceptEdits auto-applies file edits; the allowlist scopes tools (Read/Edit/Write to
        # author pages, Grep/Glob to search the wiki, Bash so the agent can run `citadel check`).
        # cwd=workspace root covers everything under it; --add-dir grants access to what is not —
        # an out-of-workspace wiki/raw (network drive) and the packaged rules under site-packages —
        # since claude's file tools are otherwise scoped to cwd. No extra_dirs in the
        # all-under-workspace dev layout, so the argv is then unchanged.
        argv = [
            cli_path,
            "-p",
            "--output-format",
            "json",
            "--permission-mode",
            "acceptEdits",
            "--allowedTools",
            "Read Edit Write Grep Glob Bash",
        ]
        for d in extra_dirs:
            argv += ["--add-dir", d]
        if config.INGEST_MODEL:
            argv += ["--model", config.INGEST_MODEL]
        return argv, prompt
    if cli == "copilot":
        # --allow-all-tools is required for non-interactive editing; --allow-all-paths lets it
        # reach the wiki/raw AND the packaged rules whether they are under cwd, on a mounted drive,
        # or in site-packages (copilot has no per-directory grant mechanism, and needs none —
        # extra_dirs is deliberately unused here); --no-ask-user keeps it autonomous.
        # --output-format json emits one JSON object per line: the ONLY way to learn which model
        # actually served the session (copilot honors --model, including 'auto', so the configured
        # id is a request, not a fact) and what it cost in AI credits. It replaces the older -s,
        # which trimmed exactly that envelope away.
        argv = [
            cli_path,
            "-p",
            prompt,
            "--allow-all-tools",
            "--allow-all-paths",
            "--no-ask-user",
            "--output-format",
            "json",
        ]
        if config.INGEST_MODEL:
            argv += ["--model", config.INGEST_MODEL]
        return argv, None
    if cli == "agy":
        # Google's Antigravity CLI. --dangerously-skip-permissions is its headless YOLO equivalent
        # (anything less prompts for tool approval, which hangs with no TTY); --add-dir is
        # repeatable and adds the out-of-workspace dirs (a wiki/raw on a mounted drive, the
        # packaged rules under site-packages) to the agent's workspace. --output-format stream-json
        # is what carries the effective model (the `init` event) plus the final token usage; the
        # plain `json` format reports usage but never names the model.
        argv = [cli_path, "-p", prompt, "--output-format", "stream-json", "--dangerously-skip-permissions"]
        if config.INGEST_MODEL:
            argv += ["--model", config.INGEST_MODEL]
        for d in extra_dirs:
            argv += ["--add-dir", d]
        return argv, None
    # Unknown CLI: a plain headless prompt as an argument (best effort).
    return [cli_path, "-p", prompt], None


# --help output per resolved CLI binary, probed at most once per process (the hermetic-flag
# feature detection; a probe failure caches "" so a broken binary is not re-probed).
_HELP_TEXT_CACHE: dict[str, str] = {}


def _cli_help_text(cli_path: str) -> str:
    """The CLI's ``--help`` output (stdout+stderr merged), cached per binary path — the
    feature probe behind :func:`_hermetic_flags`. Best-effort: any spawn failure or
    timeout degrades to "" (the probed feature simply reads as absent), because usage
    accounting must never be able to break an ingest."""
    if cli_path not in _HELP_TEXT_CACHE:
        try:
            proc = subprocess.run(
                [cli_path, "--help"],
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                timeout=30,
                # DEVNULL, never the inherited stdin: a --help that blocks reading stdin (or the
                # MCP stdio pipe under `citadel serve`) gets immediate EOF instead of stalling
                # the probe against the 30s timeout.
                stdin=subprocess.DEVNULL,
            )
            _HELP_TEXT_CACHE[cli_path] = (proc.stdout or "") + (proc.stderr or "")
        except (OSError, subprocess.SubprocessError):
            _HELP_TEXT_CACHE[cli_path] = ""
    return _HELP_TEXT_CACHE[cli_path]


# Per-CLI session-isolation flags: what to pass so the user's PERSONAL agent configuration
# (~/.claude hooks, CLAUDE.md context, MCP-server discovery) does not leak into citadel's agent
# sessions — the session then runs on the rules tree alone, reproducibly. Only claude documents
# such a switch today; copilot/agy have no equivalent (no entry, so nothing is probed or passed).
_HERMETIC_FLAGS: dict[str, tuple[str, ...]] = {"claude": ("--bare",)}


def _hermetic_flags(cli: str, cli_path: str) -> list[str]:
    """The session-isolation flags to append for this backend, or ``[]`` when hermetic mode is off
    (``CITADEL_HERMETIC=0``), the CLI has no registered isolation flag, or the installed binary
    does not ADVERTISE the flag in ``--help`` — probed once per binary with an exact flag-token
    match: an older CLI
    must never be handed an unknown flag that would fail every session over configuration
    hygiene. A false NEGATIVE is safe (the session merely runs non-hermetic, the pre-knob
    behavior)."""
    flags = _HERMETIC_FLAGS.get(cli, ()) if config.HERMETIC else ()
    if not flags:
        return []
    help_text = _cli_help_text(cli_path)
    return [flag for flag in flags if re.search(re.escape(flag) + r"(?![\w-])", help_text)]


# Auth-shaped failure signatures in a backend's own error text.
_AUTH_ERROR_RE = re.compile(r"authenticat|unauthorized|not logged in|credential|api[ _-]?key|\b401\b", re.I)


def _hermetic_auth_hint(cli: str, argv: list[str], message: str) -> str:
    """The one hint that explains an auth-shaped session failure while hermetic mode is on, or ``""``.

    ``--bare`` deliberately skips the user's own agent configuration — and on a machine whose CLI
    credentials live in exactly that configuration (a managed container, an ``apiKeyHelper`` setup),
    EVERY citadel session then fails on authentication while the same CLI works fine interactively.
    The backend blames the network for it (*"This may be a temporary network issue, please try
    again"*), which sends you looking in the wrong place, so name the knob that actually explains
    it. Derived from ``argv`` rather than the config, so the hint appears only when a flag really
    was passed (hermetic mode is probe-gated: an older binary is handed nothing)."""
    flags = [flag for flag in _HERMETIC_FLAGS.get(cli, ()) if flag in argv]
    if not flags or not _AUTH_ERROR_RE.search(message):
        return ""
    return (
        f" [hermetic session isolation is on, so {' '.join(flags)} was passed: it skips your personal"
        f" {cli} configuration, which on some machines is where the CLI keeps its credentials."
        " If that CLI works interactively but every session fails here, set CITADEL_HERMETIC=0.]"
    )


# Monotonic per-process counter so concurrent/same-second transcript files never collide. Under
# `citadel ingest --jobs N` "concurrent" is literal — several worker threads write transcripts at
# once — and `_LOG_SEQ += 1` is a read-modify-write, so without the lock two sessions can be handed
# the same number and the guarantee this counter exists for quietly stops holding.
_LOG_SEQ = 0
_LOG_SEQ_LOCK = threading.Lock()


def _decode_partial(data) -> str:
    """Normalize the partial output a ``TimeoutExpired`` carries (``output``/``stderr``) to a string:
    None -> "", str -> itself, bytes -> UTF-8 with replacement (the captured path may hand back bytes
    if it timed out before decoding). Lets the timeout transcript include what was produced so far."""
    if data is None:
        return ""
    if isinstance(data, bytes):
        return data.decode("utf-8", errors="replace")
    return str(data)


def _echo_stderr(text: str) -> None:
    """Write diagnostic text to stderr, swallowing a broken/closed pipe. The verbose stream and the
    transcript-path notice are diagnostic ONLY, so a closed stderr (``2>|``, a downstream consumer
    that exited) must never abort the ingest."""
    with contextlib.suppress(OSError):
        sys.stderr.write(text)
        sys.stderr.flush()


def _stream_subprocess(cli: str, argv: list[str], stdin_text: str | None) -> tuple[int, str, str]:
    """Run the CLI while TEEING its output to the terminal live (verbose mode), returning
    ``(returncode, stdout, stderr)`` exactly like the captured path so the shared error-detection
    and transcript-logging below are unchanged.

    stderr is merged into stdout so the interleaved agent transcript is shown (and captured) in the
    real order it was produced; the returned ``stderr`` is therefore always empty. The tiny prompt
    is written to stdin and stdin is closed BEFORE reading any output, so there is no classic
    pipe-buffer deadlock (the prompt is at most a couple KB, well under the pipe buffer). A
    ``threading.Timer`` enforces ``config.LLM_TIMEOUT`` even if the child hangs producing no output
    — killing it raises ``TimeoutExpired`` to match the captured path."""
    proc = subprocess.Popen(  # noqa: S603 - argv is built internally, never from user input
        argv,
        stdin=subprocess.PIPE if stdin_text is not None else None,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        encoding="utf-8",
        errors="replace",
        bufsize=1,
        cwd=config.child_cwd(),
        env=config.child_env(),
    )
    if stdin_text is not None and proc.stdin is not None:
        try:
            proc.stdin.write(stdin_text)
        finally:
            proc.stdin.close()

    timed_out = {"v": False}

    def _kill() -> None:
        timed_out["v"] = True
        with contextlib.suppress(OSError):
            proc.kill()

    timer = threading.Timer(config.LLM_TIMEOUT, _kill)
    timer.start()
    chunks: list[str] = []
    # All terminal echo is diagnostic — a closed/broken stderr (e.g. `2>|`, or a downstream pipe
    # consumer that exited) must never abort the ingest, so every write goes through _echo_stderr.
    _echo_stderr(f"\n--- {cli} session (live) ---\n")
    try:
        if proc.stdout is not None:
            for line in proc.stdout:
                chunks.append(line)
                # Mirror the child's output to OUR stderr (stdout stays the final report), so a
                # piped `citadel ingest > report.txt` still shows the live transcript on screen.
                _echo_stderr(line)
        proc.wait()
    finally:
        timer.cancel()
    _echo_stderr(f"--- {cli} session end (exit {proc.returncode}) ---\n")
    if timed_out["v"]:
        # Carry whatever was streamed before the kill so _run_session can log a useful PARTIAL
        # transcript for the timed-out session instead of an empty one.
        raise subprocess.TimeoutExpired(argv, config.LLM_TIMEOUT, output="".join(chunks))
    return proc.returncode, "".join(chunks), ""


def _write_transcript(
    cli: str,
    argv: list[str],
    prompt: str | None,
    returncode: int | None,
    out: str,
    err: str,
    label: str | None,
    seconds: float,
    note: str = "",
) -> None:
    """Write ONE agent session's full record to ``config.LLM_LOG_DIR`` (no-op when unset) so there
    is an after-the-fact account of what the model saw and did — the prompt, the complete
    stdout/stderr, the exit code and duration. Best-effort: a logging failure must NEVER break
    ingest, so every error is swallowed. The path is announced on stderr so the user can find it."""
    log_dir = (config.LLM_LOG_DIR or "").strip()
    if not log_dir:
        return
    try:
        global _LOG_SEQ
        with _LOG_SEQ_LOCK:
            _LOG_SEQ += 1
            seq = _LOG_SEQ
        directory = Path(log_dir)
        if not directory.is_absolute():
            directory = config.WORKSPACE_ROOT / directory
        config.robust_mkdir(directory)
        stamp = time.strftime("%Y%m%d-%H%M%S")
        safe = "".join(c if (c.isalnum() or c in "-._") else "_" for c in (label or "session"))
        if len(safe) > 80:
            # The label is `kind[.pNofM].<source key>`, and for an out-of-workspace source the key
            # is a long absolute path whose FILENAME sits at the END — a plain head-cut keeps the
            # noise (host + folders) and drops the one part that identifies the session. Keep both
            # ends instead: the kind/pass prefix and the filename tail, with an ASCII `...` marking
            # the cut (the full label is still inside the transcript itself).
            safe = safe[:24] + "..." + safe[-53:]
        path = directory / f"{stamp}.{os.getpid()}.{seq}.{safe}.log"
        body = [
            "# citadel ingest — LLM agent session transcript",
            f"time:        {stamp}",
            f"cli:         {cli}",
            f"model:       {config.ingest_model_label()}",
            f"label:       {label or ''}",
            f"returncode:  {returncode}",
            f"duration_s:  {seconds:.1f}",
            f"argv:        {argv}",
        ]
        if note:
            body.append(f"note:        {note}")
        body += ["", "## PROMPT", prompt or "(none)", "", "## STDOUT", out or "(empty)"]
        if err:
            body += ["", "## STDERR", err]
        path.write_text("\n".join(body) + "\n", encoding="utf-8")
        # Shortened for the console (config.display_path): on a network-drive workspace the
        # absolute path is a ~200-char UNC string, and one is announced per SESSION — a chunked
        # source alone prints seven, each wrapping over several rows straight through the live
        # progress display. The path stays workspace-relative, so it is still what you open.
        _echo_stderr(f"LLM transcript: {config.display_path(path)}\n")
    except OSError:
        pass  # logging is diagnostic only — never let it abort an ingest


def _run_session(
    cli: str, argv: list[str], stdin_text: str | None, *, log_label: str | None = None
) -> SessionUsage | None:
    """Run the agentic CLI once in the workspace root (``config.child_cwd()`` — the root in the
    spelling a child process can use; see ``config.NATIVE_FORMS``). Success = the session completed
    without error; the agent's edits are on disk. Returns the session's :class:`SessionUsage`
    when the backend reported one — claude's result envelope, copilot's JSONL stream, agy's
    stream-json events (:func:`_usage_from_stream`); None when the backend named nothing.
    Raises ``RuntimeError`` on timeout, a spawn error, a non-zero exit, or (for claude) an
    ``is_error`` result envelope.

    Output handling is observability-aware but otherwise unchanged: when ``config.LLM_VERBOSE`` is
    set the CLI's output is TEED to the terminal live (:func:`_stream_subprocess`); otherwise it is
    captured silently exactly as before. Either way, when ``config.LLM_LOG_DIR`` is set the full
    session (prompt + stdout/stderr + exit/duration) is written to a transcript file, labelled with
    ``log_label`` (the source key + kind). Both default OFF, so the captured, no-log path — and every
    test that monkeypatches ``subprocess.run`` — is byte-for-byte the original behavior.

    Note: empty stdout is NOT a failure — an agent that legitimately changed nothing prints
    nothing and exits 0 (ingest's snapshot diff then simply shows no changes)."""
    started = time.monotonic()
    try:
        if config.LLM_VERBOSE:
            returncode, out_raw, err_raw = _stream_subprocess(cli, argv, stdin_text)
        else:
            proc = subprocess.run(
                argv,
                input=stdin_text,
                capture_output=True,
                text=True,
                # Force UTF-8 on the piped prompt and the decoded stdout/stderr regardless of the
                # OS locale (e.g. cp1252 on German Windows); errors="replace" keeps a stray
                # undecodable byte from killing the run.
                encoding="utf-8",
                errors="replace",
                timeout=config.LLM_TIMEOUT,
                cwd=config.child_cwd(),
                # The wiki this session must edit is passed EXPLICITLY (ingest's per-source staging
                # copy, via config.wiki_redirect) rather than through a process-global
                # os.environ assignment — two concurrent sessions each get their own.
                env=config.child_env(),
            )
            returncode, out_raw, err_raw = proc.returncode, proc.stdout, proc.stderr
    except subprocess.TimeoutExpired as exc:
        # TimeoutExpired carries whatever was captured before the kill — subprocess.run populates
        # .output/.stderr on the captured path, and _stream_subprocess attaches its streamed chunks
        # via output= — so the timeout transcript logs the PARTIAL session, not an empty one.
        partial_out = _decode_partial(exc.output)
        partial_err = _decode_partial(exc.stderr)
        _write_transcript(
            cli,
            argv,
            stdin_text,
            None,
            partial_out,
            partial_err,
            log_label,
            time.monotonic() - started,
            note=f"timed out after {config.LLM_TIMEOUT}s",
        )
        raise RuntimeError(f"the {cli!r} CLI timed out after {config.LLM_TIMEOUT}s") from exc
    except OSError as exc:
        raise RuntimeError(f"failed to run the {cli!r} CLI: {exc}") from exc

    out = (out_raw or "").strip()
    err = (err_raw or "").strip()
    _write_transcript(
        cli, argv, stdin_text, returncode, out_raw or "", err_raw or "", log_label, time.monotonic() - started
    )

    if cli == "claude":
        # `--output-format json` returns a single result envelope:
        # {"type":"result","is_error":bool,"api_error_status":int|null,"result":str,...}.
        # We read it to detect failure — and, on success, for its cost/usage fields; the
        # agent's WORK is on disk, not in `result`.
        env: dict | None = None
        if out:
            try:
                env = json.loads(out)
            except json.JSONDecodeError:
                env = _last_result_envelope(out)
        if isinstance(env, dict) and env.get("is_error"):
            status = env.get("api_error_status")
            message = (
                "claude CLI error"
                + (f" ({status})" if status else "")
                + f": {env.get('result') or err or 'unknown error'}"
            )
            hint = _hermetic_auth_hint(cli, argv, message)
            error = RuntimeError(message + hint)
            error.hermetic_auth_failure = bool(hint)
            # A failure envelope still reports what the session COST (error_max_turns, API
            # errors) — carry it on the exception so the run total counts the failed spend
            # (the documented "failed sessions included" contract; the manifest stamp stays
            # success-only regardless).
            error.session_usage = _usage_from_claude_envelope(env)
            raise error
        if returncode != 0:
            message = f"the claude CLI failed (exit {returncode}): {(err or out)[:500]}"
            hint = _hermetic_auth_hint(cli, argv, message)
            error = RuntimeError(message + hint)
            error.hermetic_auth_failure = bool(hint)
            error.session_usage = _usage_from_claude_envelope(env)
            raise error
        return _usage_from_claude_envelope(env)

    # copilot / agy (and any unknown CLI): the exit code is the success signal. Their stdout
    # envelopes still carry the model + spend, so parse them on BOTH paths — a failed session's
    # money was spent too, and it is carried on the exception exactly like claude's.
    usage = _usage_from_stream(cli, out)
    if returncode != 0:
        # copilot fails with an empty stderr and the real error inside a session.error event at
        # the END of its JSONL stdout, so prefer that message over the stream's opening noise.
        detail = (_error_from_copilot_jsonl(out) if cli == "copilot" else None) or err or out
        message = f"the {cli!r} CLI failed (exit {returncode}): {detail[:500]}"
        hint = _hermetic_auth_hint(cli, argv, message)
        error = RuntimeError(message + hint)
        error.hermetic_auth_failure = bool(hint)
        error.session_usage = usage
        raise error
    return usage


def run_ingest_session(
    rel_key: str,
    kind: str = "ingest",
    read_path: str | None = None,
    segment: tuple[int, int] | None = None,
    line_range: tuple[int, int] | None = None,
) -> SessionUsage | None:
    """Run the configured agentic CLI once to propagate the raw source ``rel_key`` into the wiki.

    ``kind`` picks the propagation (see :func:`_build_instruction`): ``"ingest"`` folds in a new
    source, ``"reconcile"`` re-ingests a CHANGED source (updating/removing its stale facts),
    ``"image"``/``"image-reconcile"`` VIEW an image source, ``"pdf"``/``"pdf-reconcile"`` read a
    PDF through its pypdf text-layer extraction, and ``"delete"`` strips the provenance of a
    source that was REMOVED from disk.

    ``read_path`` (ingest/reconcile only) is the path to the pre-extracted text of a binary Office
    source — or, when ``segment`` is set, this segment's slice of a large source: when set, the
    agent is told to READ it for content while still citing ``rel_key``, and its directory is
    granted to the CLI alongside any out-of-workspace wiki/raw. ``segment`` is ``(part, total)`` for a
    large source split across passes. ``line_range`` (chunked AUDIO/PDF-extract only) is the
    1-based inclusive line window of the full prepared text this pass processes — the prepared
    file is the WHOLE transcript/extraction, shared by every pass, so locator line numbers never
    rebase.

    The agent's edits under ``config.wiki_dir()`` are the real result — ``ingest`` discovers what
    changed via a filesystem diff. The return value is only the session's best-effort
    :class:`SessionUsage` (the backend's own report of what ran and what it cost: claude's result
    envelope, copilot's JSONL stream, agy's stream-json events; None when nothing was reported or
    the parse missed). Accounting is strictly
    passive: no usage path can fail the session. Raises ``RuntimeError`` (collected per-source
    by ingest) on a missing/failed CLI or a timeout."""
    cli = resolve_cli_name(config.LLM_CLI)
    cli_path = _resolve_cli(cli)
    prompt = _build_instruction(rel_key, kind, read_path, segment, line_range)
    argv, stdin_text = _build_invocation(cli, cli_path, prompt, _external_dirs(rel_key, read_path))
    # The segment goes in FRONT of the source key: when a transcript filename is composed the
    # label is truncated in the MIDDLE (head + tail kept), so on a chunked source "which pass is
    # this" and the source's basename both survive a long absolute key.
    label = f"{kind}.p{segment[0]}of{segment[1]}.{rel_key}" if segment else f"{kind}.{rel_key}"
    hermetic = _hermetic_flags(cli, cli_path)  # session isolation (claude --bare), probe-gated
    try:
        return _run_session(cli, argv + hermetic, stdin_text, log_label=label)
    except RuntimeError as exc:
        # Isolation is a hardening nicety; ingesting at all is the product. On the machines where
        # the skipped personal config is exactly where the CLI keeps its credentials, EVERY
        # session used to die on authentication until the user found CITADEL_HERMETIC=0 — a
        # discoverability trap, since the same CLI works fine interactively. So an auth-shaped
        # failure under hermetic mode is retried ONCE without the isolation flags, loudly. Scoped
        # as tightly as possible: only when flags were actually passed AND the message was
        # auth-shaped, so a real credential problem still fails (twice) instead of looping, and
        # every other failure is re-raised untouched.
        if not hermetic or not getattr(exc, "hermetic_auth_failure", False):
            raise
        print(
            f"  ! {' '.join(hermetic)} was refused with an authentication error; retrying this"
            " session without hermetic isolation (set CITADEL_HERMETIC=0 to skip this retry).",
            file=sys.stderr,
            flush=True,
        )
        return _run_session(cli, argv, stdin_text, log_label=label)
