zettelkasten.spine

zettelkasten.spine

Shared spine materializer — the single source of truth for spine scaffolds.

A spine is a materialized synthesis structure living in a dedicated graph: an apex node for the structure itself, one dimension node per column/facet rolling up into the apex, optional hub nodes per row entity, and member attach edges wiring the corpus notes into that structure.

Three independent code paths used to stamp (or read) this scaffold with hard-coded tag/relation conventions and would silently drift:

• coordinator.extraction.prep_spine — builds an empty spine skeleton from a schema's synthesis block for a grounded-extraction run.
• zettelkasten.organizations.promote_organization — materializes an existing matrix lens (cols + rows) and bulk-attaches its current members.
• zettelkasten.server.discover_spines — reads spines back by those exact tag conventions.

This module factors the conventions (APEX_TAGS / DIMENSION_TAG / HUB_TAGS / SPEC_TAG) and the node/edge stamping (build_spine_skeleton) into one place so the three can never drift. The graph mutations themselves are abstracted behind a tiny ops protocol with two backends:

• :class:ServerSpineGraphOps — writes through the zettelkasten MCP server primitives (used by prep_spine so it keeps its warm-cache behavior).
• :class:SpineGraphOps — writes directly through :class:ZettelGraph bound to a graphs_dir (used by promotion + the dashboard + tests), so the whole flow is exercisable against a synthetic tree with no server process.

SpineError

Bases: Exception

Raised when a spine scaffold cannot be materialized (e.g. graph create).

Source code in zettelkasten/spine.py

class SpineError(Exception):
    """Raised when a spine scaffold cannot be materialized (e.g. graph create)."""

ServerSpineGraphOps

Materializer primitives backed by the zettelkasten MCP server functions.

Used by prep_spine so it keeps writing through the same server primitives (and the warm graph cache) it always has. The server functions are imported lazily on each call so test monkeypatches on zettelkasten.server are honored.

Source code in zettelkasten/spine.py

class ServerSpineGraphOps:
    """Materializer primitives backed by the zettelkasten MCP server functions.

    Used by ``prep_spine`` so it keeps writing through the same server
    primitives (and the warm graph cache) it always has. The server functions are
    imported lazily on each call so test monkeypatches on ``zettelkasten.server``
    are honored.
    """

    def ensure_graph(self, name: str, description: str = "") -> None:
        from zettelkasten.server import create_graph

        res = _as_dict(create_graph(name=name, description=description))
        if "error" in res and res.get("type") != "AlreadyExists":
            raise SpineError(
                f"create_graph failed for synthesis graph '{name}': {res.get('error')}"
            )

    def find_id(self, graph: str, title: str) -> str:
        from zettelkasten.server import find_by_title

        res = _as_dict(find_by_title(graph, title))
        return str(res.get("id") or res.get("note_id") or "")

    def ensure_node(
        self, graph: str, title: str, ntype: str, body: str, tags, *, synthesis_status: str = ""
    ) -> str:
        nid = self.find_id(graph, title)
        if nid:
            return nid
        from zettelkasten.server import add_note

        res = _as_dict(add_note(
            graph=graph, title=title, type=ntype, body=body, tags=list(tags),
            synthesis_status=synthesis_status,
        ))
        return str(res.get("id") or res.get("note_id") or "")

    def ensure_structure_node(
        self,
        graph: str,
        node_id: str,
        title: str,
        ntype: str,
        body: str,
        tags,
        *,
        synthesis_status: str = "",
        spine_title: str = "",
    ) -> str:
        """Create/find a structure (tree) node — server backend.

        The server ``add_note`` primitive cannot stamp an arbitrary ``source``, so
        a structure node materialized through this backend keeps the legacy
        find-by-title identity (``node_id`` is accepted for a uniform
        :func:`build_spine_skeleton` contract but not persisted). That is correct for the
        extraction path: a grounded-extraction spine carries no overlay corrections,
        so the durable-id payoff (relabel survival of corrections) is unneeded there.
        ``spine_title`` is likewise accepted for signature parity with
        :meth:`SpineGraphOps.ensure_structure_node` (the legacy-composite tier-2
        fallback) but unused here, as this backend does not persist durable ids.
        """
        return self.ensure_node(
            graph, title, ntype, body, tags, synthesis_status=synthesis_status
        )

    def ensure_link(
        self,
        graph: str,
        source_id: str,
        target_id: str,
        relation: str,
        *,
        target_graph: str = "",
        confidence: "float | None" = None,
        provenance: str = "",
        verified: bool = False,
        primary: bool = False,
    ) -> bool:
        from zettelkasten.server import link_notes

        # The server ``link_notes`` primitive cannot stamp the membership
        # ``confidence``/``provenance``/``verified`` metadata, so those are accepted
        # for a uniform ops contract but dropped on this backend. That is correct
        # for the callers that use it (``prep_spine``'s empty scaffold carries no
        # members); the member-bearing promote path uses ``SpineGraphOps``. The
        # ``primary`` flag IS stamped (``link_notes`` supports it) so the build-time
        # sub-spine seed writes a durable primary parent edge through this backend.
        #
        # link_notes is idempotent (it refuses a duplicate with AlreadyExists), so
        # a re-run is a no-op; we don't need to pre-check.
        #
        # ``primary`` is only forwarded when set: the overwhelming majority of
        # edges (every structural / membership wiring) carry primary=False, so the
        # call stays byte-identical to the historical signature for them (and any
        # test double stubbing ``link_notes`` without a ``primary`` param keeps
        # working). Only a genuine sub-spine seed passes it.
        extra = {"primary": True} if primary else {}
        link_notes(
            graph=graph,
            source_id=source_id,
            target_id=target_id,
            relation=relation,
            target_graph=target_graph,
            **extra,
        )
        return True

    def node_links(self, graph: str, node_id: str) -> "list[Link]":
        """The outgoing links of ``node_id`` in ``graph`` (empty if absent).

        Reads through the server's warm graph cache so it observes edges this
        backend just wrote (used by the sub-spine seed's stability check).
        """
        from zettelkasten.server import _get_graph

        try:
            zg = _get_graph(graph)
        except Exception:  # noqa: BLE001 — a missing graph just means "no links"
            return []
        note = zg.notes.get(node_id)
        return list(note.links or []) if note is not None else []

    def remove_link(
        self,
        graph: str,
        source_id: str,
        target_id: str,
        relation: str,
        *,
        target_graph: str = "",
    ) -> bool:
        """Drop a specific outgoing ``source --relation--> target`` edge — server backend.

        Mirrors :meth:`SpineGraphOps.remove_link` (the direct-IO backend) so nested
        connect/disconnect is SYMMETRIC across both backends: the sub-spine
        disconnect / delete-reparent path works no matter which ops backend it runs
        through. Matches on the SAME ``(target, relation, target_graph)`` tuple that
        :meth:`ensure_link` dedupes on, so it removes exactly the edge a prior ensure
        created and never over-removes a sibling edge sharing the source+relation but
        pointing at a different ``(target, graph)``. Returns True iff an edge was
        removed. The write goes through the server's warm graph cache (the same
        handle :meth:`node_links` / :meth:`ensure_apex_node` read/write through) so a
        removal this backend makes is immediately observed by subsequent reads.
        """
        from zettelkasten.server import _get_graph

        try:
            zg = _get_graph(graph)
        except Exception:  # noqa: BLE001 — a missing graph has nothing to remove
            return False
        note = zg.notes.get(source_id)
        if note is None:
            return False
        before = note.links or []
        kept = [
            link
            for link in before
            if not (
                link.target == target_id
                and link.relation == relation
                and (link.graph or "") == (target_graph or "")
            )
        ]
        if len(kept) == len(before):
            return False
        note.links = kept
        zg.save_note(note, lock_key=f"box::{graph}", graphs_dir=zg.path.parent)
        return True

    def apex_node_ids(self, graph: str) -> list[str]:
        """Ids of every ``APEX_TAGS`` node in ``graph`` (the single-apex guard).

        Reads through the server's warm graph cache so it observes an apex this
        backend just created. A well-formed spine has exactly one.
        """
        from zettelkasten.server import _get_graph

        try:
            zg = _get_graph(graph)
        except Exception:  # noqa: BLE001 — a missing graph just means "no apex"
            return []
        apex = set(APEX_TAGS)
        return [n.id for n in zg.notes.values() if apex <= set(n.tags or [])]

    def find_node_by_stable_id(self, graph: str, node_id: str) -> str:
        """Id of the node carrying ``node_id`` in ``source`` (server backend), or "".

        Reads through the server's warm graph cache so it observes a node this
        backend just created / stamped. Mirrors
        :meth:`SpineGraphOps.find_node_by_stable_id` — used to resolve the durable
        apex sentinel on the server-backed build path.
        """
        if not node_id:
            return ""
        from zettelkasten.server import _get_graph

        try:
            zg = _get_graph(graph)
        except Exception:  # noqa: BLE001 — a missing graph just means "no match"
            return ""
        for note in zg.notes.values():
            src = note.source if isinstance(note.source, dict) else {}
            if str(src.get(STRUCTURE_NODE_ID_KEY) or "") == node_id:
                return note.id
        return ""

    def ensure_apex_node(
        self,
        graph: str,
        node_id: str,
        title: str,
        ntype: str,
        body: str,
        tags,
        *,
        synthesis_status: str = "",
    ) -> str:
        """Create/find the spine APEX keyed by the durable ``node_id`` — server backend.

        The server ``add_note`` primitive cannot stamp an arbitrary ``source``, so
        the durable apex sentinel (``APEX_NODE_ID``) is applied by a follow-up write
        through the warm graph cache. This mirrors
        :meth:`SpineGraphOps.ensure_apex_node`'s three-tier identity so a build-path
        (:func:`coordinator.extraction.prep_spine`) title drift re-titles the SAME
        apex node instead of minting a SECOND ``APEX_TAGS`` node — which would both
        trip the exactly-one-apex guard AND orphan the primary sub-spine edge stored
        on the original apex. Tiers:

        1. Find by the durable ``source['node_id']`` (the apex sentinel); if the
           title drifted, re-title the SAME node (its id + every edge preserved).
        2. Else find the existing ``APEX_TAGS`` apex and STAMP the sentinel onto it
           in place (the one-time forward migration for a legacy title-keyed apex);
           never clobber a foreign durable id.
        3. Else create a fresh apex (via ``add_note``) and stamp the sentinel on it.

        Passing an empty ``node_id`` degrades to plain find-by-title creation
        (:meth:`ensure_node`), matching the fallback the shared builder used before.
        """
        if not node_id:
            return self.ensure_node(
                graph, title, ntype, body, tags, synthesis_status=synthesis_status
            )

        def _warm():
            from zettelkasten.server import _get_graph

            try:
                return _get_graph(graph)
            except Exception:  # noqa: BLE001 — treat a missing graph as "no node"
                return None

        # Tier 1: durable-id hit — re-title the SAME node if the title drifted.
        existing_id = self.find_node_by_stable_id(graph, node_id)
        if existing_id:
            zg = _warm()
            if zg is not None:
                note = zg.notes.get(existing_id)
                if note is not None and title and (note.title or "") != title:
                    note.title = title
                    zg.save_note(note, lock_key=f"box::{graph}", graphs_dir=zg.path.parent)
            return existing_id
        # Tier 2: migrate a legacy title-keyed apex in place (stamp the sentinel).
        zg = _warm()
        if zg is not None:
            apex = set(APEX_TAGS)
            for note in zg.notes.values():
                if not (apex <= set(note.tags or [])):
                    continue
                src = note.source if isinstance(note.source, dict) else {}
                existing_node_id = str(src.get(STRUCTURE_NODE_ID_KEY) or "")
                if existing_node_id and existing_node_id != node_id:
                    # A different durable id already claims this apex — leave it
                    # for the guard to surface rather than rewriting identity.
                    continue
                changed = False
                if existing_node_id != node_id:
                    src = dict(src)
                    src[STRUCTURE_NODE_ID_KEY] = node_id
                    note.source = src
                    changed = True
                if title and (note.title or "") != title:
                    note.title = title
                    changed = True
                if changed:
                    zg.save_note(note, lock_key=f"box::{graph}", graphs_dir=zg.path.parent)
                return note.id
        # Tier 3: no durable-id or legacy apex node exists — create a FRESH apex
        # and stamp the sentinel. Mirrors :meth:`SpineGraphOps.ensure_apex_node`,
        # which always mints a fresh node when ``node_id`` is set (never returns a
        # find-by-title collision): with tiers 1+2 having proven no ``APEX_TAGS``
        # node exists, any same-title hit here is a FOREIGN node, so we must not
        # stamp the apex sentinel onto it. ``add_note`` generates a unique id.
        from zettelkasten.server import add_note

        res = _as_dict(add_note(
            graph=graph, title=title, type=ntype, body=body, tags=list(tags),
            synthesis_status=synthesis_status,
        ))
        nid = str(res.get("id") or res.get("note_id") or "")
        if nid:
            zg = _warm()
            if zg is not None:
                note = zg.notes.get(nid)
                if note is not None:
                    src = note.source if isinstance(note.source, dict) else {}
                    if str(src.get(STRUCTURE_NODE_ID_KEY) or "") != node_id:
                        src = dict(src)
                        src[STRUCTURE_NODE_ID_KEY] = node_id
                        note.source = src
                        zg.save_note(note, lock_key=f"box::{graph}", graphs_dir=zg.path.parent)
        return nid

ensure_structure_node

ensure_structure_node(graph: str, node_id: str, title: str, ntype: str, body: str, tags, *, synthesis_status: str = '', spine_title: str = '') -> str

Create/find a structure (tree) node — server backend.

