zettelkasten.tables_nested_spines

zettelkasten.tables_nested_spines

Nested-spine composition walker + integrity (the isolation seam).

Mechanically split out of tables.py with IDENTICAL semantics: the read-only detect_nested_spines walker, nested_spine_integrity composition check, and the propagate_stale_rollup drift roll-up.

detect_nested_spines

detect_nested_spines(get_graph: GetGraph, localize: 'Callable[[str], str] | None' = None, *, members: 'list[dict[str, Any]]', scope_uids: 'set[str] | None' = None) -> dict[str, Any]

Assemble the multi-tier nested-spine composition from a set of spines.

Walks the cross-graph PRIMARY component-of edges (design §7/§8): a spine with NO primary parent is a ROOT (whether it has primary children — the top of a nested composition — or none — a degenerate/standalone single-node root); one nested-spine descriptor is emitted per root, expandable to full depth (N-tier — sector>theme>component and deeper, not just 2 tiers). Everything is uid-keyed ("<graph>::<id>") and depth-annotated so a Graph-3 overlay endpoint and a Graph-4 N-ring radial / outline can consume it directly.

members are the composition's spine orgs (each a dict with a spine_ref graph name, and optionally id / title), mirroring :func:spine_group_matrix. scope_uids (optional) is the set of base-note "<home>::<id>" uids in scope; any in scope but claimed by NO spine lands in the top-level "Unplaced notes" bucket (design §5). When None the Unplaced bucket is omitted (scope is caller-defined and unknown to the walker).

Lazy degradation (design §5, Seam 4): a spine whose primary parent graph is deleted / unresolvable is NOT dropped — it degrades to an "Unassigned" bucket at read time; edges are never scrubbed. Cross-graph cycles are guarded (uid-keyed visited sets), never infinite-looping.

Returns {kind, roots, unassigned, unplaced, index, placement} where roots is a list of spine descriptor trees (one per parentless spine), unassigned a list of bucket descriptors (dangling-parent and cycle spines, each holding its own subtree), unplaced a bucket descriptor or None, index a flat uid-keyed lookup of every emitted spine node, and placement the {note_uid: owner_uid} partition.

Source code in zettelkasten/tables_nested_spines.py

