zettelkasten.outline

zettelkasten.outline

Outline GATHER core: the deterministic bullet-outline material engine.

Where :mod:zettelkasten.review persists an author's editorial overlay and :mod:zettelkasten.claims exposes the per-claim anatomy, this module assembles the raw material an outline auto-population pass writes FROM. It is the deterministic GATHER half of the engine-gathers / agent-drafts split: given the scoped corpus it selects claims (or, when the claim layer is still sparse, the most salient source notes), groups them into themed sections, and for each unit produces a grounded bundle — supporting papers, counterclaims (each with its own evidence), caveats, per-paper CCC skeletons, verbatim quote notes, and claim-to-claim narrative bridges. The Wave-3 DRAFT pass (an LLM agent under the assemble-grounded-outline skill) fills only the generative slots left empty here; it never invents the grounded ones.

The deterministic GATHER half (:func:gather_outline_material) stays pure: no LLM, no network, no file writes. The Wave-3 layer added here builds ON it: an INJECTABLE DRAFT pass (:func:draft) hands the gathered material to an agent that writes the scaffold markdown; an independent, deterministic integrity post-pass (:func:verify_draft) re-checks that draft against the material (never trusting the agent to police itself); a :func:compute_generation_signature cache decides when a redraft is even needed; and :func:build_outline wires GATHER → cache → DRAFT → integrity → a crash-safe, git-backed _reviews/<name>.md write reusing the M3a substrate (:func:graph.atomic_write_text, :func:commit.review_write_lock, :func:commit._schedule_zettel_commit). The scaffold .md is a REGENERABLE DERIVED ARTIFACT (overwrite-on-regenerate; the durable AUTHORED state is the _reviews/<name>.yaml overlay).

Like :mod:zettelkasten.papers, :mod:zettelkasten.claims, and :mod:zettelkasten.review GATHER is parametrized by a get_graph callable. It reuses the locked anatomy/scoring primitives wholesale — :func:claims.claim_anatomy for the heavy per-claim assembly, :func:claims.core_paper_atom / :func:claims.claim_path / :func:claims.claim_relations for the core paper, reading path, and structural edges, :func:syllabus.theme_model for the section partition, and :func:syllabus.situate / :func:syllabus.why_read for the deterministic per-paper badges. It never re-derives a score or an anatomy.

Two restraint guarantees the output upholds:

• Quotes are RETRIEVED, never GENERATED. A quote row carries the verbatim note body + source.page + note id. When a claim or paper has no quote note on file, the bundle emits an explicit 'no quote on file' marker rather than fabricating one.
• Every empty/missing field is an EXPLICIT GAP. Generative slots, absent CCC material, and unresolved references are recorded in gaps so the DRAFT agent and the downstream integrity pass see gaps, not silent omissions.