The server add_note primitive cannot stamp an arbitrary source, so a structure node materialized through this backend keeps the legacy find-by-title identity (node_id is accepted for a uniform :func:build_spine_skeleton contract but not persisted). That is correct for the extraction path: a grounded-extraction spine carries no overlay corrections, so the durable-id payoff (relabel survival of corrections) is unneeded there. spine_title is likewise accepted for signature parity with :meth:SpineGraphOps.ensure_structure_node (the legacy-composite tier-2 fallback) but unused here, as this backend does not persist durable ids.

Source code in zettelkasten/spine.py

def ensure_structure_node(
    self,
    graph: str,
    node_id: str,
    title: str,
    ntype: str,
    body: str,
    tags,
    *,
    synthesis_status: str = "",
    spine_title: str = "",
) -> str:
    """Create/find a structure (tree) node — server backend.

    The server ``add_note`` primitive cannot stamp an arbitrary ``source``, so
    a structure node materialized through this backend keeps the legacy
    find-by-title identity (``node_id`` is accepted for a uniform
    :func:`build_spine_skeleton` contract but not persisted). That is correct for the
    extraction path: a grounded-extraction spine carries no overlay corrections,
    so the durable-id payoff (relabel survival of corrections) is unneeded there.
    ``spine_title`` is likewise accepted for signature parity with
    :meth:`SpineGraphOps.ensure_structure_node` (the legacy-composite tier-2
    fallback) but unused here, as this backend does not persist durable ids.
    """
    return self.ensure_node(
        graph, title, ntype, body, tags, synthesis_status=synthesis_status
    )

node_links

node_links(graph: str, node_id: str) -> 'list[Link]'

The outgoing links of node_id in graph (empty if absent).

Reads through the server's warm graph cache so it observes edges this backend just wrote (used by the sub-spine seed's stability check).

Source code in zettelkasten/spine.py

def node_links(self, graph: str, node_id: str) -> "list[Link]":
    """The outgoing links of ``node_id`` in ``graph`` (empty if absent).

    Reads through the server's warm graph cache so it observes edges this
    backend just wrote (used by the sub-spine seed's stability check).
    """
    from zettelkasten.server import _get_graph

    try:
        zg = _get_graph(graph)
    except Exception:  # noqa: BLE001 — a missing graph just means "no links"
        return []
    note = zg.notes.get(node_id)
    return list(note.links or []) if note is not None else []

remove_link

remove_link(graph: str, source_id: str, target_id: str, relation: str, *, target_graph: str = '') -> bool

Drop a specific outgoing source --relation--> target edge — server backend.

Mirrors :meth:SpineGraphOps.remove_link (the direct-IO backend) so nested connect/disconnect is SYMMETRIC across both backends: the sub-spine disconnect / delete-reparent path works no matter which ops backend it runs through. Matches on the SAME (target, relation, target_graph) tuple that :meth:ensure_link dedupes on, so it removes exactly the edge a prior ensure created and never over-removes a sibling edge sharing the source+relation but pointing at a different (target, graph). Returns True iff an edge was removed. The write goes through the server's warm graph cache (the same handle :meth:node_links / :meth:ensure_apex_node read/write through) so a removal this backend makes is immediately observed by subsequent reads.

Source code in zettelkasten/spine.py

def remove_link(
    self,
    graph: str,
    source_id: str,
    target_id: str,
    relation: str,
    *,
    target_graph: str = "",
) -> bool:
    """Drop a specific outgoing ``source --relation--> target`` edge — server backend.

    Mirrors :meth:`SpineGraphOps.remove_link` (the direct-IO backend) so nested
    connect/disconnect is SYMMETRIC across both backends: the sub-spine
    disconnect / delete-reparent path works no matter which ops backend it runs
    through. Matches on the SAME ``(target, relation, target_graph)`` tuple that
    :meth:`ensure_link` dedupes on, so it removes exactly the edge a prior ensure
    created and never over-removes a sibling edge sharing the source+relation but
    pointing at a different ``(target, graph)``. Returns True iff an edge was
    removed. The write goes through the server's warm graph cache (the same
    handle :meth:`node_links` / :meth:`ensure_apex_node` read/write through) so a
    removal this backend makes is immediately observed by subsequent reads.
    """
    from zettelkasten.server import _get_graph

    try:
        zg = _get_graph(graph)
    except Exception:  # noqa: BLE001 — a missing graph has nothing to remove
        return False
    note = zg.notes.get(source_id)
    if note is None:
        return False
    before = note.links or []
    kept = [
        link
        for link in before
        if not (
            link.target == target_id
            and link.relation == relation
            and (link.graph or "") == (target_graph or "")
        )
    ]
    if len(kept) == len(before):
        return False
    note.links = kept
    zg.save_note(note, lock_key=f"box::{graph}", graphs_dir=zg.path.parent)
    return True

apex_node_ids

apex_node_ids(graph: str) -> list[str]

Ids of every APEX_TAGS node in graph (the single-apex guard).

Reads through the server's warm graph cache so it observes an apex this backend just created. A well-formed spine has exactly one.

Source code in zettelkasten/spine.py

def apex_node_ids(self, graph: str) -> list[str]:
    """Ids of every ``APEX_TAGS`` node in ``graph`` (the single-apex guard).

    Reads through the server's warm graph cache so it observes an apex this
    backend just created. A well-formed spine has exactly one.
    """
    from zettelkasten.server import _get_graph

    try:
        zg = _get_graph(graph)
    except Exception:  # noqa: BLE001 — a missing graph just means "no apex"
        return []
    apex = set(APEX_TAGS)
    return [n.id for n in zg.notes.values() if apex <= set(n.tags or [])]

find_node_by_stable_id

find_node_by_stable_id(graph: str, node_id: str) -> str

Id of the node carrying node_id in source (server backend), or "".

Reads through the server's warm graph cache so it observes a node this backend just created / stamped. Mirrors :meth:SpineGraphOps.find_node_by_stable_id — used to resolve the durable apex sentinel on the server-backed build path.

Source code in zettelkasten/spine.py

def find_node_by_stable_id(self, graph: str, node_id: str) -> str:
    """Id of the node carrying ``node_id`` in ``source`` (server backend), or "".

    Reads through the server's warm graph cache so it observes a node this
    backend just created / stamped. Mirrors
    :meth:`SpineGraphOps.find_node_by_stable_id` — used to resolve the durable
    apex sentinel on the server-backed build path.
    """
    if not node_id:
        return ""
    from zettelkasten.server import _get_graph

    try:
        zg = _get_graph(graph)
    except Exception:  # noqa: BLE001 — a missing graph just means "no match"
        return ""
    for note in zg.notes.values():
        src = note.source if isinstance(note.source, dict) else {}
        if str(src.get(STRUCTURE_NODE_ID_KEY) or "") == node_id:
            return note.id
    return ""

ensure_apex_node

ensure_apex_node(graph: str, node_id: str, title: str, ntype: str, body: str, tags, *, synthesis_status: str = '') -> str

Create/find the spine APEX keyed by the durable node_id — server backend.

The server add_note primitive cannot stamp an arbitrary source, so the durable apex sentinel (APEX_NODE_ID) is applied by a follow-up write through the warm graph cache. This mirrors :meth:SpineGraphOps.ensure_apex_node's three-tier identity so a build-path (:func:coordinator.extraction.prep_spine) title drift re-titles the SAME apex node instead of minting a SECOND APEX_TAGS node — which would both trip the exactly-one-apex guard AND orphan the primary sub-spine edge stored on the original apex. Tiers:

1. Find by the durable source['node_id'] (the apex sentinel); if the title drifted, re-title the SAME node (its id + every edge preserved).
2. Else find the existing APEX_TAGS apex and STAMP the sentinel onto it in place (the one-time forward migration for a legacy title-keyed apex); never clobber a foreign durable id.
3. Else create a fresh apex (via add_note) and stamp the sentinel on it.

Passing an empty node_id degrades to plain find-by-title creation (:meth:ensure_node), matching the fallback the shared builder used before.

Source code in zettelkasten/spine.py

def ensure_apex_node(
    self,
    graph: str,
    node_id: str,
    title: str,
    ntype: str,
    body: str,
    tags,
    *,
    synthesis_status: str = "",
) -> str:
    """Create/find the spine APEX keyed by the durable ``node_id`` — server backend.

    The server ``add_note`` primitive cannot stamp an arbitrary ``source``, so
    the durable apex sentinel (``APEX_NODE_ID``) is applied by a follow-up write
    through the warm graph cache. This mirrors
    :meth:`SpineGraphOps.ensure_apex_node`'s three-tier identity so a build-path
    (:func:`coordinator.extraction.prep_spine`) title drift re-titles the SAME
    apex node instead of minting a SECOND ``APEX_TAGS`` node — which would both
    trip the exactly-one-apex guard AND orphan the primary sub-spine edge stored
    on the original apex. Tiers:

    1. Find by the durable ``source['node_id']`` (the apex sentinel); if the
       title drifted, re-title the SAME node (its id + every edge preserved).
    2. Else find the existing ``APEX_TAGS`` apex and STAMP the sentinel onto it
       in place (the one-time forward migration for a legacy title-keyed apex);
       never clobber a foreign durable id.
    3. Else create a fresh apex (via ``add_note``) and stamp the sentinel on it.

    Passing an empty ``node_id`` degrades to plain find-by-title creation
    (:meth:`ensure_node`), matching the fallback the shared builder used before.
    """
    if not node_id:
        return self.ensure_node(
            graph, title, ntype, body, tags, synthesis_status=synthesis_status
        )

    def _warm():
        from zettelkasten.server import _get_graph

        try:
            return _get_graph(graph)
        except Exception:  # noqa: BLE001 — treat a missing graph as "no node"
            return None

    # Tier 1: durable-id hit — re-title the SAME node if the title drifted.
    existing_id = self.find_node_by_stable_id(graph, node_id)
    if existing_id:
        zg = _warm()
        if zg is not None:
            note = zg.notes.get(existing_id)
            if note is not None and title and (note.title or "") != title:
                note.title = title
                zg.save_note(note, lock_key=f"box::{graph}", graphs_dir=zg.path.parent)
        return existing_id
    # Tier 2: migrate a legacy title-keyed apex in place (stamp the sentinel).
    zg = _warm()
    if zg is not None:
        apex = set(APEX_TAGS)
        for note in zg.notes.values():
            if not (apex <= set(note.tags or [])):
                continue
            src = note.source if isinstance(note.source, dict) else {}
            existing_node_id = str(src.get(STRUCTURE_NODE_ID_KEY) or "")
            if existing_node_id and existing_node_id != node_id:
                # A different durable id already claims this apex — leave it
                # for the guard to surface rather than rewriting identity.
                continue
            changed = False
            if existing_node_id != node_id:
                src = dict(src)
                src[STRUCTURE_NODE_ID_KEY] = node_id
                note.source = src
                changed = True
            if title and (note.title or "") != title:
                note.title = title
                changed = True
            if changed:
                zg.save_note(note, lock_key=f"box::{graph}", graphs_dir=zg.path.parent)
            return note.id
    # Tier 3: no durable-id or legacy apex node exists — create a FRESH apex
    # and stamp the sentinel. Mirrors :meth:`SpineGraphOps.ensure_apex_node`,
    # which always mints a fresh node when ``node_id`` is set (never returns a
    # find-by-title collision): with tiers 1+2 having proven no ``APEX_TAGS``
    # node exists, any same-title hit here is a FOREIGN node, so we must not
    # stamp the apex sentinel onto it. ``add_note`` generates a unique id.
    from zettelkasten.server import add_note

    res = _as_dict(add_note(
        graph=graph, title=title, type=ntype, body=body, tags=list(tags),
        synthesis_status=synthesis_status,
    ))
    nid = str(res.get("id") or res.get("note_id") or "")
    if nid:
        zg = _warm()
        if zg is not None:
            note = zg.notes.get(nid)
            if note is not None:
                src = note.source if isinstance(note.source, dict) else {}
                if str(src.get(STRUCTURE_NODE_ID_KEY) or "") != node_id:
                    src = dict(src)
                    src[STRUCTURE_NODE_ID_KEY] = node_id
                    note.source = src
                    zg.save_note(note, lock_key=f"box::{graph}", graphs_dir=zg.path.parent)
    return nid

SpineGraphOps

Materializer primitives backed by direct :class:ZettelGraph IO.

Graphs are loaded once via get_graph and reused within one operation, so node/edge writes stay coherent in memory and on disk. Used by promotion (the dashboard route + tests), where the whole flow operates against a single graphs_dir and needs no server process.

Source code in zettelkasten/spine.py