def detect_nested_spines(
    get_graph: GetGraph,
    localize: "Callable[[str], str] | None" = None,
    *,
    members: "list[dict[str, Any]]",
    scope_uids: "set[str] | None" = None,
) -> dict[str, Any]:
    """Assemble the multi-tier nested-spine composition from a set of spines.

    Walks the cross-graph PRIMARY ``component-of`` edges (design §7/§8): a spine
    with NO primary parent is a ROOT (whether it has primary children — the top of
    a nested composition — or none — a degenerate/standalone single-node root); one
    nested-spine descriptor is emitted per root, expandable to full depth (N-tier —
    sector>theme>component and deeper, not just 2 tiers). Everything is uid-keyed
    (``"<graph>::<id>"``) and depth-annotated so a Graph-3 overlay endpoint and a
    Graph-4 N-ring radial / outline can consume it directly.

    ``members`` are the composition's spine orgs (each a dict with a ``spine_ref``
    graph name, and optionally ``id`` / ``title``), mirroring
    :func:`spine_group_matrix`. ``scope_uids`` (optional) is the set of base-note
    ``"<home>::<id>"`` uids in scope; any in scope but claimed by NO spine lands in
    the top-level "Unplaced notes" bucket (design §5). When ``None`` the Unplaced
    bucket is omitted (scope is caller-defined and unknown to the walker).

    Lazy degradation (design §5, Seam 4): a spine whose primary parent graph is
    deleted / unresolvable is NOT dropped — it degrades to an "Unassigned" bucket at
    read time; edges are never scrubbed. Cross-graph cycles are guarded (uid-keyed
    visited sets), never infinite-looping.

    Returns ``{kind, roots, unassigned, unplaced, index, placement}`` where ``roots``
    is a list of spine descriptor trees (one per parentless spine), ``unassigned``
    a list of bucket descriptors (dangling-parent and cycle spines, each holding its
    own subtree), ``unplaced`` a bucket descriptor or ``None``, ``index`` a flat
    uid-keyed lookup of every emitted spine node, and ``placement`` the
    ``{note_uid: owner_uid}`` partition.
    """
    loc = localize or (lambda s: s)

    # One shared graph cache for the WHOLE walk, keyed by the LOCALIZED name so it
    # is shared with the single-graph reader (_spine_membership_index calls
    # get_graph(loc(name)) too) — each spine graph loads at most once. Generalizes
    # spine_group_matrix's per-op cache to the cross-graph tree.
    graph_cache: dict[str, Any] = {}

    def _load(localized: str) -> Any:
        if localized in graph_cache:
            val = graph_cache[localized]
            return None if val is False else val
        try:
            zg = get_graph(localized)
        except Exception:  # noqa: BLE001 — an unreadable spine graph is skipped, not fatal
            zg = False
        graph_cache[localized] = zg
        return None if zg is False else zg

    def _cached_get_graph(localized: str) -> Any:
        zg = _load(localized)
        if zg is None:
            raise KeyError(localized)
        return zg

    def _resolve(name: str) -> Any:
        return _load(loc(name))

    def _apex_getter(raw_name: str) -> Any:
        # spine.find_apex_id expects get_graph(<raw graph name>); localize + cache.
        return _cached_get_graph(loc(raw_name))

    ops = _NestedWalkerOps(_resolve)

    org_id_by_graph: dict[str, str] = {}
    org_title_by_graph: dict[str, str] = {}
    seed_graphs: list[str] = []
    for org in members:
        graph = str(org.get("spine_ref") or "").strip()
        if not graph:
            continue
        seed_graphs.append(graph)
        org_id_by_graph.setdefault(graph, str(org.get("id") or ""))
        org_title_by_graph.setdefault(graph, str(org.get("title") or ""))

    walk = _nested_primary_children(
        seed_graphs, resolve=_resolve, ops=ops, apex_getter=_apex_getter
    )
    nodes: set[str] = walk["nodes"]
    children: dict[str, list[str]] = walk["children"]
    parent: dict[str, str] = walk["parent"]
    dangling: dict[str, dict[str, str]] = walk["dangling"]
    graph_of_uid: dict[str, str] = walk["graph_of_uid"]
    apex_of_uid: dict[str, str] = walk["apex_of_uid"]
    title_of_uid: dict[str, str] = walk["title_of_uid"]

    depths = _nested_spine_depths(nodes, parent)
    membership = _nested_membership_index(
        walk, depths, get_graph=_cached_get_graph, loc=loc
    )
    placed: dict[str, set[str]] = membership["placed"]
    rollup: dict[str, set[str]] = membership["rollup"]
    placement: dict[str, str] = membership["placement"]

    index: dict[str, dict[str, Any]] = {}

    def _spine_descriptor(
        uid: str, depth: int, on_path: "frozenset[str]", seen: "set[str]"
    ) -> "dict[str, Any] | None":
        # Cross-graph cycle guard: never revisit a uid on the current path, and
        # never place the same spine twice across the forest (uid-keyed).
        if uid in on_path or uid in seen:
            return None
        seen.add(uid)
        kids: list[dict[str, Any]] = []
        for child_uid in children.get(uid, ()):  # noqa: SIM118
            sub = _spine_descriptor(child_uid, depth + 1, on_path | {uid}, seen)
            if sub is not None:
                kids.append(sub)
        graph = graph_of_uid[uid]
        rolled = rollup.get(uid, set())
        node = {
            "uid": uid,
            "kind": "spine",
            "spine_ref": graph,
            "graph": graph,
            "node_id": apex_of_uid[uid],
            "apex_id": apex_of_uid[uid],
            "org_id": org_id_by_graph.get(graph, ""),
            "label": org_title_by_graph.get(graph) or title_of_uid.get(uid, apex_of_uid[uid]),
            "depth": depth,
            "parent": parent.get(uid),
            "owned": sorted(placed.get(uid, set())),
            "rollup": sorted(rolled),
            "count": len(rolled),
            "children": kids,
        }
        index[uid] = {
            "uid": uid,
            "kind": "spine",
            "graph": graph,
            "node_id": apex_of_uid[uid],
            "depth": depth,
            "parent": parent.get(uid),
            "label": node["label"],
        }
        return node

    placed_globally: set[str] = set()

    # Roots: parentless, non-dangling spines (design §7). A parentless spine WITH
    # >= 1 primary child is the top of a nested composition; a parentless spine
    # with NO primary children — an apex with no wired primary edges (common
    # mid-build: a member spine included before ``connect_sub_spine_edge`` wires
    # it) — is a degenerate/standalone root, emitted as its own single-node root
    # (empty children). It is NOT dangling, NOT in a cycle, so it must never fall
    # through to the cycle-survivor bucket below.
    root_uids = sorted(
        uid
        for uid in nodes
        if uid not in parent and uid not in dangling
    )
    roots: list[dict[str, Any]] = []
    for uid in root_uids:
        desc = _spine_descriptor(uid, 1, frozenset(), placed_globally)
        if desc is not None:
            roots.append(desc)

    # Unassigned buckets: a dangling-parent spine (design §5, Seam 4) becomes a
    # top-level bucket holding its own subtree, so nothing is silently dropped.
    unassigned: list[dict[str, Any]] = []
    for uid in sorted(dangling):
        if uid in placed_globally:
            continue
        subtree = _spine_descriptor(uid, 2, frozenset(), placed_globally)
        if subtree is None:
            continue
        miss = dangling[uid]
        bucket_uid = f"{_UNASSIGNED_SPINE_ID}::{uid}"
        unassigned.append(
            {
                "uid": bucket_uid,
                "kind": "unassigned",
                "reason": "missing-parent",
                "label": "Unassigned",
                "depth": 1,
                "missing_parent": {"graph": miss["graph"], "node_id": miss["node_id"]},
                "children": [subtree],
                "rollup": subtree["rollup"],
                "count": subtree["count"],
            }
        )

    # Cycle survivors: any spine still unplaced after roots + dangling buckets has
    # a primary parent (parentless spines all became roots above) yet is reachable
    # from NO root — so it sits in a cross-graph cycle (A→B→A) with no parentless
    # entry. Surface it (never drop it) as its own Unassigned bucket, breaking the
    # cycle at this entry. ``reason="cycle"`` is thus reserved strictly for spines
    # actually proven to be in a cycle (a back-edge caught by the on_path guard).
    for uid in sorted(nodes):
        if uid in placed_globally:
            continue
        subtree = _spine_descriptor(uid, 2, frozenset(), placed_globally)
        if subtree is None:
            continue
        bucket_uid = f"{_UNASSIGNED_SPINE_ID}::cycle::{uid}"
        unassigned.append(
            {
                "uid": bucket_uid,
                "kind": "unassigned",
                "reason": "cycle",
                "label": "Unassigned",
                "depth": 1,
                "missing_parent": None,
                "children": [subtree],
                "rollup": subtree["rollup"],
                "count": subtree["count"],
            }
        )

    # Unplaced notes: base notes in scope claimed by NO spine in the composition
    # (design §5). Only computable when the caller supplies the scope.
    unplaced: "dict[str, Any] | None" = None
    if scope_uids is not None:
        claimed_union: set[str] = set()
        for claimed in membership["claimed_of"].values():
            claimed_union |= claimed
        orphaned = sorted(u for u in scope_uids if u not in claimed_union)
        if orphaned:
            unplaced = {
                "uid": _UNPLACED_NOTES_ID,
                "kind": "unplaced",
                "label": "Unplaced notes",
                "depth": 1,
                "members": orphaned,
                "count": len(orphaned),
            }

    return {
        "kind": "nested-spines",
        "roots": roots,
        "unassigned": unassigned,
        "unplaced": unplaced,
        "index": index,
        "placement": placement,
    }