Integrity NON-GUARANTEE. The deterministic :func:verify_draft post-pass polices STRUCTURAL integrity only — chip resolution, verbatim quoting, and header traceability. It does NOT police prose FAITHFULNESS: fabricated declarative prose that carries only real [chip]s and no double-quoted spans passes every check by construction. That class of failure is governed by the DRAFT contract (the assemble-grounded-outline skill's restraint discipline), not the verifier.

Everything is JSON-serializable (:meth:OutlineMaterial.to_dict, via :func:dataclasses.asdict plus a leaf-coercion pass) and carries every id it references — resolvable note ids in note_ids and work/source ids in a SEPARATE paper_ids field — so an integrity pass and the frontend can resolve each claim, paper, and quote without conflating the two id spaces.

CCCSkeleton dataclass

A single paper's CCC evidence card: deterministic slots, empty generative.

The DETERMINISTIC fields are filled from the graph — the citation label, the :func:syllabus.situate / :func:syllabus.why_read badges, the definition / mechanism / numbers note slots, the quotes from this paper, and note_ids for everything referenced. The generative block is left EMPTY for the Wave-3 agent to draft, and gaps records every empty slot so nothing is a silent omission.

Source code in zettelkasten/outline/materials.py

@dataclass
class CCCSkeleton:
    """A single paper's CCC evidence card: deterministic slots, empty generative.

    The DETERMINISTIC fields are filled from the graph — the citation label, the
    :func:`syllabus.situate` / :func:`syllabus.why_read` ``badges``, the
    ``definition`` / ``mechanism`` / ``numbers`` note slots, the ``quotes`` from
    this paper, and ``note_ids`` for everything referenced. The ``generative``
    block is left EMPTY for the Wave-3 agent to draft, and ``gaps`` records every
    empty slot so nothing is a silent omission.
    """

    paper_id: str
    source_graph: str | None
    title: str
    citation: str
    year: int | None
    authors: list[str]
    owned: bool
    role: str
    relation: str
    importance: float
    badges: dict[str, Any]
    definition: list[dict[str, Any]] = field(default_factory=list)
    mechanism: list[dict[str, Any]] = field(default_factory=list)
    numbers: list[dict[str, Any]] = field(default_factory=list)
    quotes: list[QuoteEvidence] = field(default_factory=list)
    note_ids: list[str] = field(default_factory=list)
    generative: dict[str, str] = field(default_factory=dict)
    gaps: list[dict[str, Any]] = field(default_factory=list)

ClaimBundle dataclass

The grounded material for one outline unit (a claim, or a fallback note).

Carries the unit's text + id, its supporting-paper CCC cards (stronger on top), its counterclaims (each WITH its own evidence) and caveats, the core paper card, its own verbatim quotes (or marker), the claim-to-claim narrative bridges + reading path, and the explicit gaps. note_ids lists every note this bundle references so the integrity pass / frontend can resolve them. source is "claim" for a claim-layer unit or "fallback" for a claim-sparse source-note unit; the shape is identical either way.

Source code in zettelkasten/outline/materials.py

@dataclass
class ClaimBundle:
    """The grounded material for one outline unit (a claim, or a fallback note).

    Carries the unit's text + id, its supporting-paper CCC cards (stronger on
    top), its counterclaims (each WITH its own evidence) and caveats, the core
    paper card, its own verbatim quotes (or marker), the claim-to-claim narrative
    bridges + reading path, and the explicit ``gaps``. ``note_ids`` lists every
    note this bundle references so the integrity pass / frontend can resolve them.
    ``source`` is ``"claim"`` for a claim-layer unit or ``"fallback"`` for a
    claim-sparse source-note unit; the shape is identical either way.
    """

    uid: str
    id: str
    graph: str
    title: str
    type: str
    text: str
    status: str
    strength: float
    salience: float
    year: int | None
    theme: str
    source: str
    supporting: list[CCCSkeleton] = field(default_factory=list)
    core_paper: CCCSkeleton | None = None
    counterclaims: list[dict[str, Any]] = field(default_factory=list)
    caveats: list[dict[str, Any]] = field(default_factory=list)
    quotes: list[QuoteEvidence] = field(default_factory=list)
    bridges: list[dict[str, Any]] = field(default_factory=list)
    reading_path: list[dict[str, Any]] = field(default_factory=list)
    note_ids: list[str] = field(default_factory=list)
    paper_ids: list[str] = field(default_factory=list)
    gaps: list[dict[str, Any]] = field(default_factory=list)

OutlineMaterial dataclass

The full deterministic GATHER result for an outline scope.

sections are the themed groups (in :func:syllabus.theme_model order), unplaced the bundles whose unit sits in no theme, gaps the outline-level gap markers, note_ids the union of every resolvable note ref any bundle references — each a graph-qualified "<graph>::<id>" token (see :func:_qualify) so a chip names its EXACT source graph and resolves regardless of the review scope — and paper_ids the union of the work/source-graph ids the bundles reference (bare folder names like "alpha", NOT note ids) — kept in a SEPARATE field so the integrity pass never tries to resolve a paper id as a note, while the frontend can still resolve every claim, paper, and quote without re-walking the graph.

Source code in zettelkasten/outline/materials.py

@dataclass
class OutlineMaterial:
    """The full deterministic GATHER result for an outline scope.

    ``sections`` are the themed groups (in :func:`syllabus.theme_model` order),
    ``unplaced`` the bundles whose unit sits in no theme, ``gaps`` the
    outline-level gap markers, ``note_ids`` the union of every *resolvable note
    ref* any bundle references — each a graph-qualified ``"<graph>::<id>"`` token
    (see :func:`_qualify`) so a chip names its EXACT source graph and resolves
    regardless of the review scope — and ``paper_ids`` the union of the
    work/source-graph ids the bundles reference (bare folder names like
    ``"alpha"``, NOT note ids) — kept in a SEPARATE field so the integrity pass
    never tries to resolve a paper id as a note, while the frontend can still
    resolve every claim, paper, and quote without re-walking the graph.
    """

    scope: dict[str, Any]
    theme_source: str
    sections: list[OutlineSection] = field(default_factory=list)
    unplaced: list[ClaimBundle] = field(default_factory=list)
    gaps: list[dict[str, Any]] = field(default_factory=list)
    note_ids: list[str] = field(default_factory=list)
    paper_ids: list[str] = field(default_factory=list)
    # Structured explicit signal for the NESTED composition (see
    # :func:`_resolve_nested_ancestors`): present ONLY when a nesting selection
    # was requested (an explicit ``ancestor_uids`` set OR a non-zero
    # ``ancestor_levels``). ``None`` on a flat outline, and dropped from
    # :meth:`to_dict` so a flat build's payload stays byte-identical.
    nesting_degradation: "dict[str, Any] | None" = None

    def to_dict(self) -> dict[str, Any]:
        """A plain, guaranteed ``json.dumps``-able dict of the whole material tree.

        Runs :func:`dataclasses.asdict` and then coerces any non-JSON-native
        leaf to a string — most importantly a ``source.page`` that a YAML
        author wrote as a bare ISO date (parsed to ``datetime.date``), which
        ``asdict`` preserves and ``json.dumps`` would otherwise reject.

        The nested-composition :class:`OutlineSection` fields ``level``/``tier``
        are emitted ONLY for a section that is NOT the flat default (i.e.
        ``level != 2`` or ``tier != "leaf"``). A flat/single-spine section drops
        both keys entirely, so a non-nested build's payload — the DRAFT prompt,
        the cached artifact, every downstream consumer — is BYTE-IDENTICAL to
        before the nested mode existed (the ``ancestor_levels == 0`` no-op).
        """
        data = _json_safe(asdict(self))
        # A flat outline carries no nesting selection, so the degradation block is
        # ``None`` — drop the key entirely so the flat payload is byte-identical to
        # before the nested-composition mode (mirrors the level/tier drop below).
        if data.get("nesting_degradation") is None:
            data.pop("nesting_degradation", None)
        for sec in data.get("sections", []):
            if not isinstance(sec, dict):
                continue
            if sec.get("level") == 2 and sec.get("tier") == "leaf":
                sec.pop("level", None)
                sec.pop("tier", None)
        return data

    def project_section(
        self, section: "OutlineSection | None"
    ) -> "OutlineMaterial":
        """A single-slice :class:`OutlineMaterial` for ONE section (or the unplaced bucket).

        Returns a new material carrying ONLY ``section`` (with ``unplaced`` empty),
        or — when ``section`` is ``None`` — only the ``unplaced`` bundles (with
        ``sections`` empty), and with ``note_ids`` / ``paper_ids`` NARROWED to just
        the refs that slice actually cites. This bounds the per-call DRAFT payload
        so a large outline can be drafted section-by-section instead of stuffing the
        entire (multi-megabyte) material into one prompt and overflowing the draft
        model's context window (the failure that makes the agent lose its material
        and emit a "re-paste the GATHER material" refusal). The shared ``scope`` and
        ``theme_source`` are preserved verbatim so a projected section fingerprints
        (:func:`_section_material_signature`) identically to its slice in the whole
        material; ``gaps`` (outline-level) and ``nesting_degradation`` are dropped —
        they are whole-outline framing, not per-section content.
        """
        bundles = list(section.claims) if section is not None else list(self.unplaced)
        concepts = list(section.concepts) if section is not None else []
        note_ids: set[str] = set()
        paper_ids: set[str] = set()
        for b in bundles:
            note_ids.update(b.note_ids)
            paper_ids.update(b.paper_ids)
        for c in concepts:
            if c.uid:
                note_ids.add(c.uid)
            paper_ids.update(c.paper_ids)
        return OutlineMaterial(
            scope=self.scope,
            theme_source=self.theme_source,
            sections=[section] if section is not None else [],
            unplaced=[] if section is not None else bundles,
            gaps=[],
            note_ids=sorted(note_ids),
            paper_ids=sorted(paper_ids),
            nesting_degradation=None,
        )

to_dict

to_dict() -> dict[str, Any]

A plain, guaranteed json.dumps-able dict of the whole material tree.

Runs :func:dataclasses.asdict and then coerces any non-JSON-native leaf to a string — most importantly a source.page that a YAML author wrote as a bare ISO date (parsed to datetime.date), which asdict preserves and json.dumps would otherwise reject.

The nested-composition :class:OutlineSection fields level/tier are emitted ONLY for a section that is NOT the flat default (i.e. level != 2 or tier != "leaf"). A flat/single-spine section drops both keys entirely, so a non-nested build's payload — the DRAFT prompt, the cached artifact, every downstream consumer — is BYTE-IDENTICAL to before the nested mode existed (the ancestor_levels == 0 no-op).

Source code in zettelkasten/outline/materials.py

def to_dict(self) -> dict[str, Any]:
    """A plain, guaranteed ``json.dumps``-able dict of the whole material tree.

    Runs :func:`dataclasses.asdict` and then coerces any non-JSON-native
    leaf to a string — most importantly a ``source.page`` that a YAML
    author wrote as a bare ISO date (parsed to ``datetime.date``), which
    ``asdict`` preserves and ``json.dumps`` would otherwise reject.

    The nested-composition :class:`OutlineSection` fields ``level``/``tier``
    are emitted ONLY for a section that is NOT the flat default (i.e.
    ``level != 2`` or ``tier != "leaf"``). A flat/single-spine section drops
    both keys entirely, so a non-nested build's payload — the DRAFT prompt,
    the cached artifact, every downstream consumer — is BYTE-IDENTICAL to
    before the nested mode existed (the ``ancestor_levels == 0`` no-op).
    """
    data = _json_safe(asdict(self))
    # A flat outline carries no nesting selection, so the degradation block is
    # ``None`` — drop the key entirely so the flat payload is byte-identical to
    # before the nested-composition mode (mirrors the level/tier drop below).
    if data.get("nesting_degradation") is None:
        data.pop("nesting_degradation", None)
    for sec in data.get("sections", []):
        if not isinstance(sec, dict):
            continue
        if sec.get("level") == 2 and sec.get("tier") == "leaf":
            sec.pop("level", None)
            sec.pop("tier", None)
    return data

project_section

project_section(section: 'OutlineSection | None') -> 'OutlineMaterial'

A single-slice :class:OutlineMaterial for ONE section (or the unplaced bucket).

Returns a new material carrying ONLY section (with unplaced empty), or — when section is None — only the unplaced bundles (with sections empty), and with note_ids / paper_ids NARROWED to just the refs that slice actually cites. This bounds the per-call DRAFT payload so a large outline can be drafted section-by-section instead of stuffing the entire (multi-megabyte) material into one prompt and overflowing the draft model's context window (the failure that makes the agent lose its material and emit a "re-paste the GATHER material" refusal). The shared scope and theme_source are preserved verbatim so a projected section fingerprints (:func:_section_material_signature) identically to its slice in the whole material; gaps (outline-level) and nesting_degradation are dropped — they are whole-outline framing, not per-section content.

Source code in zettelkasten/outline/materials.py

def project_section(
    self, section: "OutlineSection | None"
) -> "OutlineMaterial":
    """A single-slice :class:`OutlineMaterial` for ONE section (or the unplaced bucket).

    Returns a new material carrying ONLY ``section`` (with ``unplaced`` empty),
    or — when ``section`` is ``None`` — only the ``unplaced`` bundles (with
    ``sections`` empty), and with ``note_ids`` / ``paper_ids`` NARROWED to just
    the refs that slice actually cites. This bounds the per-call DRAFT payload
    so a large outline can be drafted section-by-section instead of stuffing the
    entire (multi-megabyte) material into one prompt and overflowing the draft
    model's context window (the failure that makes the agent lose its material
    and emit a "re-paste the GATHER material" refusal). The shared ``scope`` and
    ``theme_source`` are preserved verbatim so a projected section fingerprints
    (:func:`_section_material_signature`) identically to its slice in the whole
    material; ``gaps`` (outline-level) and ``nesting_degradation`` are dropped —
    they are whole-outline framing, not per-section content.
    """
    bundles = list(section.claims) if section is not None else list(self.unplaced)
    concepts = list(section.concepts) if section is not None else []
    note_ids: set[str] = set()
    paper_ids: set[str] = set()
    for b in bundles:
        note_ids.update(b.note_ids)
        paper_ids.update(b.paper_ids)
    for c in concepts:
        if c.uid:
            note_ids.add(c.uid)
        paper_ids.update(c.paper_ids)
    return OutlineMaterial(
        scope=self.scope,
        theme_source=self.theme_source,
        sections=[section] if section is not None else [],
        unplaced=[] if section is not None else bundles,
        gaps=[],
        note_ids=sorted(note_ids),
        paper_ids=sorted(paper_ids),
        nesting_degradation=None,
    )

OutlineSection dataclass

A themed section: a theme, its ordered claim bundles, and framing concepts.

concepts are condensed framing tags (see :class:ConceptTag) rendered under the section intro — distinct from claims, which are the full units.

level is the markdown header depth the section renders at (2 = ##, the flat/single-spine default). tier is "leaf" for a normal claim-bearing section (the flat default) or "ancestor" for a nested composition-tier framing section (see :func:_resolve_nested_ancestors): an ancestor tier carries no claim bundles — the DRAFT agent synthesizes 1–2 "lecture-note" framing bullets from its subtree. Both fields default to the flat values, so a non-nested outline (ancestor_levels == 0) is byte-identical to before.

Source code in zettelkasten/outline/materials.py

@dataclass
class OutlineSection:
    """A themed section: a theme, its ordered claim bundles, and framing concepts.

    ``concepts`` are condensed framing tags (see :class:`ConceptTag`) rendered
    under the section intro — distinct from ``claims``, which are the full units.

    ``level`` is the markdown header depth the section renders at (``2`` = ``## ``,
    the flat/single-spine default). ``tier`` is ``"leaf"`` for a normal
    claim-bearing section (the flat default) or ``"ancestor"`` for a nested
    composition-tier framing section (see :func:`_resolve_nested_ancestors`): an
    ancestor tier carries no claim bundles — the DRAFT agent synthesizes 1–2
    "lecture-note" framing bullets from its subtree. Both fields default to the
    flat values, so a non-nested outline (``ancestor_levels == 0``) is
    byte-identical to before.
    """

    theme_id: str
    label: str
    source: str
    claims: list[ClaimBundle] = field(default_factory=list)
    concepts: list[ConceptTag] = field(default_factory=list)
    gaps: list[dict[str, Any]] = field(default_factory=list)
    level: int = 2
    tier: str = "leaf"

QuoteEvidence dataclass

A verbatim quote note (RETRIEVED) or an explicit no-quote marker.

When present is True the row carries the quote note's verbatim body plus source.page and the note id so the frontend can resolve and the integrity pass can verify it. When present is False it is the explicit 'no quote on file' marker (everything else empty) — never a fabricated quote.

Source code in zettelkasten/outline/materials.py

@dataclass
class QuoteEvidence:
    """A verbatim quote note (RETRIEVED) or an explicit no-quote marker.

    When ``present`` is ``True`` the row carries the quote note's verbatim
    ``body`` plus ``source.page`` and the note ``id`` so the frontend can resolve
    and the integrity pass can verify it. When ``present`` is ``False`` it is the
    explicit ``'no quote on file'`` marker (everything else empty) — never a
    fabricated quote.
    """

    id: str
    graph: str
    title: str
    body: str
    page: Any
    grounded: bool
    relation: str
    present: bool = True
    marker: str = ""

Violation dataclass

A single integrity-post-pass finding against a drafted scaffold.

kind is one of "chip" (an unresolved [note-id] chip), "quote" (a quoted span not found verbatim in any referenced quote note), "header" (a claim header that traces to no claim/finding unit), or "structure" (the header skeleton drifted from the expected section headers — in a FLAT outline the ## themes; in a NESTED outline the composition tier tree, where each tier header is known by its (label, depth) — a missing/reordered tier, or a claim placed at a tier depth). ref is the offending token/text and line its 1-based line.

Source code in zettelkasten/outline/verify.py

@dataclass
class Violation:
    """A single integrity-post-pass finding against a drafted scaffold.

    ``kind`` is one of ``"chip"`` (an unresolved ``[note-id]`` chip),
    ``"quote"`` (a quoted span not found verbatim in any referenced quote note),
    ``"header"`` (a claim header that traces to no claim/finding unit), or
    ``"structure"`` (the header skeleton drifted from the expected section headers
    — in a FLAT outline the ``## `` themes; in a NESTED outline the composition
    tier tree, where each tier header is known by its ``(label, depth)`` — a
    missing/reordered tier, or a claim placed at a tier depth). ``ref`` is the
    offending token/text and ``line`` its 1-based line.
    """

    kind: str
    detail: str
    ref: str = ""
    line: "int | None" = None

OutlineConflictError

Bases: RuntimeError

The persisted scaffold no longer matches the signature the client saw.

Raised by :func:fix_outline_violation when the caller passes the artifact signature it loaded and the currently persisted draft has since moved on (a regenerate/deepen/edit happened in another tab, or a 503-retry races a completed write). Mapping this to an HTTP 409 lets the UI refuse to apply a Fix/Remove against stale line numbers — which could delete the wrong content — and prompt a refresh instead.

Source code in zettelkasten/outline/__init__.py

class OutlineConflictError(RuntimeError):
    """The persisted scaffold no longer matches the signature the client saw.

    Raised by :func:`fix_outline_violation` when the caller passes the artifact
    signature it loaded and the currently persisted draft has since moved on (a
    regenerate/deepen/edit happened in another tab, or a 503-retry races a
    completed write). Mapping this to an HTTP 409 lets the UI refuse to apply a
    Fix/Remove against stale line numbers — which could delete the wrong content
    — and prompt a refresh instead.
    """

narrative_ordering

narrative_ordering(bundles: list[ClaimBundle], edges: list[tuple[str, str, str]], *, year_of: Callable[[str], int | None] | None = None, salience_of: Callable[[str], float] | None = None) -> list[ClaimBundle]

Order a section's units foundations-first, counterclaims nested at target.

A pure function over the bundles and their (src_uid, relation, dst_uid) structural edges. Ordering rules:

• Foundations first. Lineage edges (depends-on / extends / derives-from / specializes / prerequisite-of / generalizes) establish a foundation→dependent partial order; a topological pass places every foundation before the units that build on it.
• Chronological + salient within a tier. Among units with no ordering edge between them, the earlier (by year) comes first, then the more salient, then the uid — so the order is fully deterministic. Undated units sort last within their tier.
• Counterclaims nested. A unit that contests another in-section unit (contradicts / responds-to / qualifies) is placed immediately after its target rather than floating to a tier of its own.

Cycles are broken safely (the smallest remaining unit by the tie-break key is emitted). Units absent from bundles are ignored. year_of / salience_of override the per-bundle attributes (handy for tests).

Source code in zettelkasten/outline/gather.py

def narrative_ordering(
    bundles: list[ClaimBundle],
    edges: list[tuple[str, str, str]],
    *,
    year_of: Callable[[str], int | None] | None = None,
    salience_of: Callable[[str], float] | None = None,
) -> list[ClaimBundle]:
    """Order a section's units foundations-first, counterclaims nested at target.

    A pure function over the bundles and their ``(src_uid, relation, dst_uid)``
    structural edges. Ordering rules:

    * **Foundations first.** Lineage edges (``depends-on`` / ``extends`` /
      ``derives-from`` / ``specializes`` / ``prerequisite-of`` / ``generalizes``)
      establish a foundation→dependent partial order; a topological pass places
      every foundation before the units that build on it.
    * **Chronological + salient within a tier.** Among units with no ordering
      edge between them, the earlier (by ``year``) comes first, then the more
      salient, then the uid — so the order is fully deterministic. Undated units
      sort last within their tier.
    * **Counterclaims nested.** A unit that contests another in-section unit
      (``contradicts`` / ``responds-to`` / ``qualifies``) is placed immediately
      after its target rather than floating to a tier of its own.

    Cycles are broken safely (the smallest remaining unit by the tie-break key is
    emitted). Units absent from ``bundles`` are ignored. ``year_of`` /
    ``salience_of`` override the per-bundle attributes (handy for tests).
    """
    uid_to_item: dict[str, ClaimBundle] = {}
    for b in bundles:
        uid_to_item[b.uid] = b
    uid_set = set(uid_to_item)

    def _year(uid: str) -> int:
        if year_of is not None:
            y = year_of(uid)
        else:
            y = getattr(uid_to_item[uid], "year", None)
        return y if y is not None else _UNDATED_YEAR

    def _sal(uid: str) -> float:
        if salience_of is not None:
            return salience_of(uid)
        return getattr(uid_to_item[uid], "salience", 0.0)

    def _key(uid: str) -> tuple[int, float, str]:
        return (_year(uid), -_sal(uid), uid)

    # Counter units (src contests dst, both in section) are nested, not primary.
    counter_target: dict[str, str] = {}
    for s, rel, d in edges:
        if rel in COUNTER_RELATIONS and s in uid_set and d in uid_set and s != d:
            counter_target.setdefault(s, d)

    primary = [u for u in uid_to_item if u not in counter_target]
    primary_set = set(primary)

    deps: dict[str, set[str]] = {u: set() for u in primary}
    dependents: dict[str, set[str]] = {u: set() for u in primary}
    for s, rel, d in edges:
        side = _LINEAGE_FOUNDATION_SIDE.get(rel)
        if side is None:
            continue
        foundation, dependent = (d, s) if side == "dst" else (s, d)
        if foundation in primary_set and dependent in primary_set and foundation != dependent:
            if foundation not in deps[dependent]:
                deps[dependent].add(foundation)
                dependents[foundation].add(dependent)

    indeg = {u: len(deps[u]) for u in primary}
    ready = sorted([u for u in primary if indeg[u] == 0], key=_key)
    order: list[str] = []
    placed: set[str] = set()
    while ready:
        u = ready.pop(0)
        order.append(u)
        placed.add(u)
        newly = []
        for dep in dependents[u]:
            indeg[dep] -= 1
            if indeg[dep] == 0:
                newly.append(dep)
        if newly:
            ready = sorted(ready + newly, key=_key)
    # Any units left (a lineage cycle) are emitted by the same tie-break.
    for u in sorted((u for u in primary if u not in placed), key=_key):
        order.append(u)

    counters_by_target: dict[str, list[str]] = {}
    for c, tgt in counter_target.items():
        counters_by_target.setdefault(tgt, []).append(c)

    result: list[str] = []
    emitted: set[str] = set()

    def _emit(u: str) -> None:
        # Place a unit, then DEPTH-FIRST place its counters (and their counters)
        # so a chain ``a responds-to b responds-to c`` nests as ``c, b, a`` —
        # each counter lands immediately after its target. ``emitted`` guards
        # against cycles and re-emission (a counter is never placed twice).
        if u in emitted:
            return
        result.append(u)
        emitted.add(u)
        for c in sorted(counters_by_target.get(u, [])):
            _emit(c)

    for u in order:
        _emit(u)
    # Counters whose target chain was never placed (target missing, or a cycle
    # among counters) trail in a stable order.
    for c in sorted(counter_target):
        _emit(c)

    return [uid_to_item[u] for u in result if u in uid_to_item]

verify_draft

verify_draft(markdown: str, material: OutlineMaterial) -> list[Violation]

Independently re-check a drafted scaffold against the gathered material.

Deterministic and LLM-free — it does NOT trust the agent to police itself (the skill contract is the drafting discipline; THIS is the verifier):

Code-fenced blocks and inline code spans are MASKED before scanning (see :func:_mask_code), so a scaffold that documents its own conventions — a header explaining the [graph::id] chip form, a "verbatim" example — does not self-trip the chip/quote scans. Reserved ALL-CAPS markers ([GAP], [TODO], induction-head [A]/[B] notation; see :func:_is_sentinel) are structural, not references, and are skipped by the chip scan entirely.

• chips — every [note-id] chip must resolve to an id in material.note_ids (a note) or material.paper_ids (a paper); unresolved chips are flagged. Reserved sentinel markers are exempt.
• quotes — every double-quoted span (scanned over the WHOLE document, so a multi-line span is inspected too; smart double-quotes are folded to straight first so a MISMATCHED-delimiter span — smart-open + straight-close, or the reverse — cannot evade the scan) must be a VERBATIM excerpt of a referenced quote note's body: whole-quote equality OR word-boundary containment of the span's tokens (see :func:_verbatim_contained), NOT a raw substring. A span that matches nothing — a fabrication, or a cherry-picked/recombined fragment that never ran contiguously — is flagged.
• headers — a markdown header that carries a chip is a CLAIM header; if none of its chips trace to a claim/finding UNIT in the material it is flagged as untraceable. A header with no chip is structural (the honesty header, a theme/section title) and is not a claim header.

NON-GUARANTEE: this deterministic pass polices structural integrity (chip resolution, verbatim quoting, header traceability) — it does NOT police prose FAITHFULNESS. Fabricated declarative prose that carries only real [chip]s and no double-quoted spans passes every check here by construction; guarding against that is the DRAFT contract's job (the assemble-grounded-outline skill's restraint discipline), not the verifier's.

Returns the violations in document order; the caller marks them (e.g. :func:build_outline appends them to the artifact) rather than silently passing a draft that failed the check.

Source code in zettelkasten/outline/verify.py

def verify_draft(markdown: str, material: OutlineMaterial) -> list[Violation]:
    """Independently re-check a drafted scaffold against the gathered material.

    Deterministic and LLM-free — it does NOT trust the agent to police itself
    (the skill contract is the drafting discipline; THIS is the verifier):

    Code-fenced blocks and inline ``code`` spans are MASKED before scanning (see
    :func:`_mask_code`), so a scaffold that documents its own conventions — a
    header explaining the ``[graph::id]`` chip form, a ``"verbatim"`` example —
    does not self-trip the chip/quote scans. Reserved ALL-CAPS markers
    (``[GAP]``, ``[TODO]``, induction-head ``[A]``/``[B]`` notation; see
    :func:`_is_sentinel`) are structural, not references, and are skipped by the
    chip scan entirely.

    * **chips** — every ``[note-id]`` chip must resolve to an id in
      ``material.note_ids`` (a note) or ``material.paper_ids`` (a paper);
      unresolved chips are flagged. Reserved sentinel markers are exempt.
    * **quotes** — every double-quoted span (scanned over the WHOLE document, so
      a multi-line span is inspected too; smart double-quotes are folded to
      straight first so a MISMATCHED-delimiter span — smart-open + straight-close,
      or the reverse — cannot evade the scan) must be a VERBATIM excerpt of a
      referenced quote note's body: whole-quote equality OR word-boundary
      containment of the span's tokens (see :func:`_verbatim_contained`), NOT a
      raw substring. A span that matches nothing — a fabrication, or a
      cherry-picked/recombined fragment that never ran contiguously — is flagged.
    * **headers** — a markdown header that carries a chip is a CLAIM header; if
      none of its chips trace to a claim/finding UNIT in the material it is
      flagged as untraceable. A header with no chip is structural (the honesty
      header, a theme/section title) and is not a claim header.

    NON-GUARANTEE: this deterministic pass polices structural integrity (chip
    resolution, verbatim quoting, header traceability) — it does NOT police prose
    FAITHFULNESS. Fabricated declarative prose that carries only real ``[chip]``s
    and no double-quoted spans passes every check here by construction; guarding
    against that is the DRAFT contract's job (the ``assemble-grounded-outline``
    skill's restraint discipline), not the verifier's.

    Returns the violations in document order; the caller marks them (e.g.
    :func:`build_outline` appends them to the artifact) rather than silently
    passing a draft that failed the check.
    """
    note_ids = set(material.note_ids)
    paper_ids = set(material.paper_ids)
    resolvable = note_ids | paper_ids
    # Note refs are now graph-qualified ``graph::id`` tokens; paper ids stay bare
    # folder names. Tolerate a chip that dropped its ``graph::`` prefix (the agent
    # occasionally emits a bare ``[id]``) by also matching the bare id-part of any
    # resolvable token — the verifier polices fabrication, not graph precision.
    resolvable_bare = {_chip_bare_id(t) for t in resolvable}
    unit_ids = _material_unit_ids(material)
    unit_bare_ids = {_chip_bare_id(u) for u in unit_ids}
    quote_norms = _material_quote_norms(material)

    violations: list[Violation] = []

    # ── structure: no header may exceed markdown's 6-level maximum ──────────────
    # A line of 7+ ``#`` then a space is an UNPARSEABLE header: markdown supports at
    # most 6 heading levels, so ``_HEADING_RE`` (``#{1,6}``) never matches it and
    # ``_section_headings`` (with every tier/depth check that consumes it) is BLIND
    # to it. Without this explicit scan a fabricated ``####### X`` / ``####### Unplaced``
    # — even at the common 2-ancestor depth — would pass with ZERO violations (an
    # anti-fabrication bypass). Match ``_section_headings``' code-fence discipline so
    # a ``#######`` inside a ``` block (documentation, not a header) is not flagged.
    in_fence = False
    fence_char = ""
    for i, line in enumerate(markdown.splitlines()):
        fm = _FENCE_RE.match(line)
        if fm:
            tok = fm.group(1)
            if not in_fence:
                in_fence, fence_char = True, tok[0]
            elif line.strip()[:1] == fence_char:
                in_fence = False
            continue
        if in_fence:
            continue
        if _OVERDEEP_HEADING_RE.match(line):
            violations.append(Violation(
                kind="structure",
                detail=(
                    "header exceeds maximum depth (markdown supports at most 6 "
                    "levels) — a header with 7+ '#' is unparseable"
                ),
                ref=line.lstrip("#").strip(),
                line=i + 1,
            ))

    # ── structure: the ``## `` skeleton must be the author's SETTLED themes ─────
    # ``## `` headers are reserved for the settled theme labels (in
    # ``material.sections`` order); declarative CLAIM headers live at ``### ``.
    # This is what lets the frontend map a theme → its section by exact label and
    # never falls back to a fragile positional guess. A trailing ``## Unplaced``
    # bucket is allowed; everything else at ``## `` is a claim/topic in the wrong
    # place (or a renamed/reordered/missing theme), and is flagged.
    expected_labels = [s.label for s in material.sections]
    expected_levels = [getattr(s, "level", 2) or 2 for s in material.sections]
    # Nested mode is decided by the TIER signal, not depth: an outline is nested
    # iff it stacked at least one ``tier == "ancestor"`` framing tier. A depth test
    # (``lvl != 2``) MISSES a single kept ancestor over a dimensionless (leafless)
    # subject — the only section is the ancestor at level 2 — and would take the
    # flat branch, letting a shallow ``Unplaced`` pass unchecked. A true flat
    # outline has no ancestor tiers, so ``ancestor_count == 0`` and behavior (and
    # the byte-identical flat serialization) is unchanged.
    ancestor_count = sum(
        1 for s in material.sections
        if getattr(s, "tier", "leaf") == "ancestor"
    )
    nested = ancestor_count > 0
    if expected_labels and not nested:
        expected_set = set(expected_labels)
        h2_seq = [
            (i + 1, text)
            for (i, lvl, text) in _section_headings(markdown)
            if lvl == 2
        ]
        seen_theme: list[str] = []
        for lineno, text in h2_seq:
            if text in expected_set:
                seen_theme.append(text)
            elif text != "Unplaced":
                violations.append(Violation(
                    kind="structure",
                    detail=(
                        f"'## {text}' is not a settled theme header — a claim or "
                        "topic belongs at '### '; '## ' is reserved for the "
                        "author's themes"
                    ),
                    ref=text,
                    line=lineno,
                ))
        if seen_theme != expected_labels:
            missing = [lbl for lbl in expected_labels if lbl not in set(seen_theme)]
            if missing:
                detail = (
                    "scaffold is missing settled theme header(s): "
                    + "; ".join(missing)
                )
            else:
                detail = (
                    "settled theme headers are out of order — expected: "
                    + " | ".join(expected_labels)
                    + "; got: "
                    + " | ".join(seen_theme)
                )
            violations.append(Violation(kind="structure", detail=detail))
    elif expected_labels and nested:
        # NESTED composition: the ``## ``/``### ``/``#### ``… skeleton is the
        # composition tier tree, so a section header is known by (label, DEPTH) —
        # each tier must appear at its own markdown level (root shallowest), in
        # order. This is the ONLY hierarchy-aware rule; the GLOBAL chip/quote checks
        # below still police every tier (ancestor framing bullets included). A
        # trailing ``Unplaced`` bucket stays allowed AT THE LEAF DEPTH, where the
        # membership-unrouted claims live; an ``Unplaced`` header at a shallower
        # ancestor depth is a misplaced tier and is flagged. Header depths in the
        # tier-level band that match no tier — a flattened/misplaced tier or a
        # claim promoted to a header — are flagged.
        expected_pairs = list(zip(expected_labels, expected_levels))
        expected_pair_set = set(expected_pairs)
        # The TRUE leaf depth is the depth the subject's leaf partition occupies:
        # ``2 + <kept ancestor count>``. Compute it DIRECTLY from the ancestor
        # tiers (``ancestor_count`` above) rather than as ``max(expected_levels)`` —
        # the latter COLLAPSES onto the deepest ANCESTOR depth when the subject
        # materialized NO leaf sections (a dimensionless subject whose claims all
        # fell to Unplaced), which would falsely accept a shallow ``Unplaced`` and
        # flag the real leaf-depth one.
        leaf_depth = 2 + ancestor_count
        # The DEEPEST legal nested header: leaf sections sit at ``leaf_depth`` and
        # any claim sub-header the leaf partition emits is exactly ONE deeper. A
        # header BELOW this is a claim promoted too deep (or a bogus deep bucket)
        # and is flagged — the pre-fix scan band capped at the leaf depth and
        # silently skipped everything deeper, so a ``##### `` claim or ``##### ``
        # Unplaced slipped through UNSCANNED.
        deepest_legal = leaf_depth + 1
        seen_pairs: list[tuple[str, int]] = []
        seen_unplaced = False
        for (i, lvl, text) in _section_headings(markdown):
            if lvl < 2:
                continue
            if text == "Unplaced":
                # The Unplaced bucket holds LEAF content (membership-unrouted
                # claims in no dimension), so under nesting it renders at the LEAF
                # depth (``2 + <kept ancestor count>``) and EXACTLY ONCE. An
                # Unplaced header at any other depth is a misplaced tier; a second
                # one at the leaf depth is a duplicate bucket — both flagged.
                if lvl == leaf_depth and not seen_unplaced:
                    seen_unplaced = True
                    continue
                if lvl == leaf_depth:
                    detail = (
                        f"duplicate '{'#' * lvl} Unplaced' bucket — a nested "
                        "scaffold may carry at most ONE Unplaced bucket, at the "
                        f"leaf tier depth ('{'#' * leaf_depth} ')"
                    )
                else:
                    detail = (
                        f"'{'#' * lvl} Unplaced' is not at the leaf tier depth — "
                        "the Unplaced bucket holds leaf content and must sit at "
                        f"the leaf tier depth ('{'#' * leaf_depth} ')"
                    )
                violations.append(Violation(
                    kind="structure", detail=detail, ref=text, line=i + 1,
                ))
                continue
            if lvl > deepest_legal:
                violations.append(Violation(
                    kind="structure",
                    detail=(
                        f"'{'#' * lvl} {text}' is deeper than the deepest legal "
                        f"nested header ('{'#' * deepest_legal} ') — a claim "
                        "promoted too deep or a stray sub-header"
                    ),
                    ref=text,
                    line=i + 1,
                ))
                continue
            if lvl > leaf_depth:
                # A legal claim sub-header ONE level below the leaf tier; not a
                # composition tier, so it is exempt from tier matching. Its
                # grounding (if it carries a chip) is policed by the global
                # claim-header traceability check below.
                continue
            pair = (text, lvl)
            if pair in expected_pair_set:
                seen_pairs.append(pair)
            else:
                violations.append(Violation(
                    kind="structure",
                    detail=(
                        f"'{'#' * lvl} {text}' is not a composition-tier header at "
                        "this depth — every tier header must match a spine tier "
                        "label at its tier depth (root shallowest)"
                    ),
                    ref=text,
                    line=i + 1,
                ))
        if seen_pairs != expected_pairs:
            seen_set = set(seen_pairs)
            missing = [lbl for (lbl, lv) in expected_pairs if (lbl, lv) not in seen_set]
            if missing:
                detail = (
                    "scaffold is missing composition-tier header(s): "
                    + "; ".join(missing)
                )
            else:
                detail = (
                    "composition-tier headers are out of order — expected: "
                    + " | ".join(f"{'#' * lv} {lbl}" for lbl, lv in expected_pairs)
                    + "; got: "
                    + " | ".join(f"{'#' * lv} {lbl}" for lbl, lv in seen_pairs)
                )
            violations.append(Violation(kind="structure", detail=detail))

    # double-quoted spans are verbatim source text — scanned over the WHOLE
    # document (not per line) so a multi-line span is caught, with its 1-based
    # start line recovered for the violation. Smart double-quotes are folded to
    # straight FIRST so a mismatched-delimiter span (smart-open + straight-close,
    # or the reverse) is still inspected; the fold is length-preserving so byte
    # offsets still map back to the original document's line numbers.
    # NON-GUARANTEE: a quoted single common word that happens to appear verbatim
    # in some note body passes by construction — a known, low-value gap (the
    # token-containment test cannot distinguish a deliberate citation from an
    # incidental word match), not a faithfulness guarantee.
    # Code-fenced blocks and inline ``code`` spans are masked to spaces FIRST so
    # syntax the scaffold documents about itself (the conventions header's
    # ``[graph::id]`` template, a ``"verbatim"`` example) cannot self-trip the
    # chip/quote scans. Masking is length-preserving, so offsets/line numbers are
    # unchanged.
    masked = _mask_code(markdown)
    scan_md = masked.translate(_SMART_DOUBLE_QUOTES)
    for m in _QUOTE_RE.finditer(scan_md):
        span = m.group(1)
        norm = normalize_for_match(span)
        if not norm:
            continue
        if not any(_verbatim_contained(norm, qb) for qb in quote_norms):
            violations.append(Violation(
                kind="quote",
                detail="quoted span is not verbatim in any referenced quote note",
                ref=span,
                line=scan_md.count("\n", 0, m.start()) + 1,
            ))

    # chips + claim headers are line-oriented. Scan the MASKED lines (code spans
    # blanked) so documented syntax is skipped; the ORIGINAL line is kept only to
    # recover a faithful header ref. Splitting both on ``"\n"`` keeps them aligned
    # 1:1 with the offset-derived quote line numbers above.
    raw_lines = markdown.split("\n")
    masked_lines = masked.split("\n")
    for lineno, (line, mline) in enumerate(zip(raw_lines, masked_lines), start=1):
        # Reserved ALL-CAPS markers (GAP, TODO, A/B notation) are structural, not
        # references — they are neither resolved nor treated as claim-header chips.
        chips = [
            m.group(1) for m in _CHIP_RE.finditer(mline)
            if not _is_sentinel(m.group(1))
        ]

        # chips resolve to a real note or paper
        for chip in chips:
            if chip not in resolvable and _chip_bare_id(chip) not in resolvable_bare:
                violations.append(Violation(
                    kind="chip",
                    detail=f"chip '{chip}' resolves to no note or paper in the material",
                    ref=chip,
                    line=lineno,
                ))

        # claim headers trace to a unit
        stripped = line.lstrip()
        if stripped.startswith("#") and chips:
            if not any(
                c in unit_ids or _chip_bare_id(c) in unit_bare_ids for c in chips
            ):
                header_text = stripped.lstrip("#").strip()
                violations.append(Violation(
                    kind="header",
                    detail="claim header does not trace to a claim/finding unit in the material",
                    ref=header_text,
                    line=lineno,
                ))

    violations.sort(key=lambda v: (v.line or 0))
    return violations

gather_outline_material

gather_outline_material(get_graph: GetGraph, *, project: str = '', graph: str = '', name: str = '', outline_id: str = DEFAULT_OUTLINE_ID, claim_ids: list[str] | None = None, as_of: 'str | int | None' = None, spine: str = '', spine_mode: str = '', ancestor_levels: int = 0, ancestor_uids: 'list[str] | None' = None, graphs_dir: 'Path | None' = None, namespace: Callable[[str], str] | None = None, localize: Callable[[str], str] | None = None) -> OutlineMaterial

Deterministic GATHER: the grounded raw material for an outline scope.

Builds the corpus + theme model exactly as :func:papers.assemble_papers does, ranks the claims, and for each (in salience order, or the exact claim_ids set/order when given) assembles a grounded :class:ClaimBundle via :func:claims.claim_anatomy.

Sectioning honors the author's SETTLED structure. When a review name is given and claim_ids is not pinned, the themed sections — their order, labels, membership, and per-section claim order — come from :func:_settled_structure (the Review board's own overlay reconciliation), so the scaffold's ## skeleton is exactly the themes the author settled on the board: renamed, reordered, with exclusions/placements applied. Otherwise (no review, or a pinned claim_ids set) bundles fall back to the graph-derived :func:syllabus.theme_model grouping (a claim's theme comes from its core/home/supporting papers), narrative-ordered within each section; claims in no theme go to unplaced.