class SpineGraphOps:
    """Materializer primitives backed by direct :class:`ZettelGraph` IO.

    Graphs are loaded once via ``get_graph`` and reused within one operation, so
    node/edge writes stay coherent in memory and on disk. Used by promotion (the
    dashboard route + tests), where the whole flow operates against a single
    ``graphs_dir`` and needs no server process.
    """

    def __init__(
        self,
        get_graph: Callable[[str], ZettelGraph],
        *,
        graphs_dir: "Path | None" = None,
        held_box_locks: "set[str] | None" = None,
    ):
        self._get_graph = get_graph
        self._base = Path(graphs_dir) if graphs_dir is not None else GRAPHS_DIR
        self._cache: dict[str, ZettelGraph] = {}
        # Graph names whose ``box::{graph}`` cross-process write lock the CALLER
        # already holds across a WIDER read-modify-write (a lifecycle op in
        # ``organizations.py`` — promote/resync/materialize/delete). For those
        # graphs this ops object must NOT re-acquire the same lock on each write:
        # :func:`zettelkasten.commit.review_write_lock` is NON-reentrant (its OS
        # advisory file layer would deadlock on a second acquire in the same
        # process), and it is unnecessary — the caller's outer lock already covers
        # every write to that graph AND makes the whole RMW lost-update-free. So
        # :meth:`_save` / :meth:`delete_node` skip the inner lock for a held graph
        # and rely on the caller's outer lock. Writes to any graph NOT in this set
        # (e.g. a member note's home graph, or a child sub-spine graph the caller
        # did not lock) still take their own ``box::{graph}`` lock, staying
        # torn-write- and cross-process-safe. INVARIANT: this set must equal
        # EXACTLY the set of ``box::{graph}`` locks the caller actually holds — so a
        # write is only ever un-locked here when the caller genuinely holds the lock.
        self._held_box_locks: set[str] = set(held_box_locks or ())

    def _graph(self, name: str) -> ZettelGraph:
        zg = self._cache.get(name)
        if zg is None:
            zg = self._get_graph(name)
            self._cache[name] = zg
        return zg

    def _save(self, graph: str, note: Note) -> None:
        """Persist ``note`` into ``graph``, serialized on its ``box::{graph}`` lock
        UNLESS the caller already holds that lock.

        Every SpineGraphOps writer touches SHARED spine nodes (apex / dimension /
        hub) that a SEPARATE process — the zettelkasten MCP server, which serializes
        its own note writes on ``box::{graph}`` via
        :func:`zettelkasten.commit.box_write_lock` — can write concurrently.

        * ``graph`` NOT in ``held_box_locks``: pass ``lock_key=box::{graph}`` so
          this per-note write window serializes cross-process against the MCP server
          (same key + ``graphs_dir``), so neither torn-writes nor diverges its
          in-memory index. This alone guarantees TORN-WRITE safety for the write,
          not no-lost-update across a find-then-modify (the caller owns that).
        * ``graph`` IN ``held_box_locks``: pass ``lock_key=None``. A lifecycle op in
          ``organizations.py`` already holds ``box::{graph}`` across its ENTIRE
          read-modify-write, so (a) re-acquiring the same NON-reentrant lock here
          would deadlock on its OS advisory file layer, and (b) it is unnecessary —
          the outer lock covers this write AND makes the whole RMW lost-update-free
          (no in-between window a concurrent process could interleave a stale
          read/write into). The lock still wraps only the brief per-note write
          window (see :meth:`ZettelGraph.save_note`), never a wider build.
        """
        lock_key = None if graph in self._held_box_locks else f"box::{graph}"
        self._graph(graph).save_note(
            note, lock_key=lock_key, graphs_dir=self._base
        )

    def ensure_graph(self, name: str, description: str = "") -> None:
        path = self._base / name
        meta = path / "_meta.yaml"
        if not meta.exists():
            import yaml

            path.mkdir(parents=True, exist_ok=True)
            meta.write_text(
                yaml.dump(
                    {
                        "name": name,
                        "description": description,
                        "source": "",
                        "coverage": {"human": "none", "agent": "none"},
                    },
                    default_flow_style=False,
                ),
                encoding="utf-8",
            )
            # Drop any stale cached handle so the freshly-created graph loads.
            self._cache.pop(name, None)
        # Warm the (possibly empty) graph into the per-op cache.
        self._graph(name)

    def find_id(self, graph: str, title: str) -> str:
        note = self._graph(graph).find_by_title(title)
        return note.id if note else ""

    def ensure_node(
        self, graph: str, title: str, ntype: str, body: str, tags, *, synthesis_status: str = ""
    ) -> str:
        zg = self._graph(graph)
        existing = zg.find_by_title(title)
        if existing is not None:
            return existing.id
        nid = zg.generate_unique_id(title)
        self._save(
            graph,
            Note(
                id=nid, title=title, type=ntype, source={}, body=body,
                tags=list(tags), synthesis_status=synthesis_status,
            ),
        )
        return nid

    def find_node_by_stable_id(self, graph: str, node_id: str) -> str:
        """The id of the structure node carrying ``node_id`` in ``source``, or "".

        Structure (tree) nodes are keyed by their DURABLE ``source['node_id']`` (§11.3
        of the spine-promotion design) so a relabel re-titles the SAME node instead
        of orphaning it — the structural analogue of :meth:`find_hub_by_row` for
        row hubs.
        """
        if not node_id:
            return ""
        zg = self._graph(graph)
        for note in zg.notes.values():
            src = note.source if isinstance(note.source, dict) else {}
            if str(src.get(STRUCTURE_NODE_ID_KEY) or "") == node_id:
                return note.id
        return ""

    def ensure_structure_node(
        self,
        graph: str,
        node_id: str,
        title: str,
        ntype: str,
        body: str,
        tags,
        *,
        synthesis_status: str = "",
        spine_title: str = "",
    ) -> str:
        """Idempotently create/find a structure (tree) node keyed by stable ``node_id``.

        Identity is resolved in three tiers so the durable id survives a relabel
        AND a legacy (node-id-less) spine migrates forward on its next promote:

        1. Find by the stable ``source['node_id']``. If the title drifted (a column
           relabel), re-title the SAME node — its graph id (and thus every
           ``spine-member`` edge + ``column['apex']`` reference) is preserved.
        2. Else find by title (the legacy find-by-title dedup ``build_spine_skeleton``
           always used) and STAMP ``node_id`` onto it, so a spine materialized
           before V2b acquires durable ids the first time it is re-promoted.
           Two exact title formats are tried, in order: the current BARE ``title``
           (e.g. ``"Dataset"``), then — when ``spine_title`` is supplied — the OLD
           COMPOSITE format a pre-bare-label spine minted the node under,
           ``f"{spine_title} — {title}"`` (space, em-dash, space; the exact string
           ``organizations.py`` used before dimension titles were changed to the
           bare column label). A legacy composite-title node is thus still matched
           on its first re-promote after that change; it is then re-titled IN PLACE
           to the bare ``title`` and stamped with the durable ``node_id``, so its
           ``spine-member`` edges + overlay corrections are preserved rather than
           orphaned by a fresh mint. Only these two EXACT titles are tried — no
           positional or fuzzy guessing.
           HARDENED: tier 2 migrates ONLY a GENUINE legacy DIMENSION node — one
           tagged ``DIMENSION_TAG``, with no durable id yet and no conflicting
           identity. It REFUSES to stamp ``node_id`` onto ANY non-dimension
           same-title collision: a row hub (``source['row_id']`` or a ``HUB_TAGS``
           tag), the apex or a spec (``APEX_TAGS``/``SPEC_TAG``), a plain note
           (no dimension tag), or a node already carrying a DIFFERENT ``node_id``.
           A title collision (e.g. a dimension whose label equals the apex title,
           a row-hub label, or a stray note) must never let a dimension hijack a
           foreign node's identity — doing so wired the apex ``component-of``
           itself and silently dropped a member under the apex-exclusion rollup.
           Any such foreign collision falls through to tier 3.
        3. Else create a fresh node carrying ``source['node_id']``.

        Passing an empty ``node_id`` degrades to plain find-by-title creation
        (no durable id), matching :meth:`ensure_node`. ``spine_title`` is used
        ONLY for the tier-2 legacy-composite fallback and only when a durable
        ``node_id`` is requested, so the degraded and greenfield paths are
        byte-identical to before.
        """
        zg = self._graph(graph)
        existing_id = self.find_node_by_stable_id(graph, node_id)
        if existing_id:
            note = zg.notes.get(existing_id)
            if note is not None and title and (note.title or "") != title:
                note.title = title
                self._save(graph, note)
            return existing_id
        existing = zg.find_by_title(title)
        if existing is None and node_id and spine_title:
            # Legacy-composite fallback: a dimension minted before the bare-label
            # change carries the OLD composite title and has no durable id, so the
            # bare-title lookup above misses it. Try the exact old format so it is
            # matched (and migrated below) instead of orphaned by a fresh mint.
            legacy_composite = f"{spine_title} — {title}"
            if legacy_composite != title:
                existing = zg.find_by_title(legacy_composite)
        if existing is not None:
            src = existing.source if isinstance(existing.source, dict) else {}
            existing_node_id = str(src.get(STRUCTURE_NODE_ID_KEY) or "")
            if not node_id:
                # Degraded find-by-title creation (no durable id requested).
                return existing.id
            # Only a genuine legacy DIMENSION node may be migrated in place: it
            # must be tagged ``DIMENSION_TAG`` and must NOT be a row hub, the apex,
            # a spec, or carry a conflicting durable id. Any other same-title node
            # (apex, spec, hub, plain note) is a FOREIGN collision: leave it
            # untouched and mint a fresh dimension node (tier 3) so a dimension can
            # never hijack a foreign node's identity (the apex-self-loop bug).
            existing_tags = set(existing.tags or [])
            is_dimension = DIMENSION_TAG in existing_tags
            is_hub = bool(str(src.get("row_id") or "")) or bool(
                set(HUB_TAGS) & existing_tags
            )
            is_apex_or_spec = bool(set(APEX_TAGS) & existing_tags) or (
                SPEC_TAG in existing_tags
            )
            conflicting_id = bool(existing_node_id) and existing_node_id != node_id
            if is_dimension and not is_hub and not is_apex_or_spec and not conflicting_id:
                changed = False
                if existing_node_id != node_id:
                    src = dict(src)
                    src[STRUCTURE_NODE_ID_KEY] = node_id
                    existing.source = src
                    changed = True
                # Retitle in place when matched via the legacy composite fallback,
                # so the node forward-migrates to the bare label (a bare-title
                # match already equals ``title``, so this is a no-op there).
                if title and (existing.title or "") != title:
                    existing.title = title
                    changed = True
                if changed:
                    self._save(graph, existing)
                return existing.id
            # Foreign collision: fall through to mint a fresh node below.
        nid = zg.generate_unique_id(title)
        source = {STRUCTURE_NODE_ID_KEY: node_id} if node_id else {}
        self._save(
            graph,
            Note(
                id=nid, title=title, type=ntype, source=source, body=body,
                tags=list(tags), synthesis_status=synthesis_status,
            ),
        )
        return nid

    def apex_node_ids(self, graph: str) -> list[str]:
        """Ids of every ``APEX_TAGS`` node in ``graph`` (the single-apex guard).

        A well-formed spine has exactly one; a second is the corruption
        :func:`build_spine_skeleton`'s guard fails loud on.
        """
        zg = self._graph(graph)
        apex = set(APEX_TAGS)
        return [n.id for n in zg.notes.values() if apex <= set(n.tags or [])]

    def ensure_apex_node(
        self,
        graph: str,
        node_id: str,
        title: str,
        ntype: str,
        body: str,
        tags,
        *,
        synthesis_status: str = "",
    ) -> str:
        """Idempotently create/find the spine APEX, keyed by the durable ``node_id``.

        The apex is the ROOT structure node, so — like the dimensions — it must
        survive a relabel / org-rename / ``title_template`` drift by re-titling the
        SAME node rather than minting a second ``APEX_TAGS`` node (which would
        orphan the primary sub-spine edge stored on the original apex). Identity is
        resolved in three tiers:

        1. Find by the durable ``source['node_id']`` (the apex sentinel). If the
           title drifted, re-title the SAME node — its graph id (and every edge on
           it, including the sub-spine ``component-of``) is preserved.
        2. Else find the existing apex by its ``APEX_TAGS`` (a spine has exactly one
           apex, so this is unambiguous) and STAMP the sentinel ``node_id`` onto it
           IN PLACE. This is the one-time, byte-safe forward migration for a legacy
           spine whose apex predates the durable id — it NEVER mints a duplicate.
        3. Else create a fresh apex carrying the sentinel ``source['node_id']``.

        Unlike :meth:`ensure_structure_node` (which REFUSES to migrate an
        ``APEX_TAGS`` node so a dimension can never hijack the apex's identity),
        this method is the apex's OWN builder: migrating the apex in place is
        exactly the intent. Passing an empty ``node_id`` degrades to find-by-title
        creation, matching :meth:`ensure_node`.
        """
        zg = self._graph(graph)
        existing_id = self.find_node_by_stable_id(graph, node_id)
        if existing_id:
            note = zg.notes.get(existing_id)
            if note is not None and title and (note.title or "") != title:
                note.title = title
                self._save(graph, note)
            return existing_id
        if node_id:
            # Tier 2: migrate the legacy title-keyed apex in place. Prefer an apex
            # that already carries the durable id (none here, tier 1 handled that),
            # then any apex with NO durable id, so we never clobber a foreign id.
            apex = set(APEX_TAGS)
            for note in zg.notes.values():
                if not (apex <= set(note.tags or [])):
                    continue
                src = note.source if isinstance(note.source, dict) else {}
                existing_node_id = str(src.get(STRUCTURE_NODE_ID_KEY) or "")
                if existing_node_id and existing_node_id != node_id:
                    # A different durable id already claims this apex — do not
                    # overwrite it; fall through and let the guard surface the
                    # conflict rather than silently rewriting identity.
                    continue
                changed = False
                if existing_node_id != node_id:
                    src = dict(src)
                    src[STRUCTURE_NODE_ID_KEY] = node_id
                    note.source = src
                    changed = True
                if title and (note.title or "") != title:
                    note.title = title
                    changed = True
                if changed:
                    self._save(graph, note)
                return note.id
        # Tier 3: no durable-id or legacy apex found — create a fresh one.
        existing = zg.find_by_title(title)
        if existing is not None and not node_id:
            return existing.id
        nid = zg.generate_unique_id(title)
        source = {STRUCTURE_NODE_ID_KEY: node_id} if node_id else {}
        self._save(
            graph,
            Note(
                id=nid, title=title, type=ntype, source=source, body=body,
                tags=list(tags), synthesis_status=synthesis_status,
            ),
        )
        return nid

    def find_hub_by_row(self, graph: str, row_id: str) -> str:
        """The id of the hub node carrying ``row_id`` in ``note.source``, or "".

        Hubs are keyed by the row's STABLE id (not its label) so two rows sharing
        a label never collapse into one hub and a label change re-titles the same
        hub instead of orphaning it.
        """
        zg = self._graph(graph)
        for note in zg.notes.values():
            src = note.source if isinstance(note.source, dict) else {}
            if str(src.get("row_id") or "") == row_id and row_id:
                return note.id
        return ""

    def ensure_hub(self, graph: str, row_id: str, label: str, body: str, tags=HUB_TAGS) -> str:
        """Idempotently create/update a row hub keyed by stable ``row_id``.

        Stores the original ``row_id`` on the hub's ``source`` so the link-form
        rebuild recovers it (overlay corrections key on it). Updates the title to
        ``label`` if it changed, so re-promoting after a row relabel re-titles the
        existing hub rather than orphaning it.
        """
        zg = self._graph(graph)
        existing_id = self.find_hub_by_row(graph, row_id)
        if existing_id:
            note = zg.notes.get(existing_id)
            if note is not None and (note.title or "") != label and label:
                note.title = label
                self._save(graph, note)
            return existing_id
        nid = zg.generate_unique_id(label or row_id)
        self._save(
            graph,
            Note(
                id=nid,
                title=label or row_id,
                type="concept",
                source={"row_id": row_id},
                body=body,
                tags=list(tags),
            ),
        )
        return nid

    def set_hub_basis(self, graph: str, hub_id: str, basis: "list[str]") -> None:
        """Persist a hub's MEMBER BASIS (the ``<home>::<note_id>`` uid set) on its
        ``source['member_basis']``.

        The basis is the last-known membership the overlap remap falls back to when
        a hub has zero LIVE member edges (every member was overlay-removed): without
        it a regrown semantic group could not be matched to its existing hub and
        would mint a duplicate row. Stored spine-side on the hub node, so base notes
        stay pristine.
        """
        zg = self._graph(graph)
        note = zg.notes.get(hub_id)
        if note is None:
            return
        src = note.source if isinstance(note.source, dict) else {}
        new_basis = sorted({u for u in basis if u})
        if list(src.get("member_basis") or []) == new_basis:
            return
        src = dict(src)
        src["member_basis"] = new_basis
        note.source = src
        self._save(graph, note)

    def structural_node_ids(self, graph: str, tags) -> list[str]:
        """Ids of every node in ``graph`` carrying ANY of ``tags`` (hub/dim scan)."""
        zg = self._graph(graph)
        want = set(tags)
        return [n.id for n in zg.notes.values() if want & set(n.tags or [])]

    def node_row_id(self, graph: str, node_id: str) -> str:
        """The stable ``source['row_id']`` a node carries (hubs only), or "".

        Read BEFORE :meth:`delete_node` so a caller can tell which overlay
        corrections a soon-to-be-pruned hub was carrying (orphan detection).
        """
        zg = self._graph(graph)
        note = zg.notes.get(node_id)
        if note is None:
            return ""
        src = note.source if isinstance(note.source, dict) else {}
        return str(src.get("row_id") or "")

    def delete_node(self, graph: str, node_id: str) -> bool:
        """Remove a single node (file + in-memory index) from ``graph``.

        Also scrubs any DANGLING SAME-GRAPH edge that pointed AT the deleted node
        (e.g. the structural ``apex --related--> hub`` link a hub prune would
        otherwise leave behind). A cross-graph edge whose target merely shares this
        id but lives in ANOTHER graph (a ``spine-member`` edge into a base note's
        home graph) is left intact — its target still exists in its own graph — so
        a surviving node never loses an edge it genuinely still needs.
        """
        from zettelkasten.commit import review_write_lock

        zg = self._graph(graph)
        # Take the CROSS-PROCESS ``box::{graph}`` file lock FIRST (outermost), then
        # the in-process ``_embeddings_lock`` — the SAME order ``save_note`` uses
        # (file lock → embeddings lock). delete_node mutates ``zg.notes`` directly
        # (a file unlink + index pop + an inbound-edge scrub) so it MUST hold the
        # same cross-process lock the MCP server takes on ``box::{graph}``, or a
        # concurrent same-graph writer in another process would lose an update /
        # diverge its index. The nested scrub write below therefore stays UNLOCKED
        # (plain ``save_note`` with no ``lock_key``): re-taking the file lock from
        # inside the ``_embeddings_lock`` would INVERT the lock order and risk an
        # ABBA deadlock, so the outer lock covers the whole critical section.
        #
        # When ``graph`` is in ``held_box_locks`` the caller (a lifecycle op)
        # ALREADY holds ``box::{graph}`` across its whole RMW, so we must NOT
        # re-acquire the NON-reentrant lock here (it would deadlock); a
        # ``nullcontext`` lets the caller's outer lock cover this delete instead.
        outer_lock = (
            contextlib.nullcontext()
            if graph in self._held_box_locks
            else review_write_lock(f"box::{graph}", graphs_dir=self._base)
        )
        with outer_lock:
            # Serialize every direct ``zg.notes`` mutation below under the same
            # RLock ``get_backlinks``/``_reindex_backlinks``/``save_note`` use, so
            # the ``notes.pop`` and the ``zg.notes.values()`` scrub loop cannot race
            # a concurrent get_backlinks lazy rebuild iterating ``self.notes`` (which
            # would otherwise raise "dictionary changed size during iteration" or
            # KeyError). The lock is re-entrant, so the nested ``save_note`` (which
            # re-takes it) is safe.
            with zg._embeddings_lock:
                note = zg.notes.pop(node_id, None)
                path = self._base / graph / f"{node_id}.md"
                try:
                    path.unlink(missing_ok=True)
                except OSError:
                    pass
                if note is not None:
                    title_index = getattr(zg, "_title_index", None)
                    if isinstance(title_index, dict):
                        title_index.pop((note.title or "").lower(), None)
                # Scrub dangling inbound SAME-GRAPH edges ("" means this graph).
                for other in zg.notes.values():
                    before = other.links or []
                    kept = [
                        link
                        for link in before
                        if link.target != node_id or (link.graph or "") not in ("", graph)
                    ]
                    if len(kept) != len(before):
                        other.links = kept
                        zg.save_note(other)
                return note is not None

    def node_links(self, graph: str, node_id: str) -> "list[Link]":
        """The outgoing links of ``node_id`` in ``graph`` (empty if absent)."""
        note = self._graph(graph).notes.get(node_id)
        return list(note.links or []) if note is not None else []

    def remove_link(
        self,
        graph: str,
        source_id: str,
        target_id: str,
        relation: str,
        *,
        target_graph: str = "",
    ) -> bool:
        """Drop a specific outgoing ``source --relation--> target`` edge.

        Matches on ``(target, relation, graph)`` (the same tuple :meth:`ensure_link`
        dedupes on), so it removes exactly the edge a prior ensure created. Returns
        True if an edge was removed. Used by sub-spine disconnect and by an explicit
        connect that MOVES a child to a new primary parent.
        """
        zg = self._graph(graph)
        note = zg.notes.get(source_id)
        if note is None:
            return False
        before = note.links or []
        kept = [
            link
            for link in before
            if not (
                link.target == target_id
                and link.relation == relation
                and (link.graph or "") == (target_graph or "")
            )
        ]
        if len(kept) == len(before):
            return False
        note.links = kept
        self._save(graph, note)
        return True

    def ensure_link(
        self,
        graph: str,
        source_id: str,
        target_id: str,
        relation: str,
        *,
        target_graph: str = "",
        confidence: "float | None" = None,
        provenance: str = "",
        verified: bool = False,
        primary: bool = False,
    ) -> bool:
        """Idempotently add an outgoing ``source --relation--> target`` edge.

        ``confidence`` / ``provenance`` / ``verified`` (all default-empty) are
        carried onto the edge so a ``spine-member`` edge can record the routing
        provenance of the member it attaches (inferred-vs-deterministic). They are
        REFRESHED in place on a re-run whose metadata drifted (e.g. a re-mine
        resync with a changed confidence), but identical metadata is a no-op — so
        the membership reconcile stays idempotent (no duplicate edges, no churn).
        Structural edges (apex/dimension/hub wiring) pass none and stay byte-clean.

        ``primary`` marks a durable sub-spine parent edge (``component-of`` apex→apex)
        and is UPGRADE-ONLY on a matching edge: a re-link may PROMOTE an existing
        edge to ``primary=True`` when requested, but an ordinary re-link (``primary``
        defaulting False — every structural/membership resync) NEVER demotes an
        existing ``primary=True`` edge. Clearing ``primary`` is the job of an explicit
        disconnect/unset (:func:`disconnect_sub_spine_edge` / :meth:`remove_link`),
        not this idempotent add — so a routine re-extraction can't silently wipe a
        hand-set / seeded primary parent. This mirrors the server ``link_notes``
        upgrade-only refresh so both backends obey the same stability rule.
        """
        zg = self._graph(graph)
        note = zg.notes.get(source_id)
        if note is None:
            return False
        provenance = provenance or ""
        verified = bool(verified)
        primary = bool(primary)
        for link in note.links or []:
            if (
                link.target == target_id
                and link.relation == relation
                and (link.graph or "") == (target_graph or "")
            ):
                # Dedupe — but refresh membership metadata if it drifted so a
                # re-classification updates without minting a duplicate edge.
                # ``primary`` is upgrade-only: OR the request onto the existing
                # flag so a re-link never demotes an already-primary edge.
                existing_primary = bool(getattr(link, "primary", False))
                new_primary = existing_primary or primary
                if (
                    link.confidence != confidence
                    or (link.provenance or "") != provenance
                    or bool(link.verified) != verified
                    or existing_primary != new_primary
                ):
                    link.confidence = confidence
                    link.provenance = provenance
                    link.verified = verified
                    link.primary = new_primary
                    self._save(graph, note)
                return False
        note.links.append(
            Link(
                target=target_id,
                relation=relation,
                graph=target_graph,
                confidence=confidence,
                provenance=provenance,
                verified=verified,
                primary=primary,
            )
        )
        self._save(graph, note)
        return True

    def set_member_targets(
        self,
        graph: str,
        node_id: str,
        relation: str,
        targets: "set[tuple[str, str]]",
    ) -> int:
        """Reconcile ``node_id``'s outgoing ``relation`` edges to EXACTLY ``targets``.

        ``targets`` is the desired set of ``(target_note_id, home_graph)`` pairs the
        node SHOULD carry. Any existing outgoing edge with ``relation`` whose
        ``(target, graph)`` is not in ``targets`` is DELETED — a member that LEFT the
        routing (re-typed, semantic re-cluster dropped it, or an overlay
        ``member_remove``). Edges of any OTHER relation (the structural
        dimension→apex / apex→hub wiring) are left untouched.

        This is the spine-side symmetric to :meth:`remove_links_into_graph`: it
        scrubs the SPINE graph's OWN membership edges (the v2 source of truth) so an
        add-only resync can't report stale membership to a direct reader. It NEVER
        touches the legacy base-side compat edges — those live on base notes in
        their home graph, which this never loads. Returns the number of edges
        removed (it only ever deletes; adds are :meth:`ensure_link`'s job).
        """
        zg = self._graph(graph)
        note = zg.notes.get(node_id)
        if note is None:
            return 0
        before = note.links or []
        kept = [
            link
            for link in before
            if link.relation != relation
            or (link.target, link.graph or "") in targets
        ]
        if len(kept) == len(before):
            return 0
        removed = len(before) - len(kept)
        note.links = kept
        self._save(graph, note)
        return removed

    def remove_links_into_graph(
        self, home_graph: str, target_graph: str, *, targets: "set[str] | None" = None
    ) -> int:
        """Drop outgoing links from ``home_graph`` notes that target ``target_graph``.

        The teardown primitive: an attach edge belongs to spine X iff its
        synthesis-graph endpoint is ``target_graph`` (the spine's graph), so
        scrubbing by ``link.graph`` removes exactly that spine's membership edges
        and leaves the base notes otherwise pristine. ``targets`` (optional)
        narrows the scrub to edges into specific node ids — used to detach base
        notes from an orphaned hub/dimension on re-promote without disturbing the
        rest of the spine.

        Raises on a graph-load failure so a caller (``delete_spine``) can prove the
        scrub actually covered every source graph before tearing the spine down,
        instead of silently treating an unreadable graph as "nothing to scrub".
        """
        zg = self._graph(home_graph)
        removed = 0
        for note in list(zg.notes.values()):
            before = note.links or []
            kept = [
                link
                for link in before
                if (link.graph or "") != target_graph
                or (targets is not None and link.target not in targets)
            ]
            if len(kept) != len(before):
                removed += len(before) - len(kept)
                note.links = kept
                self._save(home_graph, note)
        return removed

    def delete_graph(self, name: str) -> bool:
        """Remove a graph folder (and all its notes/edges). Returns True if it existed."""
        path = self._base / name
        existed = path.exists()
        if existed:
            shutil.rmtree(path, ignore_errors=True)
        self._cache.pop(name, None)
        return existed