nested_spine_integrity

nested_spine_integrity(composition: 'dict[str, Any]', *, members: 'list[dict[str, Any]]') -> 'dict[str, Any]'

Composition-completeness check over a :func:detect_nested_spines result.

Phase-5 integrity (design documents/260702_nested-spines.md §10/§5): every seeded spine must have a home — placed in a root chain, a degenerate standalone root, or an explicit Unassigned bucket (missing-parent / cycle) — and nothing may be silently dropped (a seed whose apex the walker could not resolve, so it emitted NO node for it). This is a PURE, deterministic rollup ON TOP of the walker output (§11: the walker itself is the isolation seam and is never modified here); it loads no graphs and does no I/O.

composition is the walker's {kind, roots, unassigned, unplaced, index, placement} dict. members is the SAME seed list passed to the walker (it does not echo its seeds), each an org dict with a spine_ref graph name and optional id/org_id. A member with a non-empty spine_ref is a seed; an empty spine_ref is an invalid-seed.

seeds_accounted_for covers ONLY the structural fate of the SEEDED spines: it is True iff every non-empty-spine_ref seed maps to its OWN distinct emitted home (a root chain, a standalone root, or a known Unassigned bucket) — i.e. no seed is silently dropped (walker resolved no apex → emitted no node) and no two seeds collided onto ONE emitted uid. Because the walker emits exactly one node per spine graph (spine.find_apex_id yields one apex per graph and the org↔graph map keeps only the FIRST seed), two members sharing one spine_ref collapse onto a single home; the later seed(s) are a silent loss and are surfaced as collided (counted in counts["collided"], and each colliding org marked in status under a "<home-uid>::collided::<org_id>" key so the lost org is visible). seeds_accounted_for intentionally does NOT gate on unplaced_notes — that is a separate advisory count of in-scope base notes claimed by no spine, and the dashboard route passes scope_uids=None so the walker omits the Unplaced bucket and this count is always 0 there.