Claim-sparse fallback. When claim_ids is not pinned, any theme that ends with no claim bundles is populated from its source notes (synthesis/finding/_cross material, salience-ranked) so a 0-claim corpus still yields themed sections with real material. The bundle shape is identical; only source ( "claim" vs "fallback" ) differs.

Spine section partition. When a spine ref (an organization id) is given and claim_ids is not pinned, the section axis switches to that spine's DIMENSIONS (the section-axis analogue of the matrix's column axis): sections are the spine's dimension nodes in matrix-column order, each section's content is the units whose uid is a MEMBER of that dimension (resolved via the same :func:tables.spine_readback_matrix membership the matrix cells read), and the apex framing rides on scope['spine']. A spine OVERRIDES both the settled structure and the theme model; an unresolvable ref degrades to the default. Theme partition stays the DEFAULT.

Nested composition tiers. When a nesting selection is requested (an explicit ancestor_uids set OR a non-zero ancestor_levels) and a spine subject resolves, the subject spine's single-spine dimension partition becomes the deepest LEAF content and the kept ANCESTOR tiers are stacked ABOVE it, drawn from the same :func:tables.detect_nested_spines composition tree the frontend displays. ancestor_uids is the AUTHORITATIVE selection: the subject's FULL root-first ancestor chain is pruned to exactly those uids in chain order (unchecked intermediate tiers drop, children reparent to the nearest kept ancestor — the frontend projectTiers pane). When ancestor_uids is None the historical int fallback applies (ancestor_levels == N → the nearest N ancestors, root-first; ancestor_levels == -1 → every ancestor to the composition root). Each ancestor tier is an :class:OutlineSection at an increasing markdown level (root shallowest, ##) with tier == "ancestor" and NO claim bundles — the DRAFT agent synthesizes 1–2 lecture-note framing bullets from its subtree. The subject's dimension sections keep their claims (tier == "leaf") but move to level = 2 + <kept ancestor count>. A flat request (ancestor_uids is None and ancestor_levels == 0, or an empty ancestor_uids) is a byte-identical no-op (no ancestor tiers, every section stays ## leaf). Nesting is ALSO skipped whenever claim_ids is pinned (an explicit set is its own authoritative ordering) — an explicit no-nest, never a nested-keyed flat.

Pure: no LLM, no network, no writes. JSON-serializable (:meth:OutlineMaterial.to_dict), carrying every referenced note id.

Source code in zettelkasten/outline/__init__.py

def gather_outline_material(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    name: str = "",
    outline_id: str = outlines_mod.DEFAULT_OUTLINE_ID,
    claim_ids: list[str] | None = None,
    as_of: "str | int | None" = None,
    spine: str = "",
    spine_mode: str = "",
    ancestor_levels: int = 0,
    ancestor_uids: "list[str] | None" = None,
    graphs_dir: "Path | None" = None,
    namespace: Callable[[str], str] | None = None,
    localize: Callable[[str], str] | None = None,
) -> OutlineMaterial:
    """Deterministic GATHER: the grounded raw material for an outline scope.

    Builds the corpus + theme model exactly as :func:`papers.assemble_papers`
    does, ranks the claims, and for each (in salience order, or the exact
    ``claim_ids`` set/order when given) assembles a grounded :class:`ClaimBundle`
    via :func:`claims.claim_anatomy`.

    **Sectioning honors the author's SETTLED structure.** When a review ``name``
    is given and ``claim_ids`` is not pinned, the themed sections — their order,
    labels, membership, and per-section claim order — come from
    :func:`_settled_structure` (the Review board's own overlay reconciliation), so
    the scaffold's ``##`` skeleton is exactly the themes the author settled on the
    board: renamed, reordered, with exclusions/placements applied. Otherwise (no
    review, or a pinned ``claim_ids`` set) bundles fall back to the graph-derived
    :func:`syllabus.theme_model` grouping (a claim's theme comes from its
    core/home/supporting papers), narrative-ordered within each section; claims in
    no theme go to ``unplaced``.

    **Claim-sparse fallback.** When ``claim_ids`` is not pinned, any theme that
    ends with no claim bundles is populated from its source notes
    (synthesis/finding/``_cross`` material, salience-ranked) so a 0-claim corpus
    still yields themed sections with real material. The bundle shape is
    identical; only ``source`` ( ``"claim"`` vs ``"fallback"`` ) differs.

    **Spine section partition.** When a ``spine`` ref (an organization id) is given
    and ``claim_ids`` is not pinned, the section axis switches to that spine's
    DIMENSIONS (the section-axis analogue of the matrix's column axis): sections
    are the spine's dimension nodes in matrix-column order, each section's content
    is the units whose uid is a MEMBER of that dimension (resolved via the same
    :func:`tables.spine_readback_matrix` membership the matrix cells read), and the
    apex framing rides on ``scope['spine']``. A spine OVERRIDES both the settled
    structure and the theme model; an unresolvable ref degrades to the default.
    Theme partition stays the DEFAULT.

    **Nested composition tiers.** When a nesting selection is requested (an
    explicit ``ancestor_uids`` set OR a non-zero ``ancestor_levels``) and a
    ``spine`` subject resolves, the subject spine's single-spine dimension
    partition becomes the deepest LEAF content and the kept ANCESTOR tiers are
    stacked ABOVE it, drawn from the same :func:`tables.detect_nested_spines`
    composition tree the frontend displays. ``ancestor_uids`` is the AUTHORITATIVE
    selection: the subject's FULL root-first ancestor chain is pruned to exactly
    those uids in chain order (unchecked intermediate tiers drop, children reparent
    to the nearest kept ancestor — the frontend ``projectTiers`` pane). When
    ``ancestor_uids is None`` the historical int fallback applies
    (``ancestor_levels == N`` → the nearest ``N`` ancestors, root-first;
    ``ancestor_levels == -1`` → every ancestor to the composition root). Each
    ancestor tier is an :class:`OutlineSection` at an increasing markdown ``level``
    (root shallowest, ``## ``) with ``tier == "ancestor"`` and NO claim bundles —
    the DRAFT agent synthesizes 1–2 lecture-note framing bullets from its subtree.
    The subject's dimension sections keep their claims (``tier == "leaf"``) but move
    to ``level = 2 + <kept ancestor count>``. A flat request (``ancestor_uids is
    None and ancestor_levels == 0``, or an empty ``ancestor_uids``) is a
    byte-identical no-op (no ancestor tiers, every section stays ``## `` leaf).
    Nesting is ALSO skipped whenever ``claim_ids`` is pinned (an explicit set is its
    own authoritative ordering) — an explicit no-nest, never a nested-keyed flat.

    Pure: no LLM, no network, no writes. JSON-serializable
    (:meth:`OutlineMaterial.to_dict`), carrying every referenced note id.
    """
    outline_id = outline_id or outlines_mod.DEFAULT_OUTLINE_ID
    base = graphs_dir or GRAPHS_DIR
    ns = namespace or (lambda s: s)
    loc = localize or (lambda s: s)
    as_of_int = _parse_as_of(as_of)

    # ── corpus + theme model (mirror assemble_papers / assemble_review) ────────
    if graph:
        works, scoped = _graph_scoped_corpus(get_graph, graph, graphs_dir=base, namespace=ns)
    else:
        works, scoped = build_corpus(get_graph, project=project, graphs_dir=base, namespace=ns)
    work_ids = {w.id for w in works}
    relation_edges = _derive_relation_edges(get_graph, scoped, loc)
    landscape_hubs = _derive_landscape_hubs(get_graph, scoped, loc, work_ids)
    citations = load_citations(graphs_dir=base)

    # ── claim index + enrichment (BASE importance, so the core-paper pick does
    #    not feed back into the salience it informs — same convention as papers) ─
    index = build_claim_index(
        get_graph, project=project, graph=graph, graphs_dir=base,
        namespace=ns, localize=loc,
    )
    base_importance = _paper_importance(works)
    centrality = claim_centrality(index)
    ranked = key_claims(index, importance_map=base_importance, centrality=centrality)
    enrich_works_with_claims(
        works, index,
        key_claim_keys=[kc.key for kc in ranked],
        citations=citations, importance_map=base_importance,
    )

    # ── deterministic scores + badges (claim hooks now folded into salience) ───
    inf = influence(works)
    sal = salience(works)
    imp = importance(works, influence_scores=inf, salience_scores=sal)
    anc = anchor_score(works)
    trends = corpus_trends(works, as_of=as_of_int)
    situations = situate(
        works, relation_edges=relation_edges, trends=trends,
        influence_scores=inf, salience_scores=sal, as_of=as_of_int,
    )
    tm = theme_model(
        works, relation_edges=relation_edges,
        landscape_hubs=landscape_hubs or None, anchor_scores=anc,
        # Match the Review board: a landscape hub (e.g. a promoted section) must not
        # collapse the scaffold to landscape-only — hybrid keeps the uncovered works
        # themed so their claims/concepts land in a section, not the unplaced bin.
        merge_remainder=True,
    )

    by_source_graph = {w.source_graph: w for w in works if w.source_graph}
    by_id = {w.id: w for w in works}
    assignment = tm.assignment

    # The claim_anatomy context (so each call reuses one prebuilt index/corpus).
    ctx = (index, works, citations, base_importance, centrality)

    # ── select claims and build their bundles, grouped by theme ────────────────
    selected, sal_by_key, sel_gaps = _select_claim_keys(index, ranked, claim_ids, loc)
    gaps: list[dict[str, Any]] = list(sel_gaps)

    bundles_by_theme: dict[str, list[ClaimBundle]] = {}
    keys_by_theme: dict[str, list[Key]] = {}
    unplaced: list[ClaimBundle] = []
    unplaced_keys: list[Key] = []
    # uid-indexed views of the same bundles, so the settled-structure assembly
    # below can pick each claim by its overlay uid (``<graph>::<id>``) in the
    # author's order without re-deriving the graph-model theme grouping.
    bundle_by_uid: dict[str, ClaimBundle] = {}
    key_by_uid: dict[str, Key] = {}
    for key in selected:
        rc = index.resolved.get(key)
        if rc is None:
            gaps.append(_gap("claim", f"claim '{key[1]}' not resolvable", ref=claim_uid(key)))
            continue
        core = core_paper_atom(
            rc, index, works=works, citations=citations, importance_map=base_importance
        )
        core_wid = None
        if core is not None and core.source_graph is not None:
            cw = by_source_graph.get(core.source_graph)
            core_wid = cw.id if cw is not None else None
        theme = _assign_claim_theme(rc, core_wid, assignment, by_source_graph)
        bundle = _claim_bundle(
            key=key, rc=rc, salience_v=sal_by_key.get(key, 0.0), theme=theme,
            get_graph=get_graph, ctx=ctx, index=index,
            by_source_graph=by_source_graph, by_id=by_id,
            situations=situations, sal=sal, imp=imp, anc=anc,
        )
        uid = claim_uid(key)
        bundle_by_uid[uid] = bundle
        key_by_uid[uid] = key
        if theme:
            bundles_by_theme.setdefault(theme, []).append(bundle)
            keys_by_theme.setdefault(theme, []).append(key)
        else:
            unplaced.append(bundle)
            unplaced_keys.append(key)

    # ── spine section partition (an ALTERNATIVE axis to themes), if referenced ─
    # A spine renders its dimensions as SECTIONS (the section-axis analogue of the
    # matrix's column axis). It is resolved only when not pinning claims — a pinned
    # ``claim_ids`` set is its own authoritative ordering — and it OVERRIDES both
    # the settled board structure and the theme model (the default partition stays
    # those; a spine is the opt-in alternative). An unresolvable ref degrades to
    # the default partition (``None``).
    #
    # Claim-pinned precedence (deliberate): when ``claim_ids`` is pinned, a ``spine``
    # ref is INTENTIONALLY ignored — the explicit set/order wins, so the spine does
    # NOT repartition a pinned outline. This is the lower-risk choice (repartitioning
    # a pinned set would override the caller's authoritative ordering and re-stamp a
    # signature divergent from the pin's). It is honest, not a silent loss: the
    # dashboard's OutlineView never pins ``claim_ids``, so its spine selector is never
    # inert; pinned callers are API-level and get pinned ordering by contract.
    spine_part: "_SpinePartition | None" = None
    if claim_ids is None and (spine or "").strip():
        spine_part = _resolve_spine_partition_ex(
            get_graph, spine, mode=spine_mode, project=project, graph=graph,
            base=base, loc=loc,
        )

    # ── the author-SETTLED section structure (overlay-reconciled), if any ──────
    # Only when claims are not explicitly pinned: a pinned ``claim_ids`` set is its
    # own authoritative ordering and must not be re-grouped by the board overlay.
    # A referenced spine takes precedence over the settled structure, so it is
    # resolved ONLY when no spine partition is active.
    settled: "SettledStructure | None" = None
    if claim_ids is None and spine_part is None:
        settled = _settled_structure(
            get_graph, project=project, graph=graph, name=name,
            base=base, ns=ns, loc=loc, outline_id=outline_id,
        )
    # The detach-carried apex framing (surfaced on ``scope`` below when a detached
    # outline reproduces the spine export's apex). Function-level so it is defined
    # regardless of which section branch runs; ``None`` for a normal outline.
    settled_apex: Any = None

    # ── claim-sparse fallback (only when claims are not explicitly pinned) ─────
    fallback_by_theme: dict[str, list[Key]] = {}
    if claim_ids is None:
        fallback_by_theme = _fallback_units_by_theme(
            index, assignment, by_source_graph, by_id, imp
        )

    # ── build sections, narrative-ordered ──────────────────────────────────────
    note_ids: set[str] = set()
    paper_ids: set[str] = set()
    sections: list[OutlineSection] = []

    if spine_part is not None:
        # SPINE partition: sections = the spine's dimensions (in the matrix's
        # column order), each section's content = the units whose uid is a MEMBER
        # of that dimension (the same spine-side membership the matrix cells read).
        # The claim-sparse fallback / concept tags are layered in per dimension
        # exactly as the theme branch does — the only change is HOW a unit is
        # routed to a section (membership, not theme assignment). Units in no
        # dimension fall to ``unplaced``; the apex framing rides on ``scope``.
        #
        # The fallback set is theme-keyed, so flatten it (de-duplicated, in theme
        # order) into one routable list and assign each entry by its uid's spine
        # membership.
        fallback_keys: list[Key] = []
        seen_fk: set[Key] = set()
        for keys in fallback_by_theme.values():
            for k in keys:
                if k not in seen_fk:
                    seen_fk.add(k)
                    fallback_keys.append(k)
        # The union of every DIMENSION's members — used to decide what is unplaced
        # (a unit attached to NO dimension) without re-querying the index. It must
        # be keyed on the dimensions ONLY, never the whole node-keyed membership:
        # ``spine_part.membership`` also carries row-hub keys, and sections are
        # built strictly over ``spine_part.dimensions``. Unioning the hubs in too
        # would mark a hub-only member (attached to a row hub but to no dimension)
        # as "placed" while it lands in no section — silently dropping it from the
        # outline. Restricting to dimension membership lets such a member fall to
        # the unplaced bucket below. (Dimensions are distinct keys, so no double-count.)
        all_member_uids: set[str] = set()
        for node_id, _label in spine_part.dimensions:
            all_member_uids |= spine_part.membership.get(node_id, set())
        placed_keys = set(unplaced_keys)
        for ks in keys_by_theme.values():
            placed_keys.update(ks)
        for node_id, label in spine_part.dimensions:
            member_uids = spine_part.membership.get(node_id, set())
            claim_bundles = []
            section_keys = []
            for uid in member_uids:
                bundle = bundle_by_uid.get(uid)
                if bundle is not None:
                    claim_bundles.append(bundle)
                    section_keys.append(key_by_uid[uid])
            section_concepts = []
            section_gaps = []
            if claim_ids is None:
                # Mirror the theme branch: a claim-BEARING dimension takes only the
                # cross-cutting framing synthesis; a claim-EMPTY one takes the full
                # fallback set it routes; concepts always condense to citation tags.
                framing_only = bool(claim_bundles)
                for fkey in fallback_keys:
                    if fkey in placed_keys:
                        continue
                    if claim_uid(fkey) not in member_uids:
                        continue
                    note = index.notes_by_key.get(fkey)
                    if note is None:
                        continue
                    if note.type == "concept":
                        section_concepts.append(_concept_tag(
                            key=fkey, note=note, by_source_graph=by_source_graph,
                        ))
                        placed_keys.add(fkey)
                        continue
                    if framing_only and note.type not in _FRAMING_FALLBACK_TYPES:
                        continue
                    claim_bundles.append(_fallback_bundle(
                        key=fkey, note=note, theme=node_id, index=index,
                        by_source_graph=by_source_graph, by_id=by_id,
                        situations=situations, sal=sal, imp=imp, anc=anc,
                    ))
                    section_keys.append(fkey)
                    placed_keys.add(fkey)
            if not claim_bundles and not section_concepts:
                section_gaps.append(_gap("section", "no claim or fallback material", theme=node_id))
            ordered = narrative_ordering(claim_bundles, _section_edges(index, section_keys))
            for b in ordered:
                note_ids.update(b.note_ids)
                paper_ids.update(b.paper_ids)
            for tag in section_concepts:
                note_ids.add(tag.uid)
                paper_ids.update(tag.paper_ids)
            sections.append(OutlineSection(
                theme_id=node_id, label=label, source="spine",
                claims=ordered, concepts=section_concepts, gaps=section_gaps,
            ))
        # Unplaced: claims attached to NO dimension (original salience order), then
        # untagged fallback notes that route to no dimension.
        unplaced = []
        unplaced_keys = []
        for uid, bundle in bundle_by_uid.items():
            if uid not in all_member_uids:
                unplaced.append(bundle)
                unplaced_keys.append(key_by_uid[uid])
        if claim_ids is None:
            for fkey in fallback_keys:
                if fkey in placed_keys:
                    continue
                if claim_uid(fkey) in all_member_uids:
                    # Routed to a dimension (possibly dropped there by the
                    # framing-only filter) — never re-homed into unplaced.
                    continue
                note = index.notes_by_key.get(fkey)
                if note is None:
                    continue
                unplaced.append(_fallback_bundle(
                    key=fkey, note=note, theme="", index=index,
                    by_source_graph=by_source_graph, by_id=by_id,
                    situations=situations, sal=sal, imp=imp, anc=anc,
                ))
                unplaced_keys.append(fkey)
                placed_keys.add(fkey)
        unplaced = narrative_ordering(unplaced, _section_edges(index, unplaced_keys))
        for b in unplaced:
            note_ids.update(b.note_ids)
            paper_ids.update(b.paper_ids)
    elif settled is not None:
        # AUTHOR-SETTLED structure: section order + labels + per-section claim
        # order come from the board overlay (see :func:`_settled_structure`). The
        # claim order is authoritative (no narrative re-ordering); only the
        # claim-sparse fallback / concept tags are layered in per theme, exactly
        # as the graph-model branch below does.
        settled_sections, settled_unplaced, section_framing, settled_apex = settled
        tm_by_id = {t.theme_id: t for t in tm.themes}
        # Every real claim is a placed unit, so the fallback never re-emits one.
        placed_keys: set[Key] = set(key_by_uid.values())
        for tid, label, claim_uids in settled_sections:
            # Proposed/overlay-only claims have no grounded bundle — skip them
            # (navigation + the H2 skeleton are theme-level, not claim-level).
            claim_bundles: list[ClaimBundle] = [
                bundle_by_uid[u] for u in claim_uids if u in bundle_by_uid
            ]
            section_concepts: list[ConceptTag] = []
            section_gaps: list[dict[str, Any]] = []
            framing_only = bool(claim_bundles)
            for fkey in fallback_by_theme.get(tid, []):
                if fkey in placed_keys:
                    continue
                note = index.notes_by_key.get(fkey)
                if note is None:
                    continue
                if note.type == "concept":
                    section_concepts.append(_concept_tag(
                        key=fkey, note=note, by_source_graph=by_source_graph,
                    ))
                    placed_keys.add(fkey)
                    continue
                if framing_only and note.type not in _FRAMING_FALLBACK_TYPES:
                    continue
                claim_bundles.append(_fallback_bundle(
                    key=fkey, note=note, theme=tid, index=index,
                    by_source_graph=by_source_graph, by_id=by_id,
                    situations=situations, sal=sal, imp=imp, anc=anc,
                ))
                placed_keys.add(fkey)
            # ── DETACH framing parity: re-layer the spine-DERIVED framing ──────
            # A "Detach from spine" persists the spine export's per-dimension
            # framing (concept tags + claim-sparse fallback units) into the
            # overlay's ``section_framing``, keyed by the settled section id.
            # Rebuild those units LIVE from their note keys so a detached outline
            # reproduces the spine export's framing, not just its claims. The
            # captured "concepts" / "fallback" split records only what each note
            # WAS at capture time; classification here is by the note's CURRENT
            # type, mirroring the live spine branch (~the ``fallback_keys`` pass)
            # EXACTLY: a note that still resolves as ``concept`` becomes a concept
            # tag; otherwise it is a fallback unit, SUPPRESSED on a claim-bearing
            # section (``framing_only``) unless it is a framing-synthesis
            # (``_FRAMING_FALLBACK_TYPES``) type. This keeps a captured note that
            # DRIFTED — e.g. retyped ``concept``→``method`` but still resolvable —
            # classified exactly as a live spine gather would (reclassified /
            # possibly dropped), never frozen at its capture-time kind. ``framing_only``
            # is the same section-level flag computed above (grounded claims present).
            # Presence-gated: a never-detached outline has no ``section_framing``
            # entry, so this is inert and the payload stays byte-identical. Units
            # already placed (a real claim, the ``fallback_by_theme`` pass above,
            # or a framing note that drifted into a claim) are skipped, and each
            # placed key is marked so the unplaced pass below never re-emits it.
            framing = section_framing.get(tid)
            if framing:
                for fuid in list(framing.get("concepts", [])) + list(framing.get("fallback", [])):
                    fkey = _uid_key(str(fuid))
                    if fkey in placed_keys:
                        continue
                    note = index.notes_by_key.get(fkey)
                    if note is None:
                        continue  # note no longer resolves — stay live with the graph
                    if note.type == "concept":
                        section_concepts.append(_concept_tag(
                            key=fkey, note=note, by_source_graph=by_source_graph,
                        ))
                        placed_keys.add(fkey)
                        continue
                    if framing_only and note.type not in _FRAMING_FALLBACK_TYPES:
                        continue
                    claim_bundles.append(_fallback_bundle(
                        key=fkey, note=note, theme=tid, index=index,
                        by_source_graph=by_source_graph, by_id=by_id,
                        situations=situations, sal=sal, imp=imp, anc=anc,
                    ))
                    placed_keys.add(fkey)
                # Slot the re-layered framing FALLBACK units within the section.
                # EDITABILITY: the board ``ordering`` overlay is AUTHORITATIVE for
                # CLAIM order — ``_settled_structure`` already applied it, so the
                # detach-frozen ``order`` must NEVER reorder the claims (an author who
                # reorders — or adds — claims after detach keeps that edit, and a claim
                # absent from the snapshot is not forced to the bottom). ``order`` (the
                # live spine export's within-section unit sequence: claims + framing
                # fallbacks, where e.g. a ``synthesis`` fallback can sit BEFORE a claim)
                # is used ONLY to position each framing fallback RELATIVE to the
                # board-ordered claims: a fallback is inserted immediately before the
                # claim it PRECEDED in the live export. A fallback with no following
                # claim in ``order`` (it trailed the last claim) — or absent from it (a
                # drifted unit, a legacy empty ``order``, or a ``fallback_by_theme`` unit
                # layered above) — trails after the claims, keeping its current relative
                # order. Inert for a never-detached outline (no ``framing``) and a stable
                # no-op when ``order`` is empty.
                order = framing.get("order")
                if order:
                    rank = {str(uid): i for i, uid in enumerate(order)}
                    claim_units = [b for b in claim_bundles if b.source == "claim"]
                    fallback_units = [b for b in claim_bundles if b.source != "claim"]
                    present_claims = {b.uid for b in claim_units}
                    before_claim: dict[str, list[ClaimBundle]] = {}
                    trailing: list[ClaimBundle] = []
                    # Anchor each fallback to the FIRST claim after it in ``order`` that
                    # still exists on the board; sorting by captured rank first keeps
                    # fallbacks sharing an anchor in live-export sequence and lands
                    # not-in-``order`` units (rank == len) in ``trailing``.
                    for f in sorted(fallback_units, key=lambda b: rank.get(b.uid, len(rank))):
                        pos = rank.get(f.uid)
                        anchor: str | None = None
                        if pos is not None:
                            for uid in order[pos + 1:]:
                                if str(uid) in present_claims:
                                    anchor = str(uid)
                                    break
                        if anchor is None:
                            trailing.append(f)
                        else:
                            before_claim.setdefault(anchor, []).append(f)
                    rebuilt: list[ClaimBundle] = []
                    for c in claim_units:
                        rebuilt.extend(before_claim.get(c.uid, []))
                        rebuilt.append(c)
                    rebuilt.extend(trailing)
                    claim_bundles = rebuilt
            if not claim_bundles and not section_concepts:
                section_gaps.append(_gap("section", "no claim or fallback material", theme=tid))
            for b in claim_bundles:
                note_ids.update(b.note_ids)
                paper_ids.update(b.paper_ids)
            for tag in section_concepts:
                note_ids.add(tag.uid)
                paper_ids.update(tag.paper_ids)
            # A proposed (overlay-only) theme has no theme-model row → "settled".
            src = tm_by_id[tid].source if tid in tm_by_id else "settled"
            sections.append(OutlineSection(
                theme_id=tid, label=label, source=src,
                claims=claim_bundles, concepts=section_concepts, gaps=section_gaps,
            ))
        # Unplaced: settled order, then untagged fallback notes after.
        unplaced = [bundle_by_uid[u] for u in settled_unplaced if u in bundle_by_uid]
        for fkey in fallback_by_theme.get("", []):
            if fkey in placed_keys:
                continue
            note = index.notes_by_key.get(fkey)
            if note is None:
                continue
            unplaced.append(_fallback_bundle(
                key=fkey, note=note, theme="", index=index,
                by_source_graph=by_source_graph, by_id=by_id,
                situations=situations, sal=sal, imp=imp, anc=anc,
            ))
            placed_keys.add(fkey)
        for b in unplaced:
            note_ids.update(b.note_ids)
            paper_ids.update(b.paper_ids)
    else:
        # GRAPH-MODEL structure: theme-model order, narrative-ordered claims.
        # Every key already materialized as a bundle, so the untagged fallback
        # below never double-emits a note that is already a placed unit.
        placed_keys = set(unplaced_keys)
        for ks in keys_by_theme.values():
            placed_keys.update(ks)
        for t in tm.themes:
            # Copy: the fallback pass below may extend this with framing units, and
            # we must not mutate the bundles_by_theme entry it is sourced from.
            claim_bundles = list(bundles_by_theme.get(t.theme_id, []))
            section_keys = list(keys_by_theme.get(t.theme_id, []))
            section_concepts = []
            section_gaps = []

            if claim_ids is None:
                # A claim-EMPTY section takes the full fallback set (it has nothing
                # else to show); a claim-BEARING section takes only the framing
                # synthesis so the cross-cutting frame rides alongside the argument
                # without the rest of the source material cluttering it. CONCEPTS
                # are never units: either way they become condensed citation TAGS.
                framing_only = bool(claim_bundles)
                for fkey in fallback_by_theme.get(t.theme_id, []):
                    if fkey in placed_keys:
                        continue
                    note = index.notes_by_key.get(fkey)
                    if note is None:
                        continue
                    if note.type == "concept":
                        # A concept frames the section (the landscape concepts that
                        # ARE section headers were already excluded upstream) —
                        # surface it as a compact tag, not a card.
                        section_concepts.append(_concept_tag(
                            key=fkey, note=note, by_source_graph=by_source_graph,
                        ))
                        placed_keys.add(fkey)
                        continue
                    if framing_only and note.type not in _FRAMING_FALLBACK_TYPES:
                        continue
                    claim_bundles.append(_fallback_bundle(
                        key=fkey, note=note, theme=t.theme_id, index=index,
                        by_source_graph=by_source_graph, by_id=by_id,
                        situations=situations, sal=sal, imp=imp, anc=anc,
                    ))
                    section_keys.append(fkey)
                    placed_keys.add(fkey)

            if not claim_bundles and not section_concepts:
                section_gaps.append(_gap("section", "no claim or fallback material", theme=t.theme_id))

            ordered = narrative_ordering(claim_bundles, _section_edges(index, section_keys))
            for b in ordered:
                note_ids.update(b.note_ids)
                paper_ids.update(b.paper_ids)
            # Concept chips are resolvable refs the scaffold cites; carry them (and
            # their home papers) into the material-level id unions for the verifier.
            for tag in section_concepts:
                note_ids.add(tag.uid)
                paper_ids.update(tag.paper_ids)
            sections.append(OutlineSection(
                theme_id=t.theme_id, label=t.label, source=t.source,
                claims=ordered, concepts=section_concepts, gaps=section_gaps,
            ))

        # Untagged claims keep their own narrative order; untagged fallback notes
        # (``_cross``/synthesis material with no theme) are appended after them.
        # Decoupled from ``unplaced`` emptiness: a single untagged claim already in
        # ``unplaced`` must NOT suppress legitimate fallback units — but a note that
        # is already a placed unit (``placed_keys``) is never emitted twice.
        if claim_ids is None:
            for fkey in fallback_by_theme.get("", []):
                if fkey in placed_keys:
                    continue
                note = index.notes_by_key.get(fkey)
                if note is None:
                    continue
                unplaced.append(_fallback_bundle(
                    key=fkey, note=note, theme="", index=index,
                    by_source_graph=by_source_graph, by_id=by_id,
                    situations=situations, sal=sal, imp=imp, anc=anc,
                ))
                unplaced_keys.append(fkey)
                placed_keys.add(fkey)
        unplaced = narrative_ordering(unplaced, _section_edges(index, unplaced_keys))
        for b in unplaced:
            note_ids.update(b.note_ids)
            paper_ids.update(b.paper_ids)

    # ── nested composition tiers: stack N ANCESTOR framing tiers above the subject
    # spine's leaf partition (only when a spine subject resolved and nesting is
    # requested). The subject's dimension sections keep their claims but move down
    # to ``level = 2 + <ancestor count>``; each ancestor becomes a claim-less
    # framing section at an increasing markdown level (root shallowest). This is a
    # pure re-leveling + prepend — the leaf partition (membership-routed claims) is
    # unchanged, so ``ancestor_levels == 0`` remains byte-identical. ──────────────
    ancestor_tiers: list[dict[str, Any]] = []
    # The explicit NESTED-composition degradation signal (W3): populated only when
    # a nesting selection is actually requested (and a spine subject resolves), so
    # a flat outline leaves it ``None`` and the payload stays byte-identical.
    nesting_degradation: "dict[str, Any] | None" = None
    nesting_requested = ancestor_uids is not None or ancestor_levels != 0
    if spine_part is not None and nesting_requested and claim_ids is None:
        ancestor_tiers, nesting_degradation = _resolve_nested_ancestors(
            get_graph, subject_graph=spine_part.graph_name, project=project, graph=graph,
            base=base, loc=loc, ancestor_levels=ancestor_levels, ancestor_uids=ancestor_uids,
            subject_org_id=spine_part.org_id,
        )
    if ancestor_tiers:
        depth = len(ancestor_tiers)
        for sec in sections:
            sec.level = 2 + depth
        ancestor_sections = [
            OutlineSection(
                theme_id=tier["uid"],
                label=tier["label"],
                source="spine",
                claims=[],
                concepts=[],
                gaps=[],
                level=2 + i,
                tier="ancestor",
            )
            for i, tier in enumerate(ancestor_tiers)
        ]
        sections = ancestor_sections + sections

    if not sections:
        gaps.append(_gap("outline", "no themes in scope"))

    scope_name = graph or project or "all"
    scope: dict[str, Any] = {"type": "graph" if graph else "project", "name": scope_name}
    # A spine partition: record the spine identity + the apex framing on the scope
    # so the DRAFT prompt and the frontend can frame the whole document by the
    # spine's synthesis root, and ``theme_source`` names the spine (the "themes
    # from …" cue) instead of the graph theme model.
    theme_source = tm.source
    if spine_part is not None:
        scope["spine"] = {
            "org_id": spine_part.org_id,
            "graph": spine_part.graph_name,
            "title": spine_part.title,
            "apex": spine_part.apex,
        }
        # Nested composition: surface the stacked ancestor tiers (root-first) so the
        # DRAFT agent and the frontend can frame the document by its full tier path.
        if ancestor_tiers:
            scope["spine"]["ancestor_tiers"] = [
                {"label": t["label"], "graph": t["graph"], "level": 2 + i}
                for i, t in enumerate(ancestor_tiers)
            ]
        theme_source = f"spine: {spine_part.title}"
    elif settled_apex is not None:
        # A DETACHED-from-spine outline is no longer a spine partition (its section
        # axis is the settled board structure), but it carries the spine export's
        # apex synthesis root in its overlay. Re-surface JUST the apex on
        # ``scope['spine']`` so the DRAFT prompt / frontend can frame the detached
        # document by the apex, reproducing the spine export's framing. This is a
        # DELIBERATELY PARTIAL shape: unlike the live spine branch above it omits
        # the spine-identity fields (``org_id`` / ``graph`` / ``title`` /
        # ``ancestor_tiers``) because the outline is no longer a spine — those
        # would falsely advertise a live spine partition. Every consumer reads the
        # apex defensively (``(scope.get("spine") or {}).get("apex")``), and
        # spine-mode / read-only is gated on the outline RECORD's ``spine`` field
        # (cleared by the detach), NOT on this scope — so the apex-only shape
        # cannot flip any consumer into spine-organized / read-only mode.
        # Presence-gated on ``settled_apex`` — a normal settled outline leaves
        # ``scope`` untouched (byte-identical). ``theme_source`` stays the graph
        # theme model: the outline IS a theme outline now, not a spine.
        # LIVENESS: the snapshot stores the apex synthesis note's ``uid`` / ``graph``
        # / ``id``. Re-resolve it against the LIVE note and REFRESH the apex from the
        # current note — self-healing, exactly like the framing units above which skip
        # a uid whose note no longer resolves (``if note is None: continue``) AND
        # re-check the note's CURRENT kind. If the apex note was deleted after detach
        # so it no longer resolves, OR was retyped to drop its apex tags (so a live
        # spine gather via ``_resolve_spine_partition`` would no longer derive it as
        # the apex), OMIT the apex rather than surface the FROZEN one: a detached
        # export must not advertise a synthesis root that no longer exists or no
        # longer qualifies, and must stay consistent with the self-healing framing
        # units in the same payload (which re-check ``note.type``). The apex synthesis root
        # lives in the SPINE graph (``graph`` on the snapshot) — a DIFFERENT graph
        # from this outline's review scope — so it is resolved against ITS OWN graph
        # (mirroring :func:`_resolve_spine_partition`), NOT the review-scoped
        # ``index`` (which never carries the spine graph's notes).
        apex_graph = str(settled_apex.get("graph") or "") if isinstance(settled_apex, dict) else ""
        apex_id = str(settled_apex.get("id") or "") if isinstance(settled_apex, dict) else ""
        if isinstance(settled_apex, dict) and not (apex_graph and apex_id):
            # Legacy snapshot without split fields — recover them from the uid.
            apex_graph, apex_id = _uid_key(str(settled_apex.get("uid") or ""))
        if apex_graph and apex_id:
            # Reuse the SAME apex predicate the live spine branch uses
            # (:func:`_resolve_spine_partition` → ``APEX_TAGS <= note.tags``) so a
            # note RETYPED to drop its apex tags is dropped here too — matching how
            # the concept/fallback framing units re-check ``note.type``.
            from zettelkasten.spine import APEX_TAGS

            try:
                apex_note = get_graph(loc(apex_graph)).notes.get(apex_id)
            except Exception:  # noqa: BLE001 — an unreadable spine graph → treat as gone
                apex_note = None
            if apex_note is not None and set(APEX_TAGS) <= set(apex_note.tags or []):
                refreshed = dict(settled_apex)
                refreshed["label"] = apex_note.title or apex_note.id
                refreshed["body"] = (apex_note.body or "").strip()
                scope["spine"] = {"apex": refreshed}
            # else: the apex note no longer resolves, OR it no longer carries the
            # apex tags (retyped out of the apex role) → OMIT it (no
            # ``scope['spine']``), exactly as a live spine gather would.
        else:
            # Cannot determine the apex's graph (a legacy/unqualified snapshot) — we
            # can't prove it dead, so surface it as-is rather than drop it.
            scope["spine"] = {"apex": settled_apex}
    return OutlineMaterial(
        scope=scope,
        theme_source=theme_source,
        sections=sections,
        unplaced=unplaced,
        gaps=gaps,
        note_ids=sorted(note_ids),
        paper_ids=sorted(paper_ids),
        nesting_degradation=nesting_degradation,
    )

draft

draft(material: OutlineMaterial, *, draft_fn: 'Callable[[str, str], str] | None' = None, skill_path: 'str | Path | None' = None) -> str

DRAFT the scaffold markdown from gathered material via an INJECTABLE agent.

Loads the assemble-grounded-outline skill as the system context, builds the agent prompt (the drafting task instruction + the serialized GATHER material), and calls draft_fn(system, prompt) -> str. draft_fn defaults to :func:_default_draft_fn (the in-process dashboard agent, imported lazily) so tests can stub it deterministically. The engine GATHERS; the agent DRAFTS — this never writes prose itself.

Source code in zettelkasten/outline/__init__.py

def draft(
    material: OutlineMaterial,
    *,
    draft_fn: "Callable[[str, str], str] | None" = None,
    skill_path: "str | Path | None" = None,
) -> str:
    """DRAFT the scaffold markdown from gathered material via an INJECTABLE agent.

    Loads the ``assemble-grounded-outline`` skill as the system context, builds
    the agent prompt (the drafting task instruction + the serialized GATHER
    material), and calls ``draft_fn(system, prompt) -> str``. ``draft_fn``
    defaults to :func:`_default_draft_fn` (the in-process dashboard agent, imported
    lazily) so tests can stub it deterministically. The engine GATHERS; the agent
    DRAFTS — this never writes prose itself.
    """
    contract, _version = _load_skill_contract(skill_path)
    prompt = _build_draft_prompt(material)
    fn = draft_fn or _default_draft_fn
    return fn(contract, prompt)

draft_section_expansion

draft_section_expansion(material: OutlineMaterial, section_md: str, *, draft_fn: 'Callable[[str, str], str] | None' = None, skill_path: 'str | Path | None' = None) -> str

DEEPEN one section's markdown via the SAME injectable agent as DRAFT.

Loads the restraint contract as system context and builds the DEEPEN prompt (the deepen instruction + the section being deepened + the serialized GATHER material), then calls draft_fn(system, prompt) -> str. Mirrors :func:draft exactly so the two task framings share one agent config and one stubbing seam in tests; the engine never writes prose itself.

Source code in zettelkasten/outline/__init__.py

def draft_section_expansion(
    material: OutlineMaterial,
    section_md: str,
    *,
    draft_fn: "Callable[[str, str], str] | None" = None,
    skill_path: "str | Path | None" = None,
) -> str:
    """DEEPEN one section's markdown via the SAME injectable agent as DRAFT.

    Loads the restraint contract as system context and builds the DEEPEN prompt
    (the deepen instruction + the section being deepened + the serialized GATHER
    material), then calls ``draft_fn(system, prompt) -> str``. Mirrors
    :func:`draft` exactly so the two task framings share one agent
    config and one stubbing seam in tests; the engine never writes prose itself.
    """
    contract, _version = _load_skill_contract(skill_path)
    prompt = _build_expand_prompt(material, section_md)
    fn = draft_fn or _default_draft_fn
    return fn(contract, prompt)

compute_generation_signature

compute_generation_signature(get_graph: 'GetGraph | None' = None, *, project: str = '', graph: str = '', name: str = '', claim_ids: 'list[str] | None' = None, as_of: 'str | int | None' = None, spine: str = '', spine_mode: str = '', ancestor_levels: int = 0, ancestor_uids: 'list[str] | None' = None, graphs_dir: 'Path | None' = None, skill_path: 'str | Path | None' = None, localize: 'Callable[[str], str] | None' = None) -> str

hash(graph_sig + meta_sig + citations_sig + overlay_sig + skill_version).

• graph_sig — the :func:_scope_fingerprint over the scope (plus the scope descriptor: project/graph/claim_ids/as_of, which also define what is drafted), so any note edit or scope change invalidates the cache.
• meta_sig — the :func:_scope_meta_fingerprint over the in-scope sources' _meta.yaml files, so a year/author/title/coverage edit (which changes the drafted material — citation labels, badges, theme ordering, bundle years — but never touches a *.md note) busts the cache.
• citations_sig — the :func:_citations_fingerprint over the global _citations store threaded into the works, so any citation edit busts it.
• project_sig — the :func:_project_membership_sig over a project scope's _projects/<name>.yaml source list (GATHER scopes the corpus to exactly this set), so reassigning a project's membership busts the cache even though the project file is observed by no other probe.
• overlay_sig — the _reviews/<name>.yaml editorial overlay (ordering, scaffolding, tiering, …); an editorial change forces a redraft.
• skill_version — the content hash of the DRAFT skill contract.
• spine_sig — folded in ONLY when a spine section axis is referenced (:func:_spine_signature): the ref + the spine graph's note fingerprint, so selecting/editing a spine busts the cache exactly as a scope/column change does. Absent for every spine-less outline and for the matrix's call here, so their signatures (and caches) are byte-identical to before.
• ancestor_uids — the RESOLVED, ORDERED (root-first) ancestor uid list the nested build will actually stack (see :func:_resolved_ancestor_uids), folded in ONLY when non-empty. Because it is the resolved uid list — NOT the bare int ancestor_levels — two DISTINCT selections of the same SIZE (e.g. a gapped [root, leaf-parent] vs a contiguous [root, mid]) cache distinctly instead of colliding, and the fold condition matches the GATHER nesting guard exactly (a spine subject resolved, claim_ids not pinned, a non-empty kept chain). An empty resolved list folds NOTHING, so every flat outline (ancestor_levels == 0 and no ancestor_uids), a pinned claim_ids build, a spine-less scope, and the matrix's call here all hash byte-identically to before. Resolving requires get_graph; when it is absent (e.g. the matrix's :func:_table_signature call) nothing is folded — the flat key.