find_node_by_stable_id

find_node_by_stable_id(graph: str, node_id: str) -> str

The id of the structure node carrying node_id in source, or "".

Structure (tree) nodes are keyed by their DURABLE source['node_id'] (§11.3 of the spine-promotion design) so a relabel re-titles the SAME node instead of orphaning it — the structural analogue of :meth:find_hub_by_row for row hubs.

Source code in zettelkasten/spine.py

def find_node_by_stable_id(self, graph: str, node_id: str) -> str:
    """The id of the structure node carrying ``node_id`` in ``source``, or "".

    Structure (tree) nodes are keyed by their DURABLE ``source['node_id']`` (§11.3
    of the spine-promotion design) so a relabel re-titles the SAME node instead
    of orphaning it — the structural analogue of :meth:`find_hub_by_row` for
    row hubs.
    """
    if not node_id:
        return ""
    zg = self._graph(graph)
    for note in zg.notes.values():
        src = note.source if isinstance(note.source, dict) else {}
        if str(src.get(STRUCTURE_NODE_ID_KEY) or "") == node_id:
            return note.id
    return ""

ensure_structure_node

ensure_structure_node(graph: str, node_id: str, title: str, ntype: str, body: str, tags, *, synthesis_status: str = '', spine_title: str = '') -> str