complete is a DEPRECATED alias of seeds_accounted_for (kept, with the identical value, for one release so existing readers keep working). It always over-promised — it means "every seed is accounted for", NOT "the graph is healthy" — so it is renamed to seeds_accounted_for; migrate readers to the new key and stop consuming complete.

There is intentionally NO cycles_incomplete flag. An earlier draft shipped one to flag a supposed root-attached-cycle BLIND SPOT (a cross-graph cycle whose entry is a parentless root, rendered as a broken child chain rather than an Unassigned(cycle) bucket), computed by following index[...]["parent"] up from each placed spine and reporting a loop. That guard was a false-negative dead guard: real roots carry parent=None and :func:detect_nested_spines renders children acyclically, so a placed node's canonical parent chain ALWAYS terminates at a parentless root — a genuine cross-graph cycle instead lands in unassigned and is reported in cycles. The only input that ever tripped the flag True was a walker-impossible hand-built composition. Computing it CORRECTLY would require the raw component-of parent relations, which this layer does not have (it is pure over the walker output and does no I/O) — and recovering them would mean changing the walker's traversal or return shape, the byte-identical isolation seam (§11) this function must not touch. Rather than ship a guard that is False on every real composition, the key was removed; genuine cross-graph cycles remain fully covered by cycles.

This is a PURE, deterministic rollup ON TOP of the walker output (§11: the walker is the isolation seam, never modified here); it loads no graphs and does no I/O.

Returns {total_seeded, counts, orphans, cycles, unplaced_notes, seeds_accounted_for, complete, status} — see the module design doc for field semantics.

Source code in zettelkasten/tables_nested_spines.py