Designed so the Wave-5 MCP tool / endpoint is a thin delegate (it passes the same scope through), mirroring build_syllabus.

Source code in zettelkasten/outline/__init__.py

def compute_generation_signature(
    get_graph: "GetGraph | None" = None,
    *,
    project: str = "",
    graph: str = "",
    name: str = "",
    claim_ids: "list[str] | None" = None,
    as_of: "str | int | None" = None,
    spine: str = "",
    spine_mode: str = "",
    ancestor_levels: int = 0,
    ancestor_uids: "list[str] | None" = None,
    graphs_dir: "Path | None" = None,
    skill_path: "str | Path | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> str:
    """``hash(graph_sig + meta_sig + citations_sig + overlay_sig + skill_version)``.

    * ``graph_sig`` — the :func:`_scope_fingerprint` over the scope (plus the
      scope descriptor: project/graph/claim_ids/as_of, which also define what is
      drafted), so any note edit or scope change invalidates the cache.
    * ``meta_sig`` — the :func:`_scope_meta_fingerprint` over the in-scope
      sources' ``_meta.yaml`` files, so a year/author/title/coverage edit (which
      changes the drafted material — citation labels, badges, theme ordering,
      bundle years — but never touches a ``*.md`` note) busts the cache.
    * ``citations_sig`` — the :func:`_citations_fingerprint` over the global
      ``_citations`` store threaded into the works, so any citation edit busts it.
    * ``project_sig`` — the :func:`_project_membership_sig` over a project scope's
      ``_projects/<name>.yaml`` source list (GATHER scopes the corpus to exactly
      this set), so reassigning a project's membership busts the cache even though
      the project file is observed by no other probe.
    * ``overlay_sig`` — the ``_reviews/<name>.yaml`` editorial overlay (ordering,
      scaffolding, tiering, …); an editorial change forces a redraft.
    * ``skill_version`` — the content hash of the DRAFT skill contract.
    * ``spine_sig`` — folded in ONLY when a ``spine`` section axis is referenced
      (:func:`_spine_signature`): the ref + the spine graph's note fingerprint, so
      selecting/editing a spine busts the cache exactly as a scope/column change
      does. Absent for every spine-less outline and for the matrix's call here, so
      their signatures (and caches) are byte-identical to before.
    * ``ancestor_uids`` — the RESOLVED, ORDERED (root-first) ancestor uid list the
      nested build will actually stack (see :func:`_resolved_ancestor_uids`), folded
      in ONLY when non-empty. Because it is the resolved uid list — NOT the bare int
      ``ancestor_levels`` — two DISTINCT selections of the same SIZE (e.g. a gapped
      ``[root, leaf-parent]`` vs a contiguous ``[root, mid]``) cache distinctly
      instead of colliding, and the fold condition matches the GATHER nesting guard
      exactly (a spine subject resolved, ``claim_ids`` not pinned, a non-empty kept
      chain). An empty resolved list folds NOTHING, so every flat outline
      (``ancestor_levels == 0`` and no ``ancestor_uids``), a pinned ``claim_ids``
      build, a spine-less scope, and the matrix's call here all hash byte-identically
      to before. Resolving requires ``get_graph``; when it is absent (e.g. the
      matrix's :func:`_table_signature` call) nothing is folded — the flat key.

    Designed so the Wave-5 MCP tool / endpoint is a thin delegate (it passes the
    same scope through), mirroring ``build_syllabus``.
    """
    base = Path(graphs_dir) if graphs_dir is not None else GRAPHS_DIR
    graph_sig = _scope_fingerprint(base, project=project, graph=graph)
    meta_sig = _scope_meta_fingerprint(base, project=project, graph=graph)
    citations_sig = _citations_fingerprint(base)
    project_sig = _project_membership_sig(base, project)
    scope_desc = {
        "project": project or "",
        "graph": graph or "",
        "claim_ids": list(claim_ids) if claim_ids else None,
        "as_of": _parse_as_of(as_of),
    }
    # Normalize the overlay to its canonical empty shape so a not-yet-created
    # review (no manifest → ``{}``) and a freshly seeded empty overlay hash
    # IDENTICALLY — otherwise the first build (manifest absent) and the next
    # (manifest now present with an empty overlay) would disagree and never cache.
    raw_overlay: Any = None
    if name:
        manifest = load_review(name, graphs_dir=base)
        if manifest:
            raw_overlay = manifest.get("overlay")
    overlay = _normalize_overlay(raw_overlay)
    _content, skill_version = _load_skill_contract(skill_path)

    payload_obj: dict[str, Any] = {
        "scope": scope_desc,
        "graph_sig": graph_sig,
        "meta_sig": meta_sig,
        "citations_sig": citations_sig,
        "project_sig": project_sig,
        "overlay_sig": overlay,
        "skill_version": skill_version,
    }
    # The spine section axis is folded in ONLY when referenced, so a non-spine
    # outline (and the matrix's spine-less call) hashes byte-identically to today
    # — and a spine change (ref or graph edit) busts the cache like a scope change.
    spine_sig = _spine_signature(
        base, project=project, graph=graph, spine=spine, spine_mode=spine_mode
    )
    if spine_sig is not None:
        payload_obj["spine_sig"] = spine_sig
    # The RESOLVED ordered ancestor uid list is folded in ONLY when non-empty, so a
    # flat outline (and the matrix's spine-less, un-nested call) hashes
    # byte-identically to today, while nested vs flat — and DISTINCT equal-size
    # selections — each cache separately. Resolved via the SAME guard GATHER nests
    # on, so the key matches the tiers that will actually be drafted.
    resolved_ancestors = _resolved_ancestor_uids(
        get_graph, project=project, graph=graph, base=base,
        loc=(localize or (lambda s: s)), spine=spine, claim_ids=claim_ids,
        ancestor_levels=ancestor_levels, ancestor_uids=ancestor_uids,
    )
    if resolved_ancestors:
        payload_obj["ancestor_uids"] = resolved_ancestors

    payload = json.dumps(payload_obj, sort_keys=True, default=str)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

compute_outline_signature

compute_outline_signature(get_graph: 'GetGraph | None' = None, *, project: str = '', graph: str = '', name: str = '', outline_id: str = DEFAULT_OUTLINE_ID, claim_ids: 'list[str] | None' = None, as_of: 'str | int | None' = None, spine: str = '', spine_mode: str = '', ancestor_levels: int = 0, ancestor_uids: 'list[str] | None' = None, graphs_dir: 'Path | None' = None, skill_path: 'str | Path | None' = None, localize: 'Callable[[str], str] | None' = None) -> str

The cache key for ONE outline's scaffold.

Folds the shared corpus/overlay :func:compute_generation_signature together with this outline's identity, its research QUESTION, and (for an additional outline whose structure is NOT in the manifest overlay) its own merged overlay. So:

• editing the outline's question forces a redraft of THAT outline;
• the default outline (outline_id == DEFAULT_OUTLINE_ID, question == the review's own question) keeps using the manifest-overlay-backed corpus signature — its cache key only gains the question term;
• the matrix's _table_signature (which calls :func:compute_generation_signature directly) is UNAFFECTED — the outline question never leaks into it.

Source code in zettelkasten/outline/__init__.py

def compute_outline_signature(
    get_graph: "GetGraph | None" = None,
    *,
    project: str = "",
    graph: str = "",
    name: str = "",
    outline_id: str = outlines_mod.DEFAULT_OUTLINE_ID,
    claim_ids: "list[str] | None" = None,
    as_of: "str | int | None" = None,
    spine: str = "",
    spine_mode: str = "",
    ancestor_levels: int = 0,
    ancestor_uids: "list[str] | None" = None,
    graphs_dir: "Path | None" = None,
    skill_path: "str | Path | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> str:
    """The cache key for ONE outline's scaffold.

    Folds the shared corpus/overlay :func:`compute_generation_signature` together
    with this outline's identity, its research QUESTION, and (for an additional
    outline whose structure is NOT in the manifest overlay) its own merged
    overlay. So:

    * editing the outline's question forces a redraft of THAT outline;
    * the default outline (``outline_id == DEFAULT_OUTLINE_ID``, question == the
      review's own question) keeps using the manifest-overlay-backed corpus
      signature — its cache key only gains the question term;
    * the matrix's ``_table_signature`` (which calls :func:`compute_generation_signature`
      directly) is UNAFFECTED — the outline question never leaks into it.
    """
    outline_id = outline_id or outlines_mod.DEFAULT_OUTLINE_ID
    base = Path(graphs_dir) if graphs_dir is not None else GRAPHS_DIR
    corpus_sig = compute_generation_signature(
        get_graph, project=project, graph=graph, name=name, claim_ids=claim_ids,
        as_of=as_of, spine=spine, spine_mode=spine_mode, ancestor_levels=ancestor_levels,
        ancestor_uids=ancestor_uids, graphs_dir=base, skill_path=skill_path,
        localize=localize,
    )
    question = outlines_mod.get_question(name, outline_id, graphs_dir=base) if name else ""
    payload: dict[str, Any] = {
        "corpus": corpus_sig,
        "outline_id": outline_id,
        "question": question,
    }
    if outline_id != outlines_mod.DEFAULT_OUTLINE_ID:
        # An additional outline's STRUCTURE lives in the sidecar, not the manifest
        # overlay folded into ``corpus_sig`` — bind it in explicitly.
        payload["overlay"] = outlines_mod.merged_overlay(name, outline_id, graphs_dir=base)
    blob = json.dumps(payload, sort_keys=True, default=str)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()

compute_content_signature

compute_content_signature(markdown: str) -> str

Stable hash of the PERSISTED draft markdown, for the edit-staleness guard.

Distinct from the generation/material :func:compute_outline_signature: that signature fingerprints the SCOPE inputs and is INVARIANT across a force/deepen that rewrites the derived prose for the SAME material — so it cannot tell that a Fix/Remove's line now points at different content. This hashes the actual persisted .md bytes, so ANY rewrite of the draft (a regenerate, a deepen, a prior fix) changes it and a stale client's edit is refused (HTTP 409) instead of editing/deleting the wrong line. It NEVER drives caching or redraft — that stays the generation signature.

Source code in zettelkasten/outline/__init__.py

def compute_content_signature(markdown: str) -> str:
    """Stable hash of the PERSISTED draft markdown, for the edit-staleness guard.

    Distinct from the generation/material :func:`compute_outline_signature`: that
    signature fingerprints the SCOPE inputs and is INVARIANT across a
    force/deepen that rewrites the derived prose for the SAME material — so it
    cannot tell that a Fix/Remove's ``line`` now points at different content. This
    hashes the actual persisted ``.md`` bytes, so ANY rewrite of the draft (a
    regenerate, a deepen, a prior fix) changes it and a stale client's edit is
    refused (HTTP 409) instead of editing/deleting the wrong line. It NEVER drives
    caching or redraft — that stays the generation ``signature``.
    """
    return hashlib.sha256((markdown or "").encode("utf-8")).hexdigest()

build_outline

build_outline(get_graph: GetGraph, *, project: str = '', graph: str = '', name: str = '', outline_id: str = DEFAULT_OUTLINE_ID, claim_ids: 'list[str] | None' = None, as_of: 'str | int | None' = None, spine: str = '', spine_mode: str = '', ancestor_levels: int = 0, ancestor_uids: 'list[str] | None' = None, force: bool = False, peek: bool = False, draft_fn: 'Callable[[str, str], str] | None' = None, graphs_dir: 'Path | None' = None, skill_path: 'str | Path | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None) -> dict[str, Any]

Top-level orchestrator: GATHER → cache check → DRAFT → integrity → write.

Computes the :func:compute_generation_signature; if it matches the stored signature in _reviews/<name>.yaml and the .md exists (and force is false), returns the stored markdown with NO agent call (cached=True). Otherwise it GATHERs the material, DRAFTs the scaffold via :func:draft (the injectable draft_fn), runs the deterministic :func:verify_draft integrity pass, annotates any violations into the artifact, and persists _reviews/<name>.md as a REGENERABLE DERIVED ARTIFACT (overwrite-on-regenerate) through the M3a substrate (:func:review_write_lock → :func:atomic_write_text → :func:_schedule_zettel_commit), also storing the signature under the manifest's scaffold key through the same write path.

When peek is set the call is COST-FREE and read-only: it computes the current signature, reads the stored scaffold signature + the cached .md (if either exists), and reports staleness WITHOUT ever GATHERing, DRAFTing, writing, committing, or seeding an empty manifest. The dashboard fires this on plain navigation so opening a never-built/stale review never DRAFTs — only an explicit force regenerate does.

ancestor_uids/ancestor_levels thread the NESTED composition selection through GATHER and the cache signature (see :func:gather_outline_material / :func:compute_generation_signature): a flat request (no ancestor_uids and ancestor_levels == 0) is the flat single-spine/theme partition, byte-identical to before; an explicit ancestor_uids prune (authoritative) or a non-zero ancestor_levels (fallback) stacks the kept ancestor framing tiers above the subject's leaf partition and caches separately per DISTINCT selection.

Returns {markdown, artifact_path, signature, stale, violations, cached, exists} (a peek also carries stored_signature). The Wave-5 MCP tool + POST endpoint are thin delegates over this.

Source code in zettelkasten/outline/__init__.py

def build_outline(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    name: str = "",
    outline_id: str = outlines_mod.DEFAULT_OUTLINE_ID,
    claim_ids: "list[str] | None" = None,
    as_of: "str | int | None" = None,
    spine: str = "",
    spine_mode: str = "",
    ancestor_levels: int = 0,
    ancestor_uids: "list[str] | None" = None,
    force: bool = False,
    peek: bool = False,
    draft_fn: "Callable[[str, str], str] | None" = None,
    graphs_dir: "Path | None" = None,
    skill_path: "str | Path | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> dict[str, Any]:
    """Top-level orchestrator: GATHER → cache check → DRAFT → integrity → write.

    Computes the :func:`compute_generation_signature`; if it matches the stored
    signature in ``_reviews/<name>.yaml`` and the ``.md`` exists (and ``force``
    is false), returns the stored markdown with NO agent call (``cached=True``).
    Otherwise it GATHERs the material, DRAFTs the scaffold via
    :func:`draft` (the injectable ``draft_fn``), runs the deterministic
    :func:`verify_draft` integrity pass, annotates any violations into the
    artifact, and persists ``_reviews/<name>.md`` as a REGENERABLE DERIVED
    ARTIFACT (overwrite-on-regenerate) through the M3a substrate
    (:func:`review_write_lock` → :func:`atomic_write_text` →
    :func:`_schedule_zettel_commit`), also storing the signature under the
    manifest's ``scaffold`` key through the same write path.

    When ``peek`` is set the call is COST-FREE and read-only: it computes the
    current signature, reads the stored scaffold signature + the cached ``.md``
    (if either exists), and reports staleness WITHOUT ever GATHERing, DRAFTing,
    writing, committing, or seeding an empty manifest. The dashboard fires this
    on plain navigation so opening a never-built/stale review never DRAFTs — only
    an explicit ``force`` regenerate does.

    ``ancestor_uids``/``ancestor_levels`` thread the NESTED composition selection
    through GATHER and the cache signature (see :func:`gather_outline_material` /
    :func:`compute_generation_signature`): a flat request (no ``ancestor_uids`` and
    ``ancestor_levels == 0``) is the flat single-spine/theme partition,
    byte-identical to before; an explicit ``ancestor_uids`` prune (authoritative) or
    a non-zero ``ancestor_levels`` (fallback) stacks the kept ancestor framing tiers
    above the subject's leaf partition and caches separately per DISTINCT selection.

    Returns ``{markdown, artifact_path, signature, stale, violations, cached,
    exists}`` (a ``peek`` also carries ``stored_signature``). The Wave-5 MCP tool
    + POST endpoint are thin delegates over this.
    """
    outline_id = outline_id or outlines_mod.DEFAULT_OUTLINE_ID
    base = Path(graphs_dir) if graphs_dir is not None else GRAPHS_DIR
    if not name:
        name = graph or project or "all"
    validate_id(name, kind="review name", for_filename=True)

    md_path = outlines_mod.draft_path(name, outline_id, base)
    yaml_path = base / "_reviews" / f"{name}.yaml"

    signature = compute_outline_signature(
        get_graph, project=project, graph=graph, name=name, outline_id=outline_id,
        claim_ids=claim_ids, as_of=as_of, spine=spine, spine_mode=spine_mode, ancestor_levels=ancestor_levels,
        ancestor_uids=ancestor_uids, graphs_dir=base, skill_path=skill_path,
        localize=localize,
    )

    # The W3 NESTED-composition degradation block for the read-only return paths
    # (peek, cache hit) — recomputed cheaply from the SAME nesting resolve GATHER
    # and the cache signature use, so cached/peek payloads carry the SAME block a
    # fresh build would (``None`` on a flat request ⇒ the key is dropped).
    _cached_nesting_degradation = _resolve_nesting(
        get_graph, project=project, graph=graph, base=base,
        loc=(localize or (lambda s: s)), spine=spine, claim_ids=claim_ids,
        ancestor_levels=ancestor_levels, ancestor_uids=ancestor_uids,
    )[1]

    # ── peek: read-only staleness probe (NEVER gather/draft/write/seed) ────────
    if peek:
        manifest = load_review(name, graphs_dir=base)
        scaffold = outlines_mod.read_draft_cache(
            name, outline_id, graphs_dir=base, manifest=manifest
        )
        stored_sig = (
            scaffold.get("signature") if isinstance(scaffold, dict) else None
        ) or ""
        cached_md = ""
        if md_path.exists():
            try:
                cached_md = md_path.read_text(encoding="utf-8")
            except OSError:
                cached_md = ""
        return _attach_nesting_degradation({
            "markdown": cached_md,
            "artifact_path": str(md_path),
            "signature": signature,
            "content_signature": compute_content_signature(cached_md),
            "stored_signature": stored_sig,
            "stale": bool(cached_md and stored_sig != signature),
            "violations": [],
            "cached": bool(cached_md),
            "exists": bool(cached_md),
            "deepened": sorted(_load_deepened({"scaffold": scaffold})),
        }, _cached_nesting_degradation)

    # ── cache check: unchanged signature + extant artifact ⇒ no agent call ─────
    if not force:
        manifest = load_review(name, graphs_dir=base)
        scaffold = outlines_mod.read_draft_cache(
            name, outline_id, graphs_dir=base, manifest=manifest
        )
        if (
            isinstance(scaffold, dict)
            and scaffold.get("signature") == signature
            and md_path.exists()
        ):
            try:
                cached_md = md_path.read_text(encoding="utf-8")
            except OSError:
                cached_md = None
            if cached_md is not None:
                return _attach_nesting_degradation({
                    "markdown": cached_md,
                    "artifact_path": str(md_path),
                    "signature": signature,
                    "content_signature": compute_content_signature(cached_md),
                    "stale": False,
                    "violations": [],
                    "cached": True,
                    "exists": True,
                    "deepened": sorted(_load_deepened({"scaffold": scaffold})),
                }, _cached_nesting_degradation)

    # ── GATHER → (incremental | full) DRAFT ────────────────────────────────────
    material = gather_outline_material(
        get_graph, project=project, graph=graph, name=name, outline_id=outline_id,
        claim_ids=claim_ids, as_of=as_of, spine=spine, spine_mode=spine_mode, ancestor_levels=ancestor_levels,
        ancestor_uids=ancestor_uids, graphs_dir=base, namespace=namespace, localize=localize,
    )
    prior_scaffold = outlines_mod.read_draft_cache(name, outline_id, graphs_dir=base)
    deepened = _load_deepened({"scaffold": prior_scaffold})

    # ── incremental regenerate: re-draft ONLY the sections whose material changed ─
    # With ``force`` false on a FLAT outline, carry every unchanged section over
    # byte-identically from the prior artifact and re-draft just the changed/new
    # ones (:func:`_rebuild_incremental`). ``None`` (a first build, a nested
    # outline, or no reusable section) falls through to the historical FULL draft.
    draft_md = None
    section_prov = None
    if not force and not _is_nested_material(material):
        prior_sections = _load_section_provenance(prior_scaffold)
        prior_md = ""
        if prior_sections and md_path.exists():
            try:
                prior_md = md_path.read_text(encoding="utf-8")
            except OSError:
                prior_md = ""
        if prior_sections and prior_md.strip():
            incremental = _rebuild_incremental(
                material, prior_md, prior_sections, deepened,
                draft_fn=draft_fn, skill_path=skill_path,
            )
            if incremental is not None:
                draft_md, section_prov, deepened = incremental

    # ── FULL draft (first build / force / nested / no reusable section) ─────────
    # A regenerate redraws the whole document; carry every remembered deepened
    # section forward from the prior artifact so it is never silently collapsed.
    # When the whole-material payload would overflow the draft model's context
    # (a large review), draft section-by-section against narrowed per-section
    # projections instead of stuffing all of it into one prompt.
    if draft_md is None:
        if _should_draft_by_section(material):
            draft_md, section_prov = _draft_by_section(
                material, draft_fn=draft_fn, skill_path=skill_path
            )
        else:
            draft_md = draft(material, draft_fn=draft_fn, skill_path=skill_path)
            section_prov = _section_provenance(material)
        draft_md, deepened = _merge_deepened_from_disk(draft_md, md_path, deepened)

    # ── integrity post-pass over the (possibly merged) document ────────────────
    violations = verify_draft(draft_md, material)
    final_md = _annotate_violations(draft_md, violations)

    # ── durable, crash-safe, git-backed write (REUSE the M3a substrate) ────────
    _persist_draft(
        base=base, name=name, project=project, graph=graph,
        md_path=md_path, yaml_path=yaml_path, final_md=final_md,
        signature=signature, deepened=deepened, sections=section_prov,
        outline_id=outline_id,
    )

    return _attach_nesting_degradation({
        "markdown": final_md,
        "artifact_path": str(md_path),
        "signature": signature,
        "content_signature": compute_content_signature(final_md),
        "stale": False,
        "violations": [asdict(v) for v in violations],
        "cached": False,
        "exists": True,
        "deepened": sorted(deepened),
    }, material.nesting_degradation)

stream_outline async

stream_outline(get_graph: GetGraph, *, project: str = '', graph: str = '', name: str = '', outline_id: str = DEFAULT_OUTLINE_ID, claim_ids: 'list[str] | None' = None, as_of: 'str | int | None' = None, spine: str = '', spine_mode: str = '', ancestor_levels: int = 0, ancestor_uids: 'list[str] | None' = None, force: bool = False, draft_stream: 'Callable[[str, str], Any] | None' = None, graphs_dir: 'Path | None' = None, skill_path: 'str | Path | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None)

The STREAMING orchestrator: same GATHER→cache→DRAFT→integrity→write as :func:build_outline, but an async generator that yields PROGRESS so the build is never an opaque, crash-looking black box.

Yields plain event dicts {"event": <name>, "data": {...}} (the route formats them as SSE — this stays framework-agnostic):

• stage — a phase marker (gather → gathered → draft → verify), each with a short human message (and the GATHER counts on gathered) so the UI shows what is happening.
• token — one assistant delta of the scaffold AS it is drafted, so the prose streams in live.
• result — the terminal payload, identical in shape to :func:build_outline's return (markdown/signature/violations/ cached/…), so the frontend reconciles it exactly like the JSON build.

A cache hit (unchanged signature + extant artifact, force false) emits a single result with cached=True and no agent call. peek is NOT a streaming concern — navigation still uses the cost-free JSON :func:build_outline peek path. Errors are raised; the route wraps them into a terminal error event.

Source code in zettelkasten/outline/__init__.py

async def stream_outline(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    name: str = "",
    outline_id: str = outlines_mod.DEFAULT_OUTLINE_ID,
    claim_ids: "list[str] | None" = None,
    as_of: "str | int | None" = None,
    spine: str = "",
    spine_mode: str = "",
    ancestor_levels: int = 0,
    ancestor_uids: "list[str] | None" = None,
    force: bool = False,
    draft_stream: "Callable[[str, str], Any] | None" = None,
    graphs_dir: "Path | None" = None,
    skill_path: "str | Path | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
):
    """The STREAMING orchestrator: same GATHER→cache→DRAFT→integrity→write as
    :func:`build_outline`, but an async generator that yields PROGRESS so the
    build is never an opaque, crash-looking black box.

    Yields plain event dicts ``{"event": <name>, "data": {...}}`` (the route
    formats them as SSE — this stays framework-agnostic):

    * ``stage`` — a phase marker (``gather`` → ``gathered`` → ``draft`` →
      ``verify``), each with a short human ``message`` (and the GATHER counts on
      ``gathered``) so the UI shows what is happening.
    * ``token`` — one assistant delta of the scaffold AS it is drafted, so the
      prose streams in live.
    * ``result`` — the terminal payload, identical in shape to
      :func:`build_outline`'s return (``markdown``/``signature``/``violations``/
      ``cached``/…), so the frontend reconciles it exactly like the JSON build.

    A cache hit (unchanged signature + extant artifact, ``force`` false) emits a
    single ``result`` with ``cached=True`` and no agent call. ``peek`` is NOT a
    streaming concern — navigation still uses the cost-free JSON
    :func:`build_outline` ``peek`` path. Errors are raised; the route wraps them
    into a terminal ``error`` event.
    """
    import asyncio

    outline_id = outline_id or outlines_mod.DEFAULT_OUTLINE_ID
    base = Path(graphs_dir) if graphs_dir is not None else GRAPHS_DIR
    if not name:
        name = graph or project or "all"
    validate_id(name, kind="review name", for_filename=True)

    md_path = outlines_mod.draft_path(name, outline_id, base)
    yaml_path = base / "_reviews" / f"{name}.yaml"

    signature = compute_outline_signature(
        get_graph, project=project, graph=graph, name=name, outline_id=outline_id,
        claim_ids=claim_ids, as_of=as_of, spine=spine, spine_mode=spine_mode, ancestor_levels=ancestor_levels,
        ancestor_uids=ancestor_uids, graphs_dir=base, skill_path=skill_path,
        localize=localize,
    )

    # The W3 NESTED-composition degradation block for the read-only cache-hit path —
    # recomputed cheaply from the SAME nesting resolve GATHER and the cache signature
    # use, so a streamed cache hit carries the SAME block a fresh build would
    # (``None`` on a flat request ⇒ the key is dropped).
    _cached_nesting_degradation = _resolve_nesting(
        get_graph, project=project, graph=graph, base=base,
        loc=(localize or (lambda s: s)), spine=spine, claim_ids=claim_ids,
        ancestor_levels=ancestor_levels, ancestor_uids=ancestor_uids,
    )[1]

    # ── cache check: unchanged signature + extant artifact ⇒ no agent call ─────
    if not force:
        manifest = load_review(name, graphs_dir=base)
        scaffold = outlines_mod.read_draft_cache(
            name, outline_id, graphs_dir=base, manifest=manifest
        )
        if (
            isinstance(scaffold, dict)
            and scaffold.get("signature") == signature
            and md_path.exists()
        ):
            try:
                cached_md = md_path.read_text(encoding="utf-8")
            except OSError:
                cached_md = None
            if cached_md is not None:
                yield {"event": "result", "data": _attach_nesting_degradation({
                    "markdown": cached_md,
                    "artifact_path": str(md_path),
                    "signature": signature,
                    "content_signature": compute_content_signature(cached_md),
                    "stale": False,
                    "violations": [],
                    "cached": True,
                    "exists": True,
                    "deepened": sorted(_load_deepened({"scaffold": scaffold})),
                }, _cached_nesting_degradation)}
                return

    # ── GATHER (deterministic; off the event loop so streaming stays live) ─────
    yield {"event": "stage", "data": {"phase": "gather", "message": "Gathering grounded material…"}}
    material = await asyncio.to_thread(
        gather_outline_material,
        get_graph, project=project, graph=graph, name=name, outline_id=outline_id,
        claim_ids=claim_ids, as_of=as_of, spine=spine, spine_mode=spine_mode, ancestor_levels=ancestor_levels,
        ancestor_uids=ancestor_uids, graphs_dir=base, namespace=namespace, localize=localize,
    )
    claim_count = sum(len(s.claims) for s in material.sections) + len(material.unplaced)
    yield {"event": "stage", "data": {
        "phase": "gathered",
        "message": "Material gathered — drafting scaffold…",
        "sections": len(material.sections),
        "claims": claim_count,
        "notes": len(material.note_ids),
    }}

    streamer = draft_stream or _default_draft_stream
    prior_scaffold = outlines_mod.read_draft_cache(name, outline_id, graphs_dir=base)
    deepened = _load_deepened({"scaffold": prior_scaffold})

    # ── incremental regenerate: re-draft ONLY the sections whose material changed ─
    # Mirrors :func:`build_outline`: with ``force`` false on a FLAT outline, the
    # shared synchronous :func:`_rebuild_incremental` runs off the event loop
    # (draining ``draft_stream`` per changed section via :func:`_sync_draft_from_stream`)
    # so the one incremental implementation serves both paths. Per-section tokens
    # are not forwarded here (a stage marker stands in); ``None`` falls through to
    # the historical FULL streamed draft below.
    draft_text = None
    section_prov = None
    if not force and not _is_nested_material(material):
        prior_sections = _load_section_provenance(prior_scaffold)
        prior_md = ""
        if prior_sections and md_path.exists():
            try:
                prior_md = md_path.read_text(encoding="utf-8")
            except OSError:
                prior_md = ""
        if prior_sections and prior_md.strip():
            yield {"event": "stage", "data": {
                "phase": "draft", "message": "Re-drafting changed sections…"
            }}
            incremental = await asyncio.to_thread(
                _rebuild_incremental,
                material, prior_md, prior_sections, deepened,
                draft_fn=_sync_draft_from_stream(streamer), skill_path=skill_path,
            )
            if incremental is not None:
                draft_text, section_prov, deepened = incremental

    # ── FULL draft (streamed token-by-token, or section-by-section if oversized) ─
    if draft_text is None and _should_draft_by_section(material):
        # The whole-material payload would overflow the draft model's context, so
        # draft section-by-section off the event loop (draining the stream per
        # section via _sync_draft_from_stream). Per-section tokens are not
        # forwarded here — a stage marker stands in, mirroring the incremental path.
        yield {"event": "stage", "data": {
            "phase": "draft", "message": "Drafting scaffold section-by-section…"
        }}
        draft_text, section_prov = await asyncio.to_thread(
            _draft_by_section,
            material,
            draft_fn=_sync_draft_from_stream(streamer), skill_path=skill_path,
        )
        draft_text, deepened = await asyncio.to_thread(
            _merge_deepened_from_disk, draft_text, md_path, deepened
        )

    if draft_text is None:
        contract, _version = _load_skill_contract(skill_path)
        prompt = _build_draft_prompt(material)
        yield {"event": "stage", "data": {"phase": "draft", "message": "Drafting scaffold…"}}

        draft_text = ""
        async for kind, payload in streamer(contract, prompt):
            if kind == "assistant":
                yield {"event": "token", "data": {"text": payload}}
            elif kind == "result":
                draft_text = payload

        # preserve user-DEEPENED sections (intelligent, non-collapsing regen)
        draft_text, deepened = await asyncio.to_thread(
            _merge_deepened_from_disk, draft_text, md_path, deepened
        )
        section_prov = _section_provenance(material)

    # ── integrity post-pass + durable write ────────────────────────────────────
    yield {"event": "stage", "data": {"phase": "verify", "message": "Checking integrity…"}}
    violations = verify_draft(draft_text, material)
    final_md = _annotate_violations(draft_text, violations)
    await asyncio.to_thread(
        _persist_draft,
        base=base, name=name, project=project, graph=graph,
        md_path=md_path, yaml_path=yaml_path, final_md=final_md,
        signature=signature, deepened=deepened, sections=section_prov,
        outline_id=outline_id,
    )

    yield {"event": "result", "data": _attach_nesting_degradation({
        "markdown": final_md,
        "artifact_path": str(md_path),
        "signature": signature,
        "content_signature": compute_content_signature(final_md),
        "stale": False,
        "violations": [asdict(v) for v in violations],
        "cached": False,
        "exists": True,
        "deepened": sorted(deepened),
    }, material.nesting_degradation)}

expand_outline_section

expand_outline_section(get_graph: GetGraph, *, project: str = '', graph: str = '', name: str = '', outline_id: str = DEFAULT_OUTLINE_ID, header: str = '', occurrence: int = 0, claim_ids: 'list[str] | None' = None, as_of: 'str | int | None' = None, spine: str = '', spine_mode: str = '', ancestor_levels: int = 0, ancestor_uids: 'list[str] | None' = None, draft_fn: 'Callable[[str, str], str] | None' = None, graphs_dir: 'Path | None' = None, skill_path: 'str | Path | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None) -> dict[str, Any]

Re-draft ONE section of the stored scaffold deeper, splicing it back in.

Reads the persisted _reviews/<name>.md, locates the section under header (:func:_locate_section), GATHERs the scope's material, asks the agent to DEEPEN only that section (:func:draft_section_expansion), splices the replacement back into the full document, runs the SAME deterministic :func:verify_draft integrity pass over the whole spliced doc, and persists it through the shared :func:_persist_draft write path. The generation signature is unchanged (the scope inputs did not change — only the derived prose got richer), so a deepened scaffold does not read as stale.

Returns the same payload shape as :func:build_outline.

Source code in zettelkasten/outline/__init__.py

def expand_outline_section(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    name: str = "",
    outline_id: str = outlines_mod.DEFAULT_OUTLINE_ID,
    header: str = "",
    occurrence: int = 0,
    claim_ids: "list[str] | None" = None,
    as_of: "str | int | None" = None,
    spine: str = "",
    spine_mode: str = "",
    ancestor_levels: int = 0,
    ancestor_uids: "list[str] | None" = None,
    draft_fn: "Callable[[str, str], str] | None" = None,
    graphs_dir: "Path | None" = None,
    skill_path: "str | Path | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> dict[str, Any]:
    """Re-draft ONE section of the stored scaffold deeper, splicing it back in.

    Reads the persisted ``_reviews/<name>.md``, locates the section under
    ``header`` (:func:`_locate_section`), GATHERs the scope's material, asks the
    agent to DEEPEN only that section (:func:`draft_section_expansion`), splices
    the replacement back into the full document, runs the SAME deterministic
    :func:`verify_draft` integrity pass over the whole spliced doc, and
    persists it through the shared :func:`_persist_draft` write path. The
    generation signature is unchanged (the scope inputs did not change — only the
    derived prose got richer), so a deepened scaffold does not read as stale.

    Returns the same payload shape as :func:`build_outline`.
    """
    base = Path(graphs_dir) if graphs_dir is not None else GRAPHS_DIR
    if not name:
        name = graph or project or "all"
    validate_id(name, kind="review name", for_filename=True)
    if not header.strip():
        raise ValueError("A section header is required to deepen.")

    outline_id = outline_id or outlines_mod.DEFAULT_OUTLINE_ID
    md_path = outlines_mod.draft_path(name, outline_id, base)
    yaml_path = base / "_reviews" / f"{name}.yaml"
    current = _read_draft_for_expand(md_path)

    loc = _locate_section(current, header, occurrence)
    if loc is None:
        raise ValueError(f"Section not found in the scaffold: {header!r}")
    start, end = loc
    lines = current.splitlines()
    section_md = "\n".join(lines[start:end]).strip("\n")

    material = gather_outline_material(
        get_graph, project=project, graph=graph, name=name, outline_id=outline_id,
        claim_ids=claim_ids, as_of=as_of, spine=spine, spine_mode=spine_mode, ancestor_levels=ancestor_levels,
        ancestor_uids=ancestor_uids, graphs_dir=base, namespace=namespace, localize=localize,
    )
    new_section = _strip_section_fence(
        draft_section_expansion(material, section_md, draft_fn=draft_fn, skill_path=skill_path)
    )

    spliced = _splice_section(lines, start, end, new_section)
    violations = verify_draft(spliced, material)
    final_md = _annotate_violations(spliced, violations)

    signature = compute_outline_signature(
        get_graph, project=project, graph=graph, name=name, outline_id=outline_id,
        claim_ids=claim_ids, as_of=as_of, spine=spine, spine_mode=spine_mode, ancestor_levels=ancestor_levels,
        ancestor_uids=ancestor_uids, graphs_dir=base, skill_path=skill_path,
        localize=localize,
    )
    # Remember this section as DEEPENED so a future regenerate preserves it.
    # Key by the header as it lands in the artifact (a deepen may sharpen it).
    deepened = _record_deepened(
        _load_deepened({"scaffold": outlines_mod.read_draft_cache(name, outline_id, graphs_dir=base)}),
        header, new_section, signature,
    )
    _persist_draft(
        base=base, name=name, project=project, graph=graph,
        md_path=md_path, yaml_path=yaml_path, final_md=final_md,
        signature=signature, deepened=deepened, outline_id=outline_id,
    )

    return {
        "markdown": final_md,
        "artifact_path": str(md_path),
        "signature": signature,
        "content_signature": compute_content_signature(final_md),
        "stale": False,
        "violations": [asdict(v) for v in violations],
        "cached": False,
        "exists": True,
        "deepened": sorted(deepened),
    }

stream_section_expansion async

stream_section_expansion(get_graph: GetGraph, *, project: str = '', graph: str = '', name: str = '', outline_id: str = DEFAULT_OUTLINE_ID, header: str = '', occurrence: int = 0, claim_ids: 'list[str] | None' = None, as_of: 'str | int | None' = None, spine: str = '', spine_mode: str = '', ancestor_levels: int = 0, ancestor_uids: 'list[str] | None' = None, draft_stream: 'Callable[[str, str], Any] | None' = None, graphs_dir: 'Path | None' = None, skill_path: 'str | Path | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None)

Streaming twin of :func:expand_outline_section (stage/token/result frames).

Same GATHER → DEEPEN → integrity → splice → write as the sync path, but an async generator that streams the section AS it is re-drafted so the deepen is not an opaque spinner. The token frames carry the new SECTION's deltas (not the whole document); the terminal result carries the full spliced markdown so the frontend swaps the finished document in exactly like a build.

Source code in zettelkasten/outline/__init__.py

async def stream_section_expansion(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    name: str = "",
    outline_id: str = outlines_mod.DEFAULT_OUTLINE_ID,
    header: str = "",
    occurrence: int = 0,
    claim_ids: "list[str] | None" = None,
    as_of: "str | int | None" = None,
    spine: str = "",
    spine_mode: str = "",
    ancestor_levels: int = 0,
    ancestor_uids: "list[str] | None" = None,
    draft_stream: "Callable[[str, str], Any] | None" = None,
    graphs_dir: "Path | None" = None,
    skill_path: "str | Path | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
):
    """Streaming twin of :func:`expand_outline_section` (stage/token/result frames).

    Same GATHER → DEEPEN → integrity → splice → write as the sync path, but an
    async generator that streams the section AS it is re-drafted so the deepen is
    not an opaque spinner. The ``token`` frames carry the new SECTION's deltas
    (not the whole document); the terminal ``result`` carries the full spliced
    markdown so the frontend swaps the finished document in exactly like a build.
    """
    import asyncio

    base = Path(graphs_dir) if graphs_dir is not None else GRAPHS_DIR
    if not name:
        name = graph or project or "all"
    validate_id(name, kind="review name", for_filename=True)
    if not header.strip():
        raise ValueError("A section header is required to deepen.")

    outline_id = outline_id or outlines_mod.DEFAULT_OUTLINE_ID
    md_path = outlines_mod.draft_path(name, outline_id, base)
    yaml_path = base / "_reviews" / f"{name}.yaml"
    current = _read_draft_for_expand(md_path)

    loc = _locate_section(current, header, occurrence)
    if loc is None:
        raise ValueError(f"Section not found in the scaffold: {header!r}")
    start, end = loc
    lines = current.splitlines()
    section_md = "\n".join(lines[start:end]).strip("\n")

    yield {"event": "stage", "data": {"phase": "gather", "message": "Gathering grounded material…"}}
    material = await asyncio.to_thread(
        gather_outline_material,
        get_graph, project=project, graph=graph, name=name, outline_id=outline_id,
        claim_ids=claim_ids, as_of=as_of, spine=spine, spine_mode=spine_mode, ancestor_levels=ancestor_levels,
        ancestor_uids=ancestor_uids, graphs_dir=base, namespace=namespace, localize=localize,
    )
    claim_count = sum(len(s.claims) for s in material.sections) + len(material.unplaced)
    yield {"event": "stage", "data": {
        "phase": "gathered",
        "message": f"Deepening “{header}”…",
        "sections": len(material.sections),
        "claims": claim_count,
        "notes": len(material.note_ids),
    }}

    contract, _version = _load_skill_contract(skill_path)
    prompt = _build_expand_prompt(material, section_md)
    yield {"event": "stage", "data": {"phase": "draft", "message": f"Deepening “{header}”…"}}

    streamer = draft_stream or _default_draft_stream
    new_text = ""
    async for kind, payload in streamer(contract, prompt):
        if kind == "assistant":
            yield {"event": "token", "data": {"text": payload}}
        elif kind == "result":
            new_text = payload

    new_section = _strip_section_fence(new_text)
    spliced = _splice_section(lines, start, end, new_section)

    yield {"event": "stage", "data": {"phase": "verify", "message": "Checking integrity…"}}
    violations = verify_draft(spliced, material)
    final_md = _annotate_violations(spliced, violations)
    signature = compute_outline_signature(
        get_graph, project=project, graph=graph, name=name, outline_id=outline_id,
        claim_ids=claim_ids, as_of=as_of, spine=spine, spine_mode=spine_mode, ancestor_levels=ancestor_levels,
        ancestor_uids=ancestor_uids, graphs_dir=base, skill_path=skill_path,
        localize=localize,
    )
    # Remember this section as DEEPENED so a future regenerate preserves it.
    deepened = _record_deepened(
        _load_deepened({"scaffold": outlines_mod.read_draft_cache(name, outline_id, graphs_dir=base)}),
        header, new_section, signature,
    )
    await asyncio.to_thread(
        _persist_draft,
        base=base, name=name, project=project, graph=graph,
        md_path=md_path, yaml_path=yaml_path, final_md=final_md,
        signature=signature, deepened=deepened, outline_id=outline_id,
    )

    yield {"event": "result", "data": {
        "markdown": final_md,
        "artifact_path": str(md_path),
        "signature": signature,
        "content_signature": compute_content_signature(final_md),
        "stale": False,
        "violations": [asdict(v) for v in violations],
        "cached": False,
        "exists": True,
        "deepened": sorted(deepened),
    }}

fix_outline_violation

fix_outline_violation(get_graph: GetGraph, *, project: str = '', graph: str = '', name: str = '', outline_id: str = DEFAULT_OUTLINE_ID, kind: str = '', ref: str = '', line: 'int | None' = None, action: str = '', expected_content_signature: 'str | None' = None, claim_ids: 'list[str] | None' = None, as_of: 'str | int | None' = None, spine: str = '', spine_mode: str = '', ancestor_levels: int = 0, ancestor_uids: 'list[str] | None' = None, extract_fn: 'Callable[[str, str], str] | None' = None, corpus_fn: 'Callable[[str], tuple[str | None, list[dict[str, Any]]]] | None' = None, graphs_dir: 'Path | None' = None, skill_path: 'str | Path | None' = None, namespace: 'Callable[[str], str] | None' = None, localize: 'Callable[[str], str] | None' = None) -> dict[str, Any]

Resolve ONE integrity violation in the stored scaffold and persist the draft.

Loads the persisted _reviews/<name>.md (banner peeled), rebuilds the scope's :class:OutlineMaterial, applies the requested action ("fix" or "remove") to the violation identified by kind/ref/line, re-runs the deterministic :func:verify_draft pass over the edited draft, re-annotates any remaining violations, and persists through the SAME crash-safe write path the build uses. The generation signature is unchanged (only derived prose moved), so the artifact does not read as stale and the user-DEEPENED provenance map is preserved.

The load → edit → persist runs under a SINGLE :func:review_write_lock acquisition, so a concurrent writer can never slip a lost update between the read and the write. When expected_content_signature is given it is checked against a :func:compute_content_signature of the CURRENTLY persisted draft markdown (read fresh INSIDE that lock); a mismatch means the draft moved on since the client loaded it (stale banner / second tab / 503-retry, OR a force/deepen that rewrote the prose for the SAME material) and :class:OutlineConflictError is raised (the route maps it to a 409) rather than editing stale line numbers and deleting the wrong content. A CONTENT hash — not the generation signature — is used precisely because the generation signature stays byte-identical across a force/deepen rewrite of the same material, so it cannot detect that the client's line now points elsewhere.

A FIX never fabricates: a quote fix only splices a passage re-verified against the source (grounding.verify_quote) AND present verbatim in a referenced quote note; a chip fix only rewrites to a unique fuzzy match above a fixed threshold. When nothing resolves the draft is left untouched (NOT rewritten / committed) and resolved=False is returned so the UI can fall back to Remove.

Returns the :func:build_outline payload shape plus resolved (bool) and message (str).

Source code in zettelkasten/outline/__init__.py

def fix_outline_violation(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    name: str = "",
    outline_id: str = outlines_mod.DEFAULT_OUTLINE_ID,
    kind: str = "",
    ref: str = "",
    line: "int | None" = None,
    action: str = "",
    expected_content_signature: "str | None" = None,
    claim_ids: "list[str] | None" = None,
    as_of: "str | int | None" = None,
    spine: str = "",
    spine_mode: str = "",
    ancestor_levels: int = 0,
    ancestor_uids: "list[str] | None" = None,
    extract_fn: "Callable[[str, str], str] | None" = None,
    corpus_fn: "Callable[[str], tuple[str | None, list[dict[str, Any]]]] | None" = None,
    graphs_dir: "Path | None" = None,
    skill_path: "str | Path | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    localize: "Callable[[str], str] | None" = None,
) -> dict[str, Any]:
    """Resolve ONE integrity violation in the stored scaffold and persist the draft.

    Loads the persisted ``_reviews/<name>.md`` (banner peeled), rebuilds the scope's
    :class:`OutlineMaterial`, applies the requested ``action`` (``"fix"`` or
    ``"remove"``) to the violation identified by ``kind``/``ref``/``line``, re-runs
    the deterministic :func:`verify_draft` pass over the edited draft, re-annotates
    any remaining violations, and persists through the SAME crash-safe write path the
    build uses. The generation signature is unchanged (only derived prose moved), so
    the artifact does not read as stale and the user-DEEPENED provenance map is
    preserved.

    The load → edit → persist runs under a SINGLE :func:`review_write_lock`
    acquisition, so a concurrent writer can never slip a lost update between the
    read and the write. When ``expected_content_signature`` is given it is checked
    against a :func:`compute_content_signature` of the CURRENTLY persisted draft
    markdown (read fresh INSIDE that lock); a mismatch means the draft moved on
    since the client loaded it (stale banner / second tab / 503-retry, OR a
    force/deepen that rewrote the prose for the SAME material) and
    :class:`OutlineConflictError` is raised (the route maps it to a 409) rather than
    editing stale line numbers and deleting the wrong content. A CONTENT hash —
    not the generation ``signature`` — is used precisely because the generation
    signature stays byte-identical across a force/deepen rewrite of the same
    material, so it cannot detect that the client's ``line`` now points elsewhere.

    A FIX never fabricates: a ``quote`` fix only splices a passage re-verified against
    the source (``grounding.verify_quote``) AND present verbatim in a referenced quote
    note; a ``chip`` fix only rewrites to a unique fuzzy match above a fixed threshold.
    When nothing resolves the draft is left untouched (NOT rewritten / committed) and
    ``resolved=False`` is returned so the UI can fall back to Remove.

    Returns the :func:`build_outline` payload shape plus ``resolved`` (bool) and
    ``message`` (str).
    """
    outline_id = outline_id or outlines_mod.DEFAULT_OUTLINE_ID
    base = Path(graphs_dir) if graphs_dir is not None else GRAPHS_DIR
    if not name:
        name = graph or project or "all"
    validate_id(name, kind="review name", for_filename=True)

    action = (action or "").strip().lower()
    if action not in ("fix", "remove"):
        raise ValueError("action must be 'fix' or 'remove'.")
    if kind not in ("chip", "quote", "header", "structure"):
        raise ValueError(f"unknown violation kind {kind!r}.")

    md_path = outlines_mod.draft_path(name, outline_id, base)
    yaml_path = base / "_reviews" / f"{name}.yaml"

    # The material (graph read) and signature (scope fingerprint) are independent
    # of the persisted draft, so compute them outside the write lock to keep the
    # lock hold tight; only the draft read-modify-write needs serializing.
    material = gather_outline_material(
        get_graph, project=project, graph=graph, name=name, outline_id=outline_id,
        claim_ids=claim_ids, as_of=as_of, spine=spine, spine_mode=spine_mode, ancestor_levels=ancestor_levels,
        ancestor_uids=ancestor_uids, graphs_dir=base, namespace=namespace, localize=localize,
    )
    signature = compute_outline_signature(
        get_graph, project=project, graph=graph, name=name, outline_id=outline_id,
        claim_ids=claim_ids, as_of=as_of, spine=spine, spine_mode=spine_mode, ancestor_levels=ancestor_levels,
        ancestor_uids=ancestor_uids, graphs_dir=base, skill_path=skill_path,
        localize=localize,
    )
    expected_content = (expected_content_signature or "").strip()

    # ── atomic read-modify-write: one lock spans load → edit → verify → persist ──
    with review_write_lock(name, graphs_dir=base):
        current = _read_draft_for_expand(md_path)

        # Staleness guard: refuse to act when the persisted draft has moved on
        # since the client loaded it. Compared against a CONTENT hash of the draft
        # markdown actually on disk now (read fresh inside the lock) — NOT the
        # generation signature, which stays byte-identical across a force/deepen
        # that rewrote the prose for the SAME material and would let a stale
        # client edit/delete the wrong ``line``.
        if expected_content and compute_content_signature(current) != expected_content:
            raise OutlineConflictError(
                "The outline changed since you loaded it — refresh and try again."
            )

        draft = _strip_integrity_banner(current)

        if action == "remove":
            new_draft, changed, message = _apply_remove(draft, kind=kind, ref=ref, line=line)
        else:
            new_draft, changed, message = _apply_fix(
                draft, material, kind=kind, ref=ref, line=line,
                extract_fn=extract_fn, corpus_fn=corpus_fn,
            )

        if not changed:
            # Nothing resolved — re-check the UNCHANGED draft so the caller still
            # gets the current violations, but never rewrite or commit the artifact.
            violations = verify_draft(draft, material)
            return {
                "markdown": current,
                "artifact_path": str(md_path),
                "signature": signature,
                "content_signature": compute_content_signature(current),
                "stale": False,
                "violations": [asdict(v) for v in violations],
                "resolved": False,
                "message": message,
                "cached": False,
                "exists": True,
                "deepened": sorted(_load_deepened(
                    {"scaffold": outlines_mod.read_draft_cache(name, outline_id, graphs_dir=base)}
                )),
            }

        violations = verify_draft(new_draft, material)
        final_md = _annotate_violations(new_draft, violations)
        deepened = _load_deepened(
            {"scaffold": outlines_mod.read_draft_cache(name, outline_id, graphs_dir=base)}
        )
        _persist_draft_locked(
            base=base, name=name, project=project, graph=graph,
            md_path=md_path, yaml_path=yaml_path, final_md=final_md,
            signature=signature, deepened=None, outline_id=outline_id,
        )

    return {
        "markdown": final_md,
        "artifact_path": str(md_path),
        "signature": signature,
        "content_signature": compute_content_signature(final_md),
        "stale": False,
        "violations": [asdict(v) for v in violations],
        "resolved": True,
        "message": message,
        "cached": False,
        "exists": True,
        "deepened": sorted(deepened),
    }