Idempotently create/find a structure (tree) node keyed by stable node_id.

Identity is resolved in three tiers so the durable id survives a relabel AND a legacy (node-id-less) spine migrates forward on its next promote:

1. Find by the stable source['node_id']. If the title drifted (a column relabel), re-title the SAME node — its graph id (and thus every spine-member edge + column['apex'] reference) is preserved.
2. Else find by title (the legacy find-by-title dedup build_spine_skeleton always used) and STAMP node_id onto it, so a spine materialized before V2b acquires durable ids the first time it is re-promoted. Two exact title formats are tried, in order: the current BARE title (e.g. "Dataset"), then — when spine_title is supplied — the OLD COMPOSITE format a pre-bare-label spine minted the node under, f"{spine_title} — {title}" (space, em-dash, space; the exact string organizations.py used before dimension titles were changed to the bare column label). A legacy composite-title node is thus still matched on its first re-promote after that change; it is then re-titled IN PLACE to the bare title and stamped with the durable node_id, so its spine-member edges + overlay corrections are preserved rather than orphaned by a fresh mint. Only these two EXACT titles are tried — no positional or fuzzy guessing. HARDENED: tier 2 migrates ONLY a GENUINE legacy DIMENSION node — one tagged DIMENSION_TAG, with no durable id yet and no conflicting identity. It REFUSES to stamp node_id onto ANY non-dimension same-title collision: a row hub (source['row_id'] or a HUB_TAGS tag), the apex or a spec (APEX_TAGS/SPEC_TAG), a plain note (no dimension tag), or a node already carrying a DIFFERENT node_id. A title collision (e.g. a dimension whose label equals the apex title, a row-hub label, or a stray note) must never let a dimension hijack a foreign node's identity — doing so wired the apex component-of itself and silently dropped a member under the apex-exclusion rollup. Any such foreign collision falls through to tier 3.
3. Else create a fresh node carrying source['node_id'].

Passing an empty node_id degrades to plain find-by-title creation (no durable id), matching :meth:ensure_node. spine_title is used ONLY for the tier-2 legacy-composite fallback and only when a durable node_id is requested, so the degraded and greenfield paths are byte-identical to before.

Source code in zettelkasten/spine.py

def ensure_structure_node(
    self,
    graph: str,
    node_id: str,
    title: str,
    ntype: str,
    body: str,
    tags,
    *,
    synthesis_status: str = "",
    spine_title: str = "",
) -> str:
    """Idempotently create/find a structure (tree) node keyed by stable ``node_id``.

    Identity is resolved in three tiers so the durable id survives a relabel
    AND a legacy (node-id-less) spine migrates forward on its next promote:

    1. Find by the stable ``source['node_id']``. If the title drifted (a column
       relabel), re-title the SAME node — its graph id (and thus every
       ``spine-member`` edge + ``column['apex']`` reference) is preserved.
    2. Else find by title (the legacy find-by-title dedup ``build_spine_skeleton``
       always used) and STAMP ``node_id`` onto it, so a spine materialized
       before V2b acquires durable ids the first time it is re-promoted.
       Two exact title formats are tried, in order: the current BARE ``title``
       (e.g. ``"Dataset"``), then — when ``spine_title`` is supplied — the OLD
       COMPOSITE format a pre-bare-label spine minted the node under,
       ``f"{spine_title} — {title}"`` (space, em-dash, space; the exact string
       ``organizations.py`` used before dimension titles were changed to the
       bare column label). A legacy composite-title node is thus still matched
       on its first re-promote after that change; it is then re-titled IN PLACE
       to the bare ``title`` and stamped with the durable ``node_id``, so its
       ``spine-member`` edges + overlay corrections are preserved rather than
       orphaned by a fresh mint. Only these two EXACT titles are tried — no
       positional or fuzzy guessing.
       HARDENED: tier 2 migrates ONLY a GENUINE legacy DIMENSION node — one
       tagged ``DIMENSION_TAG``, with no durable id yet and no conflicting
       identity. It REFUSES to stamp ``node_id`` onto ANY non-dimension
       same-title collision: a row hub (``source['row_id']`` or a ``HUB_TAGS``
       tag), the apex or a spec (``APEX_TAGS``/``SPEC_TAG``), a plain note
       (no dimension tag), or a node already carrying a DIFFERENT ``node_id``.
       A title collision (e.g. a dimension whose label equals the apex title,
       a row-hub label, or a stray note) must never let a dimension hijack a
       foreign node's identity — doing so wired the apex ``component-of``
       itself and silently dropped a member under the apex-exclusion rollup.
       Any such foreign collision falls through to tier 3.
    3. Else create a fresh node carrying ``source['node_id']``.

    Passing an empty ``node_id`` degrades to plain find-by-title creation
    (no durable id), matching :meth:`ensure_node`. ``spine_title`` is used
    ONLY for the tier-2 legacy-composite fallback and only when a durable
    ``node_id`` is requested, so the degraded and greenfield paths are
    byte-identical to before.
    """
    zg = self._graph(graph)
    existing_id = self.find_node_by_stable_id(graph, node_id)
    if existing_id:
        note = zg.notes.get(existing_id)
        if note is not None and title and (note.title or "") != title:
            note.title = title
            self._save(graph, note)
        return existing_id
    existing = zg.find_by_title(title)
    if existing is None and node_id and spine_title:
        # Legacy-composite fallback: a dimension minted before the bare-label
        # change carries the OLD composite title and has no durable id, so the
        # bare-title lookup above misses it. Try the exact old format so it is
        # matched (and migrated below) instead of orphaned by a fresh mint.
        legacy_composite = f"{spine_title} — {title}"
        if legacy_composite != title:
            existing = zg.find_by_title(legacy_composite)
    if existing is not None:
        src = existing.source if isinstance(existing.source, dict) else {}
        existing_node_id = str(src.get(STRUCTURE_NODE_ID_KEY) or "")
        if not node_id:
            # Degraded find-by-title creation (no durable id requested).
            return existing.id
        # Only a genuine legacy DIMENSION node may be migrated in place: it
        # must be tagged ``DIMENSION_TAG`` and must NOT be a row hub, the apex,
        # a spec, or carry a conflicting durable id. Any other same-title node
        # (apex, spec, hub, plain note) is a FOREIGN collision: leave it
        # untouched and mint a fresh dimension node (tier 3) so a dimension can
        # never hijack a foreign node's identity (the apex-self-loop bug).
        existing_tags = set(existing.tags or [])
        is_dimension = DIMENSION_TAG in existing_tags
        is_hub = bool(str(src.get("row_id") or "")) or bool(
            set(HUB_TAGS) & existing_tags
        )
        is_apex_or_spec = bool(set(APEX_TAGS) & existing_tags) or (
            SPEC_TAG in existing_tags
        )
        conflicting_id = bool(existing_node_id) and existing_node_id != node_id
        if is_dimension and not is_hub and not is_apex_or_spec and not conflicting_id:
            changed = False
            if existing_node_id != node_id:
                src = dict(src)
                src[STRUCTURE_NODE_ID_KEY] = node_id
                existing.source = src
                changed = True
            # Retitle in place when matched via the legacy composite fallback,
            # so the node forward-migrates to the bare label (a bare-title
            # match already equals ``title``, so this is a no-op there).
            if title and (existing.title or "") != title:
                existing.title = title
                changed = True
            if changed:
                self._save(graph, existing)
            return existing.id
        # Foreign collision: fall through to mint a fresh node below.
    nid = zg.generate_unique_id(title)
    source = {STRUCTURE_NODE_ID_KEY: node_id} if node_id else {}
    self._save(
        graph,
        Note(
            id=nid, title=title, type=ntype, source=source, body=body,
            tags=list(tags), synthesis_status=synthesis_status,
        ),
    )
    return nid

apex_node_ids

apex_node_ids(graph: str) -> list[str]

Ids of every APEX_TAGS node in graph (the single-apex guard).

A well-formed spine has exactly one; a second is the corruption :func:build_spine_skeleton's guard fails loud on.

Source code in zettelkasten/spine.py

def apex_node_ids(self, graph: str) -> list[str]:
    """Ids of every ``APEX_TAGS`` node in ``graph`` (the single-apex guard).

    A well-formed spine has exactly one; a second is the corruption
    :func:`build_spine_skeleton`'s guard fails loud on.
    """
    zg = self._graph(graph)
    apex = set(APEX_TAGS)
    return [n.id for n in zg.notes.values() if apex <= set(n.tags or [])]

ensure_apex_node

ensure_apex_node(graph: str, node_id: str, title: str, ntype: str, body: str, tags, *, synthesis_status: str = '') -> str

Idempotently create/find the spine APEX, keyed by the durable node_id.

The apex is the ROOT structure node, so — like the dimensions — it must survive a relabel / org-rename / title_template drift by re-titling the SAME node rather than minting a second APEX_TAGS node (which would orphan the primary sub-spine edge stored on the original apex). Identity is resolved in three tiers:

1. Find by the durable source['node_id'] (the apex sentinel). If the title drifted, re-title the SAME node — its graph id (and every edge on it, including the sub-spine component-of) is preserved.
2. Else find the existing apex by its APEX_TAGS (a spine has exactly one apex, so this is unambiguous) and STAMP the sentinel node_id onto it IN PLACE. This is the one-time, byte-safe forward migration for a legacy spine whose apex predates the durable id — it NEVER mints a duplicate.
3. Else create a fresh apex carrying the sentinel source['node_id'].