def nested_spine_integrity(
    composition: "dict[str, Any]", *, members: "list[dict[str, Any]]"
) -> "dict[str, Any]":
    """Composition-completeness check over a :func:`detect_nested_spines` result.

    Phase-5 integrity (design ``documents/260702_nested-spines.md`` §10/§5): every
    seeded spine must have a *home* — placed in a root chain, a degenerate
    standalone root, or an explicit Unassigned bucket (missing-parent / cycle) —
    and nothing may be silently ``dropped`` (a seed whose apex the walker could not
    resolve, so it emitted NO node for it). This is a PURE, deterministic rollup
    ON TOP of the walker output (§11: the walker itself is the isolation seam and is
    never modified here); it loads no graphs and does no I/O.

    ``composition`` is the walker's ``{kind, roots, unassigned, unplaced, index,
    placement}`` dict. ``members`` is the SAME seed list passed to the walker (it
    does not echo its seeds), each an org dict with a ``spine_ref`` graph name and
    optional ``id``/``org_id``. A member with a non-empty ``spine_ref`` is a seed;
    an empty ``spine_ref`` is an ``invalid-seed``.

    ``seeds_accounted_for`` covers ONLY the structural fate of the SEEDED spines:
    it is True iff every non-empty-``spine_ref`` seed maps to its OWN distinct
    emitted home (a root chain, a standalone root, or a known Unassigned bucket) —
    i.e. no seed is silently ``dropped`` (walker resolved no apex → emitted no node)
    and no two seeds ``collided`` onto ONE emitted uid. Because the walker emits
    exactly one node per spine graph (``spine.find_apex_id`` yields one apex per
    graph and the org↔graph map keeps only the FIRST seed), two members sharing one
    ``spine_ref`` collapse onto a single home; the later seed(s) are a silent loss
    and are surfaced as ``collided`` (counted in ``counts["collided"]``, and each
    colliding org marked in ``status`` under a ``"<home-uid>::collided::<org_id>"``
    key so the lost org is visible). ``seeds_accounted_for`` intentionally does NOT
    gate on ``unplaced_notes`` — that is a separate advisory count of in-scope base
    notes claimed by no spine, and the dashboard route passes ``scope_uids=None`` so
    the walker omits the Unplaced bucket and this count is always 0 there.

    ``complete`` is a DEPRECATED alias of ``seeds_accounted_for`` (kept, with the
    identical value, for one release so existing readers keep working). It always
    over-promised — it means "every seed is accounted for", NOT "the graph is
    healthy" — so it is renamed to ``seeds_accounted_for``; migrate readers to the
    new key and stop consuming ``complete``.

    There is intentionally NO ``cycles_incomplete`` flag. An earlier draft shipped
    one to flag a supposed root-attached-cycle BLIND SPOT (a cross-graph cycle whose
    entry is a parentless root, rendered as a broken child chain rather than an
    Unassigned(cycle) bucket), computed by following ``index[...]["parent"]`` up from
    each placed spine and reporting a loop. That guard was a false-negative dead
    guard: real roots carry ``parent=None`` and :func:`detect_nested_spines` renders
    ``children`` acyclically, so a placed node's canonical parent chain ALWAYS
    terminates at a parentless root — a genuine cross-graph cycle instead lands in
    ``unassigned`` and is reported in ``cycles``. The only input that ever tripped
    the flag True was a walker-impossible hand-built composition. Computing it
    CORRECTLY would require the raw ``component-of`` parent relations, which this
    layer does not have (it is pure over the walker output and does no I/O) — and
    recovering them would mean changing the walker's traversal or return shape, the
    byte-identical isolation seam (§11) this function must not touch. Rather than
    ship a guard that is False on every real composition, the key was removed;
    genuine cross-graph cycles remain fully covered by ``cycles``.

    This is a PURE, deterministic rollup ON TOP of the walker output (§11: the
    walker is the isolation seam, never modified here); it loads no graphs and does
    no I/O.

    Returns ``{total_seeded, counts, orphans, cycles, unplaced_notes,
    seeds_accounted_for, complete, status}`` — see the module design doc for field
    semantics.
    """
    roots = composition.get("roots") or []
    unassigned = composition.get("unassigned") or []
    unplaced = composition.get("unplaced")

    # Per-spine status over every emitted node (flattened roots + unassigned
    # subtrees). A top-level root with no children is a degenerate standalone
    # root; every other in-tree node is placed; bucket subtrees inherit the
    # bucket's reason.
    status: "dict[str, str]" = {}
    emitted_nodes: "dict[str, dict[str, Any]]" = {}

    for root in roots:
        subtree = _nested_spine_nodes_in(root)
        standalone = root.get("kind") == "spine" and not (root.get("children") or [])
        for node in subtree:
            emitted_nodes[node["uid"]] = node
            if node["uid"] == root.get("uid") and standalone:
                status[node["uid"]] = "standalone-root"
            else:
                status[node["uid"]] = "placed-in-tree"

    for bucket in unassigned:
        reason = bucket.get("reason")
        label = (
            "unassigned-missing-parent"
            if reason == "missing-parent"
            else "unassigned-cycle"
        )
        for node in _nested_spine_nodes_in(bucket):
            emitted_nodes[node["uid"]] = node
            status[node["uid"]] = label

    # One emitted apex per spine graph, so graph ↔ uid is 1:1 — the match key
    # between a seed (its ``spine_ref``) and its emitted node (its ``graph``).
    emitted_by_graph: "dict[str, str]" = {}
    for uid, node in emitted_nodes.items():
        graph = str(node.get("graph") or "")
        if graph:
            emitted_by_graph.setdefault(graph, uid)

    # Reconcile each seed against its DISTINCT emitted home. The walker emits at
    # most one node per graph, so ``emitted_by_graph`` is the seed→home lookup:
    #   • no home           → the apex was unresolvable, so NO node was emitted
    #                          (a genuine silent loss) → ``dropped``.
    #   • home already taken → an earlier seed already claimed this emitted uid,
    #                          i.e. two members share one ``spine_ref`` and the
    #                          walker collapsed them → the later seed(s) are a
    #                          silent loss → ``collided`` (its org marked in status).
    total_seeded = 0
    invalid_seed = 0
    dropped = 0
    collided = 0
    home_claimed_by: "dict[str, str]" = {}  # emitted uid → the org_id that homed there
    for org in members:
        graph = str(org.get("spine_ref") or "").strip()
        if not graph:
            invalid_seed += 1
            continue
        total_seeded += 1
        org_id = str(org.get("org_id") or org.get("id") or "")
        home = emitted_by_graph.get(graph)
        if home is None:
            dropped += 1
            continue
        if home in home_claimed_by:
            collided += 1
            # Key by the colliding org so which seed lost its home is visible;
            # fall back to the emitted graph when no org_id is available.
            status[f"{home}::collided::{org_id or graph}"] = "collided"
        else:
            home_claimed_by[home] = org_id

    counts = {
        "placed-in-tree": sum(1 for v in status.values() if v == "placed-in-tree"),
        "standalone-root": sum(1 for v in status.values() if v == "standalone-root"),
        "unassigned-missing-parent": sum(
            1 for v in status.values() if v == "unassigned-missing-parent"
        ),
        "unassigned-cycle": sum(1 for v in status.values() if v == "unassigned-cycle"),
        "dropped": dropped,
        "collided": collided,
        "invalid-seed": invalid_seed,
    }

    orphans: "list[dict[str, Any]]" = []
    cycles: "list[dict[str, Any]]" = []
    for bucket in unassigned:
        kids = bucket.get("children") or []
        if not kids:
            continue
        top = kids[0]
        if bucket.get("reason") == "missing-parent":
            orphans.append(
                {
                    "uid": top.get("uid"),
                    "graph": str(top.get("graph") or ""),
                    "org_id": str(top.get("org_id") or ""),
                    "missing_parent": bucket.get("missing_parent"),
                }
            )
        elif bucket.get("reason") == "cycle":
            cycles.append(
                {
                    "entry_uid": top.get("uid"),
                    "root_uid": bucket.get("uid"),
                    "spine_uids": sorted(
                        n["uid"] for n in _nested_spine_nodes_in(bucket)
                    ),
                }
            )
    orphans.sort(key=lambda o: str(o.get("uid") or ""))
    cycles.sort(key=lambda c: str(c.get("entry_uid") or ""))

    unplaced_notes = int(unplaced.get("count") or 0) if unplaced else 0

    seeds_accounted_for = (
        counts["dropped"] == 0
        and counts["collided"] == 0
        and set(emitted_nodes) <= set(status)
    )

    return {
        "total_seeded": total_seeded,
        "counts": counts,
        "orphans": orphans,
        "cycles": cycles,
        "unplaced_notes": unplaced_notes,
        "seeds_accounted_for": seeds_accounted_for,
        # Deprecated alias of ``seeds_accounted_for`` (same value); see docstring.
        "complete": seeds_accounted_for,
        "status": status,
    }