Unlike :meth:ensure_structure_node (which REFUSES to migrate an APEX_TAGS node so a dimension can never hijack the apex's identity), this method is the apex's OWN builder: migrating the apex in place is exactly the intent. Passing an empty node_id degrades to find-by-title creation, matching :meth:ensure_node.

Source code in zettelkasten/spine.py

def ensure_apex_node(
    self,
    graph: str,
    node_id: str,
    title: str,
    ntype: str,
    body: str,
    tags,
    *,
    synthesis_status: str = "",
) -> str:
    """Idempotently create/find the spine APEX, keyed by the durable ``node_id``.

    The apex is the ROOT structure node, so — like the dimensions — it must
    survive a relabel / org-rename / ``title_template`` drift by re-titling the
    SAME node rather than minting a second ``APEX_TAGS`` node (which would
    orphan the primary sub-spine edge stored on the original apex). Identity is
    resolved in three tiers:

    1. Find by the durable ``source['node_id']`` (the apex sentinel). If the
       title drifted, re-title the SAME node — its graph id (and every edge on
       it, including the sub-spine ``component-of``) is preserved.
    2. Else find the existing apex by its ``APEX_TAGS`` (a spine has exactly one
       apex, so this is unambiguous) and STAMP the sentinel ``node_id`` onto it
       IN PLACE. This is the one-time, byte-safe forward migration for a legacy
       spine whose apex predates the durable id — it NEVER mints a duplicate.
    3. Else create a fresh apex carrying the sentinel ``source['node_id']``.

    Unlike :meth:`ensure_structure_node` (which REFUSES to migrate an
    ``APEX_TAGS`` node so a dimension can never hijack the apex's identity),
    this method is the apex's OWN builder: migrating the apex in place is
    exactly the intent. Passing an empty ``node_id`` degrades to find-by-title
    creation, matching :meth:`ensure_node`.
    """
    zg = self._graph(graph)
    existing_id = self.find_node_by_stable_id(graph, node_id)
    if existing_id:
        note = zg.notes.get(existing_id)
        if note is not None and title and (note.title or "") != title:
            note.title = title
            self._save(graph, note)
        return existing_id
    if node_id:
        # Tier 2: migrate the legacy title-keyed apex in place. Prefer an apex
        # that already carries the durable id (none here, tier 1 handled that),
        # then any apex with NO durable id, so we never clobber a foreign id.
        apex = set(APEX_TAGS)
        for note in zg.notes.values():
            if not (apex <= set(note.tags or [])):
                continue
            src = note.source if isinstance(note.source, dict) else {}
            existing_node_id = str(src.get(STRUCTURE_NODE_ID_KEY) or "")
            if existing_node_id and existing_node_id != node_id:
                # A different durable id already claims this apex — do not
                # overwrite it; fall through and let the guard surface the
                # conflict rather than silently rewriting identity.
                continue
            changed = False
            if existing_node_id != node_id:
                src = dict(src)
                src[STRUCTURE_NODE_ID_KEY] = node_id
                note.source = src
                changed = True
            if title and (note.title or "") != title:
                note.title = title
                changed = True
            if changed:
                self._save(graph, note)
            return note.id
    # Tier 3: no durable-id or legacy apex found — create a fresh one.
    existing = zg.find_by_title(title)
    if existing is not None and not node_id:
        return existing.id
    nid = zg.generate_unique_id(title)
    source = {STRUCTURE_NODE_ID_KEY: node_id} if node_id else {}
    self._save(
        graph,
        Note(
            id=nid, title=title, type=ntype, source=source, body=body,
            tags=list(tags), synthesis_status=synthesis_status,
        ),
    )
    return nid

find_hub_by_row

find_hub_by_row(graph: str, row_id: str) -> str

The id of the hub node carrying row_id in note.source, or "".

Hubs are keyed by the row's STABLE id (not its label) so two rows sharing a label never collapse into one hub and a label change re-titles the same hub instead of orphaning it.

Source code in zettelkasten/spine.py

def find_hub_by_row(self, graph: str, row_id: str) -> str:
    """The id of the hub node carrying ``row_id`` in ``note.source``, or "".

    Hubs are keyed by the row's STABLE id (not its label) so two rows sharing
    a label never collapse into one hub and a label change re-titles the same
    hub instead of orphaning it.
    """
    zg = self._graph(graph)
    for note in zg.notes.values():
        src = note.source if isinstance(note.source, dict) else {}
        if str(src.get("row_id") or "") == row_id and row_id:
            return note.id
    return ""

ensure_hub

ensure_hub(graph: str, row_id: str, label: str, body: str, tags=HUB_TAGS) -> str

Idempotently create/update a row hub keyed by stable row_id.

Stores the original row_id on the hub's source so the link-form rebuild recovers it (overlay corrections key on it). Updates the title to label if it changed, so re-promoting after a row relabel re-titles the existing hub rather than orphaning it.

Source code in zettelkasten/spine.py

def ensure_hub(self, graph: str, row_id: str, label: str, body: str, tags=HUB_TAGS) -> str:
    """Idempotently create/update a row hub keyed by stable ``row_id``.

    Stores the original ``row_id`` on the hub's ``source`` so the link-form
    rebuild recovers it (overlay corrections key on it). Updates the title to
    ``label`` if it changed, so re-promoting after a row relabel re-titles the
    existing hub rather than orphaning it.
    """
    zg = self._graph(graph)
    existing_id = self.find_hub_by_row(graph, row_id)
    if existing_id:
        note = zg.notes.get(existing_id)
        if note is not None and (note.title or "") != label and label:
            note.title = label
            self._save(graph, note)
        return existing_id
    nid = zg.generate_unique_id(label or row_id)
    self._save(
        graph,
        Note(
            id=nid,
            title=label or row_id,
            type="concept",
            source={"row_id": row_id},
            body=body,
            tags=list(tags),
        ),
    )
    return nid

set_hub_basis

set_hub_basis(graph: str, hub_id: str, basis: 'list[str]') -> None

Persist a hub's MEMBER BASIS (the <home>::<note_id> uid set) on its source['member_basis'].

The basis is the last-known membership the overlap remap falls back to when a hub has zero LIVE member edges (every member was overlay-removed): without it a regrown semantic group could not be matched to its existing hub and would mint a duplicate row. Stored spine-side on the hub node, so base notes stay pristine.

Source code in zettelkasten/spine.py

def set_hub_basis(self, graph: str, hub_id: str, basis: "list[str]") -> None:
    """Persist a hub's MEMBER BASIS (the ``<home>::<note_id>`` uid set) on its
    ``source['member_basis']``.

    The basis is the last-known membership the overlap remap falls back to when
    a hub has zero LIVE member edges (every member was overlay-removed): without
    it a regrown semantic group could not be matched to its existing hub and
    would mint a duplicate row. Stored spine-side on the hub node, so base notes
    stay pristine.
    """
    zg = self._graph(graph)
    note = zg.notes.get(hub_id)
    if note is None:
        return
    src = note.source if isinstance(note.source, dict) else {}
    new_basis = sorted({u for u in basis if u})
    if list(src.get("member_basis") or []) == new_basis:
        return
    src = dict(src)
    src["member_basis"] = new_basis
    note.source = src
    self._save(graph, note)

structural_node_ids

structural_node_ids(graph: str, tags) -> list[str]

Ids of every node in graph carrying ANY of tags (hub/dim scan).

Source code in zettelkasten/spine.py

def structural_node_ids(self, graph: str, tags) -> list[str]:
    """Ids of every node in ``graph`` carrying ANY of ``tags`` (hub/dim scan)."""
    zg = self._graph(graph)
    want = set(tags)
    return [n.id for n in zg.notes.values() if want & set(n.tags or [])]

node_row_id

node_row_id(graph: str, node_id: str) -> str

The stable source['row_id'] a node carries (hubs only), or "".

Read BEFORE :meth:delete_node so a caller can tell which overlay corrections a soon-to-be-pruned hub was carrying (orphan detection).

Source code in zettelkasten/spine.py

def node_row_id(self, graph: str, node_id: str) -> str:
    """The stable ``source['row_id']`` a node carries (hubs only), or "".

    Read BEFORE :meth:`delete_node` so a caller can tell which overlay
    corrections a soon-to-be-pruned hub was carrying (orphan detection).
    """
    zg = self._graph(graph)
    note = zg.notes.get(node_id)
    if note is None:
        return ""
    src = note.source if isinstance(note.source, dict) else {}
    return str(src.get("row_id") or "")

delete_node

delete_node(graph: str, node_id: str) -> bool

Remove a single node (file + in-memory index) from graph.

Also scrubs any DANGLING SAME-GRAPH edge that pointed AT the deleted node (e.g. the structural apex --related--> hub link a hub prune would otherwise leave behind). A cross-graph edge whose target merely shares this id but lives in ANOTHER graph (a spine-member edge into a base note's home graph) is left intact — its target still exists in its own graph — so a surviving node never loses an edge it genuinely still needs.

Source code in zettelkasten/spine.py

def delete_node(self, graph: str, node_id: str) -> bool:
    """Remove a single node (file + in-memory index) from ``graph``.

    Also scrubs any DANGLING SAME-GRAPH edge that pointed AT the deleted node
    (e.g. the structural ``apex --related--> hub`` link a hub prune would
    otherwise leave behind). A cross-graph edge whose target merely shares this
    id but lives in ANOTHER graph (a ``spine-member`` edge into a base note's
    home graph) is left intact — its target still exists in its own graph — so
    a surviving node never loses an edge it genuinely still needs.
    """
    from zettelkasten.commit import review_write_lock

    zg = self._graph(graph)
    # Take the CROSS-PROCESS ``box::{graph}`` file lock FIRST (outermost), then
    # the in-process ``_embeddings_lock`` — the SAME order ``save_note`` uses
    # (file lock → embeddings lock). delete_node mutates ``zg.notes`` directly
    # (a file unlink + index pop + an inbound-edge scrub) so it MUST hold the
    # same cross-process lock the MCP server takes on ``box::{graph}``, or a
    # concurrent same-graph writer in another process would lose an update /
    # diverge its index. The nested scrub write below therefore stays UNLOCKED
    # (plain ``save_note`` with no ``lock_key``): re-taking the file lock from
    # inside the ``_embeddings_lock`` would INVERT the lock order and risk an
    # ABBA deadlock, so the outer lock covers the whole critical section.
    #
    # When ``graph`` is in ``held_box_locks`` the caller (a lifecycle op)
    # ALREADY holds ``box::{graph}`` across its whole RMW, so we must NOT
    # re-acquire the NON-reentrant lock here (it would deadlock); a
    # ``nullcontext`` lets the caller's outer lock cover this delete instead.
    outer_lock = (
        contextlib.nullcontext()
        if graph in self._held_box_locks
        else review_write_lock(f"box::{graph}", graphs_dir=self._base)
    )
    with outer_lock:
        # Serialize every direct ``zg.notes`` mutation below under the same
        # RLock ``get_backlinks``/``_reindex_backlinks``/``save_note`` use, so
        # the ``notes.pop`` and the ``zg.notes.values()`` scrub loop cannot race
        # a concurrent get_backlinks lazy rebuild iterating ``self.notes`` (which
        # would otherwise raise "dictionary changed size during iteration" or
        # KeyError). The lock is re-entrant, so the nested ``save_note`` (which
        # re-takes it) is safe.
        with zg._embeddings_lock:
            note = zg.notes.pop(node_id, None)
            path = self._base / graph / f"{node_id}.md"
            try:
                path.unlink(missing_ok=True)
            except OSError:
                pass
            if note is not None:
                title_index = getattr(zg, "_title_index", None)
                if isinstance(title_index, dict):
                    title_index.pop((note.title or "").lower(), None)
            # Scrub dangling inbound SAME-GRAPH edges ("" means this graph).
            for other in zg.notes.values():
                before = other.links or []
                kept = [
                    link
                    for link in before
                    if link.target != node_id or (link.graph or "") not in ("", graph)
                ]
                if len(kept) != len(before):
                    other.links = kept
                    zg.save_note(other)
            return note is not None

node_links

node_links(graph: str, node_id: str) -> 'list[Link]'

The outgoing links of node_id in graph (empty if absent).

Source code in zettelkasten/spine.py

def node_links(self, graph: str, node_id: str) -> "list[Link]":
    """The outgoing links of ``node_id`` in ``graph`` (empty if absent)."""
    note = self._graph(graph).notes.get(node_id)
    return list(note.links or []) if note is not None else []

remove_link

remove_link(graph: str, source_id: str, target_id: str, relation: str, *, target_graph: str = '') -> bool

Drop a specific outgoing source --relation--> target edge.

Matches on (target, relation, graph) (the same tuple :meth:ensure_link dedupes on), so it removes exactly the edge a prior ensure created. Returns True if an edge was removed. Used by sub-spine disconnect and by an explicit connect that MOVES a child to a new primary parent.

Source code in zettelkasten/spine.py

def remove_link(
    self,
    graph: str,
    source_id: str,
    target_id: str,
    relation: str,
    *,
    target_graph: str = "",
) -> bool:
    """Drop a specific outgoing ``source --relation--> target`` edge.

    Matches on ``(target, relation, graph)`` (the same tuple :meth:`ensure_link`
    dedupes on), so it removes exactly the edge a prior ensure created. Returns
    True if an edge was removed. Used by sub-spine disconnect and by an explicit
    connect that MOVES a child to a new primary parent.
    """
    zg = self._graph(graph)
    note = zg.notes.get(source_id)
    if note is None:
        return False
    before = note.links or []
    kept = [
        link
        for link in before
        if not (
            link.target == target_id
            and link.relation == relation
            and (link.graph or "") == (target_graph or "")
        )
    ]
    if len(kept) == len(before):
        return False
    note.links = kept
    self._save(graph, note)
    return True

ensure_link

ensure_link(graph: str, source_id: str, target_id: str, relation: str, *, target_graph: str = '', confidence: 'float | None' = None, provenance: str = '', verified: bool = False, primary: bool = False) -> bool

Idempotently add an outgoing source --relation--> target edge.

confidence / provenance / verified (all default-empty) are carried onto the edge so a spine-member edge can record the routing provenance of the member it attaches (inferred-vs-deterministic). They are REFRESHED in place on a re-run whose metadata drifted (e.g. a re-mine resync with a changed confidence), but identical metadata is a no-op — so the membership reconcile stays idempotent (no duplicate edges, no churn). Structural edges (apex/dimension/hub wiring) pass none and stay byte-clean.

primary marks a durable sub-spine parent edge (component-of apex→apex) and is UPGRADE-ONLY on a matching edge: a re-link may PROMOTE an existing edge to primary=True when requested, but an ordinary re-link (primary defaulting False — every structural/membership resync) NEVER demotes an existing primary=True edge. Clearing primary is the job of an explicit disconnect/unset (:func:disconnect_sub_spine_edge / :meth:remove_link), not this idempotent add — so a routine re-extraction can't silently wipe a hand-set / seeded primary parent. This mirrors the server link_notes upgrade-only refresh so both backends obey the same stability rule.

Source code in zettelkasten/spine.py

def ensure_link(
    self,
    graph: str,
    source_id: str,
    target_id: str,
    relation: str,
    *,
    target_graph: str = "",
    confidence: "float | None" = None,
    provenance: str = "",
    verified: bool = False,
    primary: bool = False,
) -> bool:
    """Idempotently add an outgoing ``source --relation--> target`` edge.

    ``confidence`` / ``provenance`` / ``verified`` (all default-empty) are
    carried onto the edge so a ``spine-member`` edge can record the routing
    provenance of the member it attaches (inferred-vs-deterministic). They are
    REFRESHED in place on a re-run whose metadata drifted (e.g. a re-mine
    resync with a changed confidence), but identical metadata is a no-op — so
    the membership reconcile stays idempotent (no duplicate edges, no churn).
    Structural edges (apex/dimension/hub wiring) pass none and stay byte-clean.

    ``primary`` marks a durable sub-spine parent edge (``component-of`` apex→apex)
    and is UPGRADE-ONLY on a matching edge: a re-link may PROMOTE an existing
    edge to ``primary=True`` when requested, but an ordinary re-link (``primary``
    defaulting False — every structural/membership resync) NEVER demotes an
    existing ``primary=True`` edge. Clearing ``primary`` is the job of an explicit
    disconnect/unset (:func:`disconnect_sub_spine_edge` / :meth:`remove_link`),
    not this idempotent add — so a routine re-extraction can't silently wipe a
    hand-set / seeded primary parent. This mirrors the server ``link_notes``
    upgrade-only refresh so both backends obey the same stability rule.
    """
    zg = self._graph(graph)
    note = zg.notes.get(source_id)
    if note is None:
        return False
    provenance = provenance or ""
    verified = bool(verified)
    primary = bool(primary)
    for link in note.links or []:
        if (
            link.target == target_id
            and link.relation == relation
            and (link.graph or "") == (target_graph or "")
        ):
            # Dedupe — but refresh membership metadata if it drifted so a
            # re-classification updates without minting a duplicate edge.
            # ``primary`` is upgrade-only: OR the request onto the existing
            # flag so a re-link never demotes an already-primary edge.
            existing_primary = bool(getattr(link, "primary", False))
            new_primary = existing_primary or primary
            if (
                link.confidence != confidence
                or (link.provenance or "") != provenance
                or bool(link.verified) != verified
                or existing_primary != new_primary
            ):
                link.confidence = confidence
                link.provenance = provenance
                link.verified = verified
                link.primary = new_primary
                self._save(graph, note)
            return False
    note.links.append(
        Link(
            target=target_id,
            relation=relation,
            graph=target_graph,
            confidence=confidence,
            provenance=provenance,
            verified=verified,
            primary=primary,
        )
    )
    self._save(graph, note)
    return True

set_member_targets

set_member_targets(graph: str, node_id: str, relation: str, targets: 'set[tuple[str, str]]') -> int

Reconcile node_id's outgoing relation edges to EXACTLY targets.

targets is the desired set of (target_note_id, home_graph) pairs the node SHOULD carry. Any existing outgoing edge with relation whose (target, graph) is not in targets is DELETED — a member that LEFT the routing (re-typed, semantic re-cluster dropped it, or an overlay member_remove). Edges of any OTHER relation (the structural dimension→apex / apex→hub wiring) are left untouched.

This is the spine-side symmetric to :meth:remove_links_into_graph: it scrubs the SPINE graph's OWN membership edges (the v2 source of truth) so an add-only resync can't report stale membership to a direct reader. It NEVER touches the legacy base-side compat edges — those live on base notes in their home graph, which this never loads. Returns the number of edges removed (it only ever deletes; adds are :meth:ensure_link's job).

Source code in zettelkasten/spine.py

def set_member_targets(
    self,
    graph: str,
    node_id: str,
    relation: str,
    targets: "set[tuple[str, str]]",
) -> int:
    """Reconcile ``node_id``'s outgoing ``relation`` edges to EXACTLY ``targets``.

    ``targets`` is the desired set of ``(target_note_id, home_graph)`` pairs the
    node SHOULD carry. Any existing outgoing edge with ``relation`` whose
    ``(target, graph)`` is not in ``targets`` is DELETED — a member that LEFT the
    routing (re-typed, semantic re-cluster dropped it, or an overlay
    ``member_remove``). Edges of any OTHER relation (the structural
    dimension→apex / apex→hub wiring) are left untouched.

    This is the spine-side symmetric to :meth:`remove_links_into_graph`: it
    scrubs the SPINE graph's OWN membership edges (the v2 source of truth) so an
    add-only resync can't report stale membership to a direct reader. It NEVER
    touches the legacy base-side compat edges — those live on base notes in
    their home graph, which this never loads. Returns the number of edges
    removed (it only ever deletes; adds are :meth:`ensure_link`'s job).
    """
    zg = self._graph(graph)
    note = zg.notes.get(node_id)
    if note is None:
        return 0
    before = note.links or []
    kept = [
        link
        for link in before
        if link.relation != relation
        or (link.target, link.graph or "") in targets
    ]
    if len(kept) == len(before):
        return 0
    removed = len(before) - len(kept)
    note.links = kept
    self._save(graph, note)
    return removed

remove_links_into_graph

remove_links_into_graph(home_graph: str, target_graph: str, *, targets: 'set[str] | None' = None) -> int

Drop outgoing links from home_graph notes that target target_graph.

The teardown primitive: an attach edge belongs to spine X iff its synthesis-graph endpoint is target_graph (the spine's graph), so scrubbing by link.graph removes exactly that spine's membership edges and leaves the base notes otherwise pristine. targets (optional) narrows the scrub to edges into specific node ids — used to detach base notes from an orphaned hub/dimension on re-promote without disturbing the rest of the spine.

Raises on a graph-load failure so a caller (delete_spine) can prove the scrub actually covered every source graph before tearing the spine down, instead of silently treating an unreadable graph as "nothing to scrub".

Source code in zettelkasten/spine.py

def remove_links_into_graph(
    self, home_graph: str, target_graph: str, *, targets: "set[str] | None" = None
) -> int:
    """Drop outgoing links from ``home_graph`` notes that target ``target_graph``.

    The teardown primitive: an attach edge belongs to spine X iff its
    synthesis-graph endpoint is ``target_graph`` (the spine's graph), so
    scrubbing by ``link.graph`` removes exactly that spine's membership edges
    and leaves the base notes otherwise pristine. ``targets`` (optional)
    narrows the scrub to edges into specific node ids — used to detach base
    notes from an orphaned hub/dimension on re-promote without disturbing the
    rest of the spine.

    Raises on a graph-load failure so a caller (``delete_spine``) can prove the
    scrub actually covered every source graph before tearing the spine down,
    instead of silently treating an unreadable graph as "nothing to scrub".
    """
    zg = self._graph(home_graph)
    removed = 0
    for note in list(zg.notes.values()):
        before = note.links or []
        kept = [
            link
            for link in before
            if (link.graph or "") != target_graph
            or (targets is not None and link.target not in targets)
        ]
        if len(kept) != len(before):
            removed += len(before) - len(kept)
            note.links = kept
            self._save(home_graph, note)
    return removed

delete_graph

delete_graph(name: str) -> bool

Remove a graph folder (and all its notes/edges). Returns True if it existed.

Source code in zettelkasten/spine.py

def delete_graph(self, name: str) -> bool:
    """Remove a graph folder (and all its notes/edges). Returns True if it existed."""
    path = self._base / name
    existed = path.exists()
    if existed:
        shutil.rmtree(path, ignore_errors=True)
    self._cache.pop(name, None)
    return existed

SubSpineCycleError

Bases: ValueError

Raised when connecting a child beneath a parent would close a parent cycle.

Source code in zettelkasten/spine.py

class SubSpineCycleError(ValueError):
    """Raised when connecting a child beneath a parent would close a parent cycle."""

build_spine_skeleton

build_spine_skeleton(ops, *, graph_name: str, graph_description: str = '', apex_title: str, apex_type: str, apex_body: str = '', apex_relation: str = DEFAULT_APEX_RELATION, attach_relation: str = DEFAULT_ATTACH_RELATION, dimensions: list[dict[str, Any]], spec: 'dict[str, Any] | None' = None, scaffold_status: str = 'scaffold') -> dict[str, Any]

Stamp the apex + (optional spec) + the dimension TREE, idempotently.

dimensions is a list of {key, tag, title, type, body} facet specs. Each becomes a node tagged [<tag>, DIMENSION_TAG]. A spine is a structure TREE rooted at the apex (§11.3 — the V2b model): a facet may carry a children list of the same shape, and each child is wired child -> parent (apex_relation, default component-of) to ARBITRARY depth — so dimension/hub stop being fixed levels and become "structure node at depth N". A FLAT list (no children) yields the historical depth-1 shape exactly: every facet is component-of the apex. spec (optional) is {title, type, body, relation}, wired spec -> apex.

Every structure node is given a DURABLE id (§11.4): an explicit node_id (or stable_id) on the facet, else derived from its schema PATH (the slash-joined chain of key\ s from the root). The id is persisted spine-side on source['node_id'] so readback + overlay corrections can bind to it and survive a relabel/drift/re-promote. Node creation is dedup'd (by stable id, then by title), so re-running on an unchanged graph is a no-op.

scaffold_status (default "scaffold") is stamped as synthesis_status on every freshly-created apex/spec/dimension node, so an un-authored spine is queryable as incomplete until a synthesizer flips it to "materialized". Pass "" to skip stamping. (Re-running over existing nodes never changes their status — the dedup returns early, preserving a materialized flip.)

Returns {synthesis_graph, apex_id, spec_id, dimension_nodes, attach_relation, apex_relation} where dimension_nodes maps EACH facet's key (interior AND leaf, full tree) to its node id (member attach edges + hub wiring are the caller's job — they differ between the empty-scaffold and bulk-attach paths).

Source code in zettelkasten/spine.py

def build_spine_skeleton(
    ops,
    *,
    graph_name: str,
    graph_description: str = "",
    apex_title: str,
    apex_type: str,
    apex_body: str = "",
    apex_relation: str = DEFAULT_APEX_RELATION,
    attach_relation: str = DEFAULT_ATTACH_RELATION,
    dimensions: list[dict[str, Any]],
    spec: "dict[str, Any] | None" = None,
    scaffold_status: str = "scaffold",
) -> dict[str, Any]:
    """Stamp the apex + (optional spec) + the dimension TREE, idempotently.

    ``dimensions`` is a list of ``{key, tag, title, type, body}`` facet specs.
    Each becomes a node tagged ``[<tag>, DIMENSION_TAG]``. A spine is a structure
    TREE rooted at the apex (§11.3 — the V2b model): a facet may carry a
    ``children`` list of the same shape, and each child is wired ``child -> parent``
    (``apex_relation``, default ``component-of``) to ARBITRARY depth — so
    ``dimension``/``hub`` stop being fixed levels and become "structure node at
    depth N". A FLAT list (no ``children``) yields the historical depth-1 shape
    exactly: every facet is ``component-of`` the apex. ``spec`` (optional) is
    ``{title, type, body, relation}``, wired ``spec -> apex``.

    Every structure node is given a DURABLE id (§11.4): an explicit ``node_id``
    (or ``stable_id``) on the facet, else derived from its schema PATH (the
    slash-joined chain of ``key``\\ s from the root). The id is persisted spine-side
    on ``source['node_id']`` so readback + overlay corrections can bind to it and
    survive a relabel/drift/re-promote. Node creation is dedup'd (by stable id,
    then by title), so re-running on an unchanged graph is a no-op.

    ``scaffold_status`` (default ``"scaffold"``) is stamped as ``synthesis_status``
    on every freshly-created apex/spec/dimension node, so an un-authored spine is
    queryable as incomplete until a synthesizer flips it to ``"materialized"``.
    Pass ``""`` to skip stamping. (Re-running over existing nodes never changes
    their status — the dedup returns early, preserving a materialized flip.)

    Returns ``{synthesis_graph, apex_id, spec_id, dimension_nodes, attach_relation,
    apex_relation}`` where ``dimension_nodes`` maps EACH facet's ``key`` (interior
    AND leaf, full tree) to its node id (member attach edges + hub wiring are the
    caller's job — they differ between the empty-scaffold and bulk-attach paths).
    """
    ops.ensure_graph(graph_name, graph_description)

    # The apex is the ROOT structure node: key it on the durable sentinel id so a
    # re-title / org-rename / ``title_template`` drift re-titles the SAME apex node
    # instead of minting a second one (which would orphan the primary sub-spine
    # edge stored on the original apex). A legacy title-keyed apex is migrated in
    # place (byte-safe, never duplicated) on this first durable-id build.
    apex_id = _ensure_apex_node(
        ops, graph_name, APEX_NODE_ID, apex_title, apex_type, apex_body, APEX_TAGS,
        synthesis_status=scaffold_status,
    )
    # Guard the tree invariant: exactly one apex per spine graph.
    _guard_single_apex(ops, graph_name, apex_id)

    spec_id = ""
    if spec:
        spec_id = ops.ensure_node(
            graph_name,
            spec["title"],
            spec.get("type") or "model",
            spec.get("body", ""),
            (SPEC_TAG, "extraction-synthesis"),
            synthesis_status=scaffold_status,
        )
        if spec_id and apex_id:
            ops.ensure_link(graph_name, spec_id, apex_id, spec.get("relation") or apex_relation)

    dimension_nodes: dict[str, str] = {}

    def _build_level(facets: list[dict[str, Any]], parent_id: str, path: list[str]) -> None:
        for dim in facets:
            tag = str(dim.get("tag") or "").strip()
            dim_tags = [t for t in (tag, DIMENSION_TAG) if t]
            key = str(dim["key"])
            stable_id = str(dim.get("node_id") or dim.get("stable_id") or "").strip()
            if not stable_id:
                # Derive from the schema PATH so the id is stable across rebuilds
                # without the caller having to mint one (and unique per tree node).
                stable_id = "/".join([*path, key])
            nid = _ensure_structure_node(
                ops,
                graph_name,
                stable_id,
                dim["title"],
                dim.get("type") or "concept",
                dim.get("body", ""),
                dim_tags,
                synthesis_status=scaffold_status,
                spine_title=apex_title,
            )
            if not nid:
                continue
            # Routing keys MUST be unique within a spine. The returned
            # ``dimension_nodes`` map is keyed by ``key`` and is the tag-based
            # routing/enforcement contract for the WHOLE spine: ``attach_to_dimension``
            # resolves ``dimension_nodes.get(tag)`` and ``reconcile_meta_tags`` uses
            # ``dimension_nodes.keys()`` as the strict vocabulary. Two facets sharing
            # a key (e.g. a duplicate tag in different branches of the tree) would
            # silently collapse here — the later node clobbering the earlier in the
            # map even though both exist on disk under distinct path-derived stable
            # ids — and route claims ambiguously. Tag-based enforcement cannot
            # distinguish them, so a duplicate key is unsupportable: fail LOUD at
            # build time rather than collapse silently.
            if key in dimension_nodes:
                raise SpineError(
                    f"Duplicate dimension routing key '{key}' in spine "
                    f"'{graph_name}'. Dimension tags must be unique within a spine "
                    "— attach routing and strict enforcement are tag-keyed, so two "
                    "facets sharing a key are indistinguishable to the scribe. "
                    "Give each facet a distinct tag/key."
                )
            dimension_nodes[key] = nid
            # A child is ``component-of`` its PARENT (the apex for a top-level
            # facet) — the recursive generalization of the flat depth-1 spine.
            if parent_id:
                ops.ensure_link(graph_name, nid, parent_id, apex_relation)
            children = dim.get("children")
            if children:
                _build_level(children, nid, [*path, key])

    _build_level(dimensions, apex_id, [])

    return {
        "synthesis_graph": graph_name,
        "apex_id": apex_id,
        "spec_id": spec_id,
        "dimension_nodes": dimension_nodes,
        "attach_relation": attach_relation,
        "apex_relation": apex_relation,
    }

resolve_apex

resolve_apex(zg: ZettelGraph) -> tuple[str, str]

THE single deterministic apex resolver for a loaded spine graph.

Every consumer that needs "which node is this spine's apex?" — the id used for UID keys, parent links, and matrix row ids — MUST resolve it through here so spine.find_apex_id, tables._spine_apex and the org overlay route can never disagree on a legacy or drifted graph. Resolution walks four tiers in strict priority order, returning (id, title):

1. Durable sentinel — the node whose source['node_id'] equals APEX_NODE_ID. This is the authoritative apex once stamped; it survives a re-title and carries the primary sub-spine edge, so it wins outright even if a legacy duplicate apex lingers.
2. Structural vote — the node the most DIMENSION_TAG nodes point to via an intra-graph component-of edge (link.graph in ('', zg.name)). This recovers the tree root on a pre-sentinel graph where tag alone is ambiguous (e.g. a derived model node tagged alongside the real apex). A vote TIE is broken by the lexicographically-first (id-sorted) candidate — never by dict iteration / note insertion order — so the winner is stable across loads.
3. First FULL-APEX_TAGS node by ID-SORTED order — a deterministic tie-break when there are no dimensions/structural edges. Requires the node to carry the FULL APEX_TAGS set (set(APEX_TAGS) <= node.tags), matching apex_node_ids / _guard_single_apex — a mere tag intersection would let a spec node (tagged ('spec', 'extraction-synthesis')) masquerade as apex. Sorting by id (rather than dict iteration order) keeps the choice stable.
4. ("", "") — the graph carries no apex at all.

Source code in zettelkasten/spine.py

def resolve_apex(zg: ZettelGraph) -> tuple[str, str]:
    """THE single deterministic apex resolver for a loaded spine graph.

    Every consumer that needs "which node is this spine's apex?" — the id used for
    UID keys, parent links, and matrix row ids — MUST resolve it through here so
    ``spine.find_apex_id``, ``tables._spine_apex`` and the org overlay route can
    never disagree on a legacy or drifted graph. Resolution walks four tiers in
    strict priority order, returning ``(id, title)``:

    1. **Durable sentinel** — the node whose ``source['node_id']`` equals
       ``APEX_NODE_ID``. This is the authoritative apex once stamped; it survives
       a re-title and carries the primary sub-spine edge, so it wins outright even
       if a legacy duplicate apex lingers.
    2. **Structural vote** — the node the most ``DIMENSION_TAG`` nodes point to via
       an intra-graph ``component-of`` edge (``link.graph in ('', zg.name)``). This
       recovers the tree root on a pre-sentinel graph where tag alone is ambiguous
       (e.g. a derived ``model`` node tagged alongside the real apex). A vote TIE is
       broken by the lexicographically-first (id-sorted) candidate — never by dict
       iteration / note insertion order — so the winner is stable across loads.
    3. **First FULL-``APEX_TAGS`` node by ID-SORTED order** — a deterministic
       tie-break when there are no dimensions/structural edges. Requires the node to
       carry the FULL ``APEX_TAGS`` set (``set(APEX_TAGS) <= node.tags``), matching
       ``apex_node_ids`` / ``_guard_single_apex`` — a mere tag intersection would let
       a spec node (tagged ``('spec', 'extraction-synthesis')``) masquerade as apex.
       Sorting by id (rather than dict iteration order) keeps the choice stable.
    4. ``("", "")`` — the graph carries no apex at all.
    """
    apex_tags = set(APEX_TAGS)

    # Tier 1: durable sentinel.
    for note in zg.notes.values():
        src = note.source if isinstance(note.source, dict) else {}
        if str(src.get(STRUCTURE_NODE_ID_KEY) or "") == APEX_NODE_ID:
            return note.id, (note.title or note.id)

    # Tier 2: structural vote (mirrors the historical ``tables._spine_apex``).
    targets: Counter[str] = Counter()
    for note in zg.notes.values():
        if DIMENSION_TAG not in set(note.tags or []):
            continue
        for link in note.links or []:
            if getattr(link, "tombstoned", False):
                continue
            if link.relation != DEFAULT_APEX_RELATION:
                continue
            if (getattr(link, "graph", "") or "") in ("", zg.name) and link.target:
                targets[link.target] += 1
    if targets:
        # Deterministic tie-break: among the max-vote candidates pick the
        # lexicographically-first id (``most_common`` would leak insertion order).
        top_votes = max(targets.values())
        apex_id = min(nid for nid, votes in targets.items() if votes == top_votes)
        note = zg.notes.get(apex_id)
        if note is not None:
            return apex_id, (note.title or apex_id)

    # Tier 3: first FULL-``APEX_TAGS`` node by ID-SORTED order (deterministic).
    # Require the full APEX_TAGS subset (not a mere intersection) so a spec node —
    # tagged ``('spec', 'extraction-synthesis')`` — is never mistaken for the apex,
    # matching ``apex_node_ids`` / ``_guard_single_apex``.
    for note_id in sorted(zg.notes.keys()):
        note = zg.notes[note_id]
        if apex_tags <= set(note.tags or []):
            return note.id, (note.title or note.id)

    # Tier 4: no apex.
    return "", ""

find_apex_id

find_apex_id(ops, graph_name: str, get_graph: Callable[[str], ZettelGraph]) -> str

The id of the spine apex node in graph_name (tagged APEX_TAGS), or "".

Thin wrapper over :func:resolve_apex: loads the graph and returns the resolved apex id, preserving the historical empty-on-failure behavior when the graph can't be loaded.

Source code in zettelkasten/spine.py

def find_apex_id(ops, graph_name: str, get_graph: Callable[[str], ZettelGraph]) -> str:
    """The id of the spine apex node in ``graph_name`` (tagged ``APEX_TAGS``), or "".

    Thin wrapper over :func:`resolve_apex`: loads the graph and returns the
    resolved apex id, preserving the historical empty-on-failure behavior when the
    graph can't be loaded.
    """
    try:
        zg = get_graph(graph_name)
    except Exception:
        return ""
    return resolve_apex(zg)[0]

find_primary_parent_link

find_primary_parent_link(ops, child_graph: str, child_apex_id: str) -> 'Link | None'

The child apex's existing PRIMARY sub-spine parent edge, or None.

Identifies a sub-spine parent by its primary flag on a component-of edge (the intra-spine dimension→apex tree uses component-of too, but never primary, so the flag alone is an unambiguous discriminator).

Source code in zettelkasten/spine.py

def find_primary_parent_link(ops, child_graph: str, child_apex_id: str) -> "Link | None":
    """The child apex's existing PRIMARY sub-spine parent edge, or ``None``.

    Identifies a sub-spine parent by its ``primary`` flag on a ``component-of``
    edge (the intra-spine dimension→apex tree uses ``component-of`` too, but never
    ``primary``, so the flag alone is an unambiguous discriminator).
    """
    for link in ops.node_links(child_graph, child_apex_id):
        if link.relation == SUB_SPINE_RELATION and bool(getattr(link, "primary", False)):
            return link
    return None

connect_sub_spine_edge

connect_sub_spine_edge(ops, *, child_graph: str, child_apex_id: str, parent_graph: str, parent_apex_id: str, respect_existing: bool = False) -> dict[str, Any]

Write the primary child-apex --component-of--> parent-apex edge.

A child spine has exactly ONE primary parent (design §2: the composition is a true tree), so an EXPLICIT connect that points the child at a DIFFERENT parent MOVES it — the stale primary edge is removed before the new one is written. The write is idempotent (ensure_link dedupes), so re-connecting to the same parent is a no-op.

respect_existing implements the STABILITY RULE (design §3): when True (the build-time seed), a pre-existing primary parent edge WINS — if the child already carries a primary parent pointing ELSEWHERE, the seed does NOT replace it (a hand-set / hand-moved parent is never silently reverted). Left False for the explicit connect front door, which is itself a hand-authoring action.

Rejects a connect that would close a parent CYCLE — connecting the child beneath a parent that is already a DESCENDANT of the child (connect(A->B) then connect(B->A)) raises :class:SubSpineCycleError. The immediate self-parent case (child_graph == parent_graph) is guarded upstream by the front doors; this covers the deeper multi-hop cycle.

Returns {connected: bool, added: bool, moved_from: {...}|None, kept_existing: {...}|None}.

Source code in zettelkasten/spine.py

def connect_sub_spine_edge(
    ops,
    *,
    child_graph: str,
    child_apex_id: str,
    parent_graph: str,
    parent_apex_id: str,
    respect_existing: bool = False,
) -> dict[str, Any]:
    """Write the primary ``child-apex --component-of--> parent-apex`` edge.

    A child spine has exactly ONE primary parent (design §2: the composition is a
    true tree), so an EXPLICIT connect that points the child at a DIFFERENT parent
    MOVES it — the stale primary edge is removed before the new one is written.
    The write is idempotent (``ensure_link`` dedupes), so re-connecting to the same
    parent is a no-op.

    ``respect_existing`` implements the STABILITY RULE (design §3): when True (the
    build-time seed), a pre-existing primary parent edge WINS — if the child already
    carries a primary parent pointing ELSEWHERE, the seed does NOT replace it (a
    hand-set / hand-moved parent is never silently reverted). Left False for the
    explicit connect front door, which is itself a hand-authoring action.

    Rejects a connect that would close a parent CYCLE — connecting the child
    beneath a parent that is already a DESCENDANT of the child (``connect(A->B)``
    then ``connect(B->A)``) raises :class:`SubSpineCycleError`. The immediate
    self-parent case (``child_graph == parent_graph``) is guarded upstream by the
    front doors; this covers the deeper multi-hop cycle.

    Returns ``{connected: bool, added: bool, moved_from: {...}|None,
    kept_existing: {...}|None}``.
    """
    if _proposed_parent_is_descendant(
        ops,
        child_graph=child_graph,
        child_apex_id=child_apex_id,
        parent_graph=parent_graph,
        parent_apex_id=parent_apex_id,
    ):
        raise SubSpineCycleError(
            f"cannot nest spine apex '{child_graph}::{child_apex_id}' beneath "
            f"'{parent_graph}::{parent_apex_id}': the proposed parent is already a "
            "descendant of the child (this would create a component-of cycle)"
        )
    existing = find_primary_parent_link(ops, child_graph, child_apex_id)
    same_target = (
        existing is not None
        and existing.target == parent_apex_id
        and (existing.graph or "") == (parent_graph or "")
    )
    if existing is not None and not same_target and respect_existing:
        # Stability rule: a hand-authored parent edge is never overwritten by a
        # build-time seed. Report the edge that was kept so the caller can flag it.
        return {
            "connected": False,
            "added": False,
            "moved_from": None,
            "kept_existing": {"apex_id": existing.target, "graph": existing.graph or ""},
        }
    moved_from: "dict[str, str] | None" = None
    if existing is not None and not same_target:
        # Explicit connect to a new parent: enforce single-primary-parent by
        # removing the stale edge first. ``remove_link`` is only on the direct-IO
        # backend (the front doors use it); a backend without it simply appends
        # (the walker keys on ``primary`` + graph, tolerating a transient dup).
        remover = getattr(ops, "remove_link", None)
        if remover is not None:
            remover(
                child_graph,
                child_apex_id,
                existing.target,
                SUB_SPINE_RELATION,
                target_graph=existing.graph or "",
            )
            moved_from = {"apex_id": existing.target, "graph": existing.graph or ""}
    added = ops.ensure_link(
        child_graph,
        child_apex_id,
        parent_apex_id,
        SUB_SPINE_RELATION,
        target_graph=parent_graph,
        primary=True,
    )
    return {
        "connected": True,
        "added": bool(added),
        "moved_from": moved_from,
        "kept_existing": None,
    }

disconnect_sub_spine_edge

disconnect_sub_spine_edge(ops, *, child_graph: str, child_apex_id: str, parent_graph: str = '', parent_apex_id: str = '') -> dict[str, Any]

Remove the child apex's primary sub-spine parent edge.

With no parent_apex_id given, removes whatever primary parent edge the child currently carries (the common "detach this sub-spine" case). When a specific parent is named, only that edge is removed (and only if it is the child's primary parent). Idempotent: removing an absent edge reports removed=False.

Source code in zettelkasten/spine.py

def disconnect_sub_spine_edge(
    ops,
    *,
    child_graph: str,
    child_apex_id: str,
    parent_graph: str = "",
    parent_apex_id: str = "",
) -> dict[str, Any]:
    """Remove the child apex's primary sub-spine parent edge.

    With no ``parent_apex_id`` given, removes whatever primary parent edge the
    child currently carries (the common "detach this sub-spine" case). When a
    specific parent is named, only that edge is removed (and only if it is the
    child's primary parent). Idempotent: removing an absent edge reports
    ``removed=False``.
    """
    existing = find_primary_parent_link(ops, child_graph, child_apex_id)
    if existing is None:
        return {"removed": False, "parent": None}
    if parent_apex_id and not (
        existing.target == parent_apex_id
        and (existing.graph or "") == (parent_graph or "")
    ):
        # A specific parent was requested but it is not the child's primary parent.
        return {"removed": False, "parent": None}
    remover = getattr(ops, "remove_link", None)
    if remover is None:
        return {"removed": False, "parent": None}
    removed = remover(
        child_graph,
        child_apex_id,
        existing.target,
        SUB_SPINE_RELATION,
        target_graph=existing.graph or "",
    )
    return {
        "removed": bool(removed),
        "parent": {"apex_id": existing.target, "graph": existing.graph or ""},
    }