propagate_stale_rollup

propagate_stale_rollup(composition: 'dict[str, Any]') -> None

Roll each node's per-spine stale flag UP its nested-spine subtree.

Phase-5 drift (design §10: "per-row staleness propagates up the chain"). A post-order DFS over roots + unassigned subtrees sets, for every node, node["stale_rollup"] = bool(node.get("stale") or any(child stale_rollup)). A bucket (kind == "unassigned") has no org so its own stale defaults False, but it still rolls up its children. PURE of I/O and in-place: assumes the caller has already set each spine node's stale (an advisory, cost-free drift peek), so this is deterministic and unit-testable on a plain dict.

Source code in zettelkasten/tables_nested_spines.py

def propagate_stale_rollup(composition: "dict[str, Any]") -> None:
    """Roll each node's per-spine ``stale`` flag UP its nested-spine subtree.

    Phase-5 drift (design §10: "per-row staleness propagates up the chain"). A
    post-order DFS over ``roots`` + ``unassigned`` subtrees sets, for every node,
    ``node["stale_rollup"] = bool(node.get("stale") or any(child stale_rollup))``.
    A bucket (``kind == "unassigned"``) has no org so its own ``stale`` defaults
    False, but it still rolls up its children. PURE of I/O and in-place: assumes
    the caller has already set each spine node's ``stale`` (an advisory, cost-free
    drift peek), so this is deterministic and unit-testable on a plain dict.
    """

    def _rec(node: "dict[str, Any]") -> bool:
        child_rollup = False
        for child in node.get("children") or []:
            child_rollup = _rec(child) or child_rollup
        rollup = bool(node.get("stale") or child_rollup)
        node["stale_rollup"] = rollup
        return rollup

    for root in composition.get("roots") or []:
        _rec(root)
    for bucket in composition.get("unassigned") or []:
        _rec(bucket)