zettelkasten.remine

zettelkasten.remine

Re-mining note->dimension classifier (synthesis-matrix Stage 1b CORE).

The deterministic matrix router (:func:zettelkasten.tables._route_column_members) fills a cell from EXACT-TAG / spine-side membership only — a note that belongs to a dimension but was never tagged for it is invisible to the grid. This module is the agent-semantic tier that closes that gap: a NOTE-CENTRIC, BATCHED classifier that reads the row's scoped notes and the schema's dimensions and returns, per note, the dimension key(s) it belongs to plus a verbatim quote per assignment. A note may land in MULTIPLE dimensions; notes that fit none are SURFACED in a residual bucket rather than silently dropped.

It composes with the deterministic tier (never replaces it) via the classify_fn seam that :func:zettelkasten.tables._route_members consumes: build_classify_fn returns a per-column callable that runs ONE batched classification per row (cached across that row's columns) and hands back the assignments for the queried dimension. The member-level rules (augment-never-replace, provenance/confidence, verbatim verification) live in tables so this module stays a pure classifier.

Cost discipline, in order:

• Exact-tag short-circuit — notes already carrying a dimension's tag are already routed deterministically and are NEVER re-sent to the LLM.
• Embedding pre-filter — an optional :class:~zettelkasten.embeddings.EmbeddingIndex gates which remaining notes reach the LLM: only notes whose vector is near a dimension's deterministic centroid (or a tagged seed) are sent. The threshold is LOOSE by default (favor recall — the LLM is the precision step) and tunable.
• One batched call — the surviving candidate notes are classified against the whole dimension set in a single LLM call, mirroring the :func:zettelkasten.tables._extract_row_cells call/parse shape.

Single-schema scope today; the return shape carries a schema field per assignment so a later note->{schema->dimensions} fan-out is additive.

RemineAssignment dataclass

One agent-inferred note->dimension assignment.

schema is carried (even though single-schema today) so the shape extends cleanly toward note->{schema->dimensions} without breaking consumers.

Source code in zettelkasten/remine.py

@dataclass
class RemineAssignment:
    """One agent-inferred note->dimension assignment.

    ``schema`` is carried (even though single-schema today) so the shape extends
    cleanly toward note->{schema->dimensions} without breaking consumers.
    """

    note_id: str
    dimension: str
    quote: str = ""
    confidence: "float | None" = None
    schema: str = ""

RemineResult dataclass

The batched classification of one note set against one dimension set.

assignments maps note_id -> [RemineAssignment, ...] (a note may land in several dimensions). residual lists the note ids that fit NO dimension and were not already deterministically tagged — surfaced, never dropped. considered is the candidate ids actually sent to the LLM (the rest were pruned by the exact-tag short-circuit or the embedding pre-filter).

Source code in zettelkasten/remine.py

@dataclass
class RemineResult:
    """The batched classification of one note set against one dimension set.

    ``assignments`` maps ``note_id -> [RemineAssignment, ...]`` (a note may land in
    several dimensions). ``residual`` lists the note ids that fit NO dimension and
    were not already deterministically tagged — surfaced, never dropped.
    ``considered`` is the candidate ids actually sent to the LLM (the rest were
    pruned by the exact-tag short-circuit or the embedding pre-filter).
    """

    assignments: dict[str, list[RemineAssignment]] = field(default_factory=dict)
    residual: list[str] = field(default_factory=list)
    considered: list[str] = field(default_factory=list)
    schema: str = ""

    def for_dimension(self, dimension: str) -> list[dict[str, Any]]:
        """The seam payload for one dimension: ``[{note_id, quote, confidence}]``."""
        out: list[dict[str, Any]] = []
        for nid, items in self.assignments.items():
            for a in items:
                if a.dimension == dimension:
                    out.append(
                        {"note_id": nid, "quote": a.quote, "confidence": a.confidence}
                    )
        return out

for_dimension

for_dimension(dimension: str) -> list[dict[str, Any]]

The seam payload for one dimension: [{note_id, quote, confidence}].

Source code in zettelkasten/remine.py

def for_dimension(self, dimension: str) -> list[dict[str, Any]]:
    """The seam payload for one dimension: ``[{note_id, quote, confidence}]``."""
    out: list[dict[str, Any]] = []
    for nid, items in self.assignments.items():
        for a in items:
            if a.dimension == dimension:
                out.append(
                    {"note_id": nid, "quote": a.quote, "confidence": a.confidence}
                )
    return out

expand_dimensions

expand_dimensions(schema_name: str) -> list[dict[str, Any]]

Build classifier dimensions from a registry schema's expanded spec.

Reuses :func:zettelkasten.extraction_schemas.expand_schema so the dimension descriptions the classifier (and the embedding pre-filter) read are exactly the schema's — keyed by the dimension tag, which is what a spine/schema_tag column keys its dimension on. Returns [] for an unknown schema.

Source code in zettelkasten/remine.py

def expand_dimensions(schema_name: str) -> list[dict[str, Any]]:
    """Build classifier dimensions from a registry schema's expanded spec.

    Reuses :func:`zettelkasten.extraction_schemas.expand_schema` so the dimension
    descriptions the classifier (and the embedding pre-filter) read are exactly
    the schema's — keyed by the dimension tag, which is what a spine/schema_tag
    column keys its dimension on. Returns ``[]`` for an unknown schema.
    """
    from zettelkasten import extraction_schemas

    spec = extraction_schemas.expand_schema(schema_name)
    if not spec:
        return []
    dims = [
        {"key": d["tag"], "tag": d["tag"], "title": d.get("claim_framing") or d["tag"], "desc": d.get("desc") or ""}
        for d in spec.get("dimensions", [])
    ]
    return _normalize_remine_dimensions(dims)

classify_notes

classify_notes(notes: 'list[Any]', dimensions: 'list[dict[str, Any]]', *, extract_fn: 'Callable[[str, str], str] | None' = None, embed_index: Any = None, similarity_threshold: float = DEFAULT_SIMILARITY_THRESHOLD, schema: str = '', body_char_budget: int = 2000) -> RemineResult

Classify notes into dimensions in ONE batched LLM call.

notes is a list of (zg, note) scope pairs (or bare note objects). dimensions are descriptors (key/tag/title/desc), e.g. from :func:expand_dimensions. extract_fn(system, prompt) -> str is the LLM seam (mirrors :func:zettelkasten.tables._extract_row_cells); inject a fake in tests. embed_index (optional) gates LLM cost via the loose embedding pre-filter.

Pipeline: exact-tag short-circuit (skip already-tagged notes per dimension) → embedding pre-filter (per dimension) → one batched LLM call over the union of surviving candidates → parse, gated so an assignment is kept only when the note actually survived that dimension's gate. Notes assigned to no dimension (and not deterministically tagged anywhere) are returned in residual.

Source code in zettelkasten/remine.py

def classify_notes(
    notes: "list[Any]",
    dimensions: "list[dict[str, Any]]",
    *,
    extract_fn: "Callable[[str, str], str] | None" = None,
    embed_index: Any = None,
    similarity_threshold: float = DEFAULT_SIMILARITY_THRESHOLD,
    schema: str = "",
    body_char_budget: int = 2000,
) -> RemineResult:
    """Classify ``notes`` into ``dimensions`` in ONE batched LLM call.

    ``notes`` is a list of ``(zg, note)`` scope pairs (or bare note objects).
    ``dimensions`` are descriptors (``key``/``tag``/``title``/``desc``), e.g. from
    :func:`expand_dimensions`. ``extract_fn(system, prompt) -> str`` is the LLM
    seam (mirrors :func:`zettelkasten.tables._extract_row_cells`); inject a fake in
    tests. ``embed_index`` (optional) gates LLM cost via the loose embedding
    pre-filter.

    Pipeline: exact-tag short-circuit (skip already-tagged notes per dimension) →
    embedding pre-filter (per dimension) → one batched LLM call over the union of
    surviving candidates → parse, gated so an assignment is kept only when the
    note actually survived that dimension's gate. Notes assigned to no dimension
    (and not deterministically tagged anywhere) are returned in ``residual``.
    """
    dims = _normalize_remine_dimensions(dimensions)
    views = [v for v in (_as_view(n) for n in notes) if v is not None]
    # De-dupe by note id (a scope can in principle repeat a note), keep first.
    seen_ids: set[str] = set()
    uniq: list[_NoteView] = []
    for v in views:
        if v.note_id in seen_ids:
            continue
        seen_ids.add(v.note_id)
        uniq.append(v)
    views = uniq
    by_id = {v.note_id: v for v in views}

    if not dims or not views:
        return RemineResult(schema=schema)

    # Exact-tag short-circuit: a note already carrying a dimension's tag is routed
    # deterministically — never re-send it to the LLM for that dimension.
    seeds: dict[str, set[str]] = {}
    tagged_any: set[str] = set()
    for d in dims:
        tag = d["tag"]
        s = {v.note_id for v in views if tag in v.tags}
        seeds[d["key"]] = s
        tagged_any |= s

    # Embedding pre-filter, per dimension, over the not-yet-tagged pool.
    send: dict[str, set[str]] = {}
    for d in dims:
        key = d["key"]
        pool = [v.note_id for v in views if v.note_id not in seeds[key]]
        send[key] = _prefilter_candidates(
            pool, embed_index=embed_index, seeds=seeds[key], threshold=similarity_threshold
        ) & set(pool)

    batch_ids: set[str] = set()
    for s in send.values():
        batch_ids |= s
    considered = sorted(batch_ids)

    assignments: dict[str, list[RemineAssignment]] = {}
    if batch_ids and extract_fn is not None:
        batch_views = [by_id[nid] for nid in considered]
        prompt = _build_prompt(batch_views, dims, body_char_budget=body_char_budget)
        try:
            raw = extract_fn(_CLASSIFY_CONTRACT, prompt) or ""
        except Exception as exc:  # noqa: BLE001 — a classifier hiccup degrades to no augmentation
            logger.warning("remine classify call failed: %s", exc)
            raw = ""
        data = _parse_json_object(raw)
        valid_keys = {d["key"] for d in dims}
        if isinstance(data, dict):
            for nid, items in data.items():
                nid = str(nid)
                if nid not in by_id or not isinstance(items, list):
                    continue
                for entry in items:
                    if not isinstance(entry, dict):
                        continue
                    dim_key = str(entry.get("dimension") or "").strip()
                    if dim_key not in valid_keys:
                        continue
                    # Honor the per-dimension gates: never accept an assignment for
                    # a note the dimension's pre-filter pruned, nor one already
                    # routed deterministically for that dimension.
                    if nid in seeds.get(dim_key, set()) or nid not in send.get(dim_key, set()):
                        continue
                    assignments.setdefault(nid, []).append(
                        RemineAssignment(
                            note_id=nid,
                            dimension=dim_key,
                            quote=str(entry.get("quote") or "").strip(),
                            confidence=_coerce_confidence(entry.get("confidence")),
                            schema=schema,
                        )
                    )

    assigned_ids = {nid for nid, items in assignments.items() if items}
    residual = [
        v.note_id for v in views if v.note_id not in assigned_ids and v.note_id not in tagged_any
    ]
    return RemineResult(
        assignments=assignments,
        residual=residual,
        considered=considered,
        schema=schema,
    )

build_classify_fn

build_classify_fn(dimensions: 'list[dict[str, Any]]', *, extract_fn: 'Callable[[str, str], str] | None' = None, embed_index: Any = None, similarity_threshold: float = DEFAULT_SIMILARITY_THRESHOLD, schema: str = '', residual_out: 'dict[str, list[str]] | None' = None) -> 'Callable[[Any, dict[str, Any]], list[dict[str, Any]]]'

A per-column classify_fn for :func:zettelkasten.tables._route_members.

The matrix router calls classify_fn(scope, column) once PER non-prompt column of a row. Re-running the batched classifier per column would be wasteful, so the returned closure runs :func:classify_notes ONCE per row (keyed by the live scope object) and caches the result, then returns just the assignments for the queried column's dimension (column["key"]).

residual_out (optional) is the sink for the residual bucket: each row's unassigned note ids are recorded under residual_out[scope.id] so the caller can SURFACE notes that fit no dimension (they are never dropped).

extract_fn defaults to the in-process dashboard agent; inject a fake in tests. Production typically does::

from zettelkasten import remine
dims = remine.expand_dimensions("decision-profile")
classify_fn = remine.build_classify_fn(dims, embed_index=index)
rows = tables.gather_rows(get_graph, cols, ..., classify_fn=classify_fn)

Source code in zettelkasten/remine.py

def build_classify_fn(
    dimensions: "list[dict[str, Any]]",
    *,
    extract_fn: "Callable[[str, str], str] | None" = None,
    embed_index: Any = None,
    similarity_threshold: float = DEFAULT_SIMILARITY_THRESHOLD,
    schema: str = "",
    residual_out: "dict[str, list[str]] | None" = None,
) -> "Callable[[Any, dict[str, Any]], list[dict[str, Any]]]":
    """A per-column ``classify_fn`` for :func:`zettelkasten.tables._route_members`.

    The matrix router calls ``classify_fn(scope, column)`` once PER non-prompt
    column of a row. Re-running the batched classifier per column would be
    wasteful, so the returned closure runs :func:`classify_notes` ONCE per row
    (keyed by the live scope object) and caches the result, then returns just the
    assignments for the queried column's dimension (``column["key"]``).

    ``residual_out`` (optional) is the sink for the residual bucket: each row's
    unassigned note ids are recorded under ``residual_out[scope.id]`` so the
    caller can SURFACE notes that fit no dimension (they are never dropped).

    ``extract_fn`` defaults to the in-process dashboard agent; inject a fake in
    tests. Production typically does::

        from zettelkasten import remine
        dims = remine.expand_dimensions("decision-profile")
        classify_fn = remine.build_classify_fn(dims, embed_index=index)
        rows = tables.gather_rows(get_graph, cols, ..., classify_fn=classify_fn)
    """
    dims = _normalize_remine_dimensions(dimensions)
    extract = extract_fn if extract_fn is not None else _default_classify_extract_fn
    cache: dict[Any, RemineResult] = {}

    def classify_fn(scope: Any, column: dict[str, Any]) -> list[dict[str, Any]]:
        # Key the per-row memo on the scope's STABLE identity (``scope.id``) when
        # present, NOT ``id(scope)``. A transient per-row scope (e.g. the enrich
        # loop's ``SimpleNamespace``) can be freed and have its memory address
        # recycled for a LATER row's scope, so ``id(scope)`` can alias distinct
        # rows and hand a later row an earlier row's cached result. A stable
        # ``.id`` is unique per row across every caller (row-scope uids, source
        # names, group ids; the single backfill scope), so distinct rows never
        # collide while the SAME row across multiple columns still hits the cache.
        # The ``("id", ...)``/``("obj", ...)`` namespacing keeps a string id and
        # an int ``id(scope)`` fallback from ever colliding.
        scope_id = getattr(scope, "id", None)
        cache_key = ("id", str(scope_id)) if scope_id else ("obj", id(scope))
        result = cache.get(cache_key)
        if result is None:
            result = classify_notes(
                getattr(scope, "notes", []) or [],
                dims,
                extract_fn=extract,
                embed_index=embed_index,
                similarity_threshold=similarity_threshold,
                schema=schema,
            )
            cache[cache_key] = result
            if residual_out is not None:
                residual_out[str(scope_id or id(scope))] = list(result.residual)
        return result.for_dimension(str(column.get("key") or ""))

    return classify_fn

induce_dimensions

induce_dimensions(scope_notes: 'list[Any]', intent: str, *, llm_fn: 'Callable[[str, str], str] | None', embed_index: Any = None, max_cols: int = 8) -> list[dict[str, Any]]

Induce classifier dimensions (FACET columns) from a corpus + user intent.

The agent (llm_fn(system, prompt) -> str, the same call/parse seam as the classifier/extract callables) reads a compact digest of scope_notes and the user's intent ("make a matrix for X") and proposes a small set of facet columns, each {tag, title, desc}. When embed_index is supplied the candidate facets are grounded with embedding clustering (:func:zettelkasten.embeddings.propose_clusters over get_vector); with embed_index=None it is agent-only.

The result is normalized through the SAME :func:_normalize_remine_dimensions registry dims pass through, so induced dims are byte-compatible with registry dims and flow through :func:_columns_from_dimensions unchanged. Facet keys/tags are slugs that are DETERMINISTIC given identical facet TEXT — but the agent that emits the facets is not stabilized, so a re-run may yield different facet text and therefore different columns (a known limitation, analogous to the Stage 2 semantic row-id follow-up). True-duplicate facets collapse; genuinely different facets that slug to the same key are disambiguated rather than dropped (see :func:_facets_to_dimensions). Returns [] when there is nothing to induce (empty corpus, empty intent, or no llm_fn).

Source code in zettelkasten/remine.py

def induce_dimensions(
    scope_notes: "list[Any]",
    intent: str,
    *,
    llm_fn: "Callable[[str, str], str] | None",
    embed_index: Any = None,
    max_cols: int = 8,
) -> list[dict[str, Any]]:
    """Induce classifier dimensions (FACET columns) from a corpus + user intent.

    The agent (``llm_fn(system, prompt) -> str``, the same call/parse seam as the
    classifier/extract callables) reads a compact digest of ``scope_notes`` and the
    user's ``intent`` ("make a matrix for X") and proposes a small set of facet
    columns, each ``{tag, title, desc}``. When ``embed_index`` is supplied the
    candidate facets are grounded with embedding clustering
    (:func:`zettelkasten.embeddings.propose_clusters` over ``get_vector``); with
    ``embed_index=None`` it is agent-only.

    The result is normalized through the SAME
    :func:`_normalize_remine_dimensions` registry dims pass through, so induced dims
    are byte-compatible with registry dims and flow through
    :func:`_columns_from_dimensions` unchanged. Facet keys/tags are slugs that are
    DETERMINISTIC given identical facet TEXT — but the agent that emits the facets
    is not stabilized, so a re-run may yield different facet text and therefore
    different columns (a known limitation, analogous to the Stage 2 semantic row-id
    follow-up). True-duplicate facets collapse; genuinely different facets that slug
    to the same key are disambiguated rather than dropped (see
    :func:`_facets_to_dimensions`). Returns ``[]`` when there is nothing to induce
    (empty corpus, empty intent, or no ``llm_fn``).
    """
    views = [v for v in (_as_view(n) for n in scope_notes) if v is not None]
    seen_ids: set[str] = set()
    uniq: list[_NoteView] = []
    for v in views:
        if v.note_id in seen_ids:
            continue
        seen_ids.add(v.note_id)
        uniq.append(v)
    views = uniq

    if not views or not str(intent or "").strip() or llm_fn is None:
        return []

    digest = _corpus_digest(views)
    hints = _cluster_hints(views, embed_index, max_k=max_cols) if embed_index is not None else []
    prompt = _build_induce_prompt(intent, digest, hints)
    try:
        raw = llm_fn(_INDUCE_CONTRACT, prompt) or ""
    except Exception as exc:  # noqa: BLE001 — an induction hiccup degrades to no columns
        logger.warning("remine induce call failed: %s", exc)
        raw = ""

    return _facets_to_dimensions(_parse_json_array(raw), max_cols=max_cols)

induce_subfacets

induce_subfacets(cell_notes: 'list[Any]', parent_facet: dict[str, Any], *, llm_fn: 'Callable[[str, str], str] | None', embed_index: Any = None, min_clusters: int = 2, min_cluster_size: int = 2, min_cohesion: float = 0.55, max_children: int = 8, max_depth: int = 1) -> list[dict[str, Any]]

Induce child FACET columns under one emergent column from its cell notes.

cell_notes are the (zg, note) pairs (or bare notes) that the classifier routed into parent_facet's column. They are sub-clustered by embedding cohesion (:func:zettelkasten.embeddings.propose_clusters with threshold=min_cohesion, max_k=max_children); the surviving clusters are NAMED by the agent into child facets through the SAME call/parse/slug seam as :func:induce_dimensions (llm_fn + :func:_build_induce_prompt + :func:_parse_json_array + :func:_facets_to_dimensions), so child keys are deterministic, disambiguated, and build_spine_skeleton-shaped.

Cohesion is the REQUIRED signal — with embed_index=None (no vectors) there is no basis to claim the column splits, so this returns [] rather than fabricating a subtree. A sufficient-clusters GATE keeps the column FLAT unless at least min_clusters clusters of at least min_cluster_size members survive; the unclustered bucket and sub-min clusters do NOT each become a child. max_depth bounds recursion: each named child's own cluster is the corpus for its depth pass (max_depth - 1), so a child carries its own children only while depth remains. Returns [] when the gate fails, the corpus is empty, llm_fn is missing, or max_depth <= 0.

Determinism: :func:zettelkasten.embeddings.propose_clusters is deterministic and the text→slug mapping is deterministic given identical facet text, so identical inputs yield identical child keys. As in Stage 3, the agent's NAMING itself is not stabilized — a re-run can emit different child titles and therefore different slugs (a known limitation).

Source code in zettelkasten/remine.py

def induce_subfacets(
    cell_notes: "list[Any]",
    parent_facet: dict[str, Any],
    *,
    llm_fn: "Callable[[str, str], str] | None",
    embed_index: Any = None,
    min_clusters: int = 2,
    min_cluster_size: int = 2,
    min_cohesion: float = 0.55,
    max_children: int = 8,
    max_depth: int = 1,
) -> list[dict[str, Any]]:
    """Induce child FACET columns under one emergent column from its cell notes.

    ``cell_notes`` are the ``(zg, note)`` pairs (or bare notes) that the classifier
    routed into ``parent_facet``'s column. They are sub-clustered by embedding
    cohesion (:func:`zettelkasten.embeddings.propose_clusters` with
    ``threshold=min_cohesion``, ``max_k=max_children``); the surviving clusters are
    NAMED by the agent into child facets through the SAME call/parse/slug seam as
    :func:`induce_dimensions` (``llm_fn`` + :func:`_build_induce_prompt` +
    :func:`_parse_json_array` + :func:`_facets_to_dimensions`), so child ``key``s
    are deterministic, disambiguated, and ``build_spine_skeleton``-shaped.

    Cohesion is the REQUIRED signal — with ``embed_index=None`` (no vectors) there
    is no basis to claim the column splits, so this returns ``[]`` rather than
    fabricating a subtree. A sufficient-clusters GATE keeps the column FLAT unless
    at least ``min_clusters`` clusters of at least ``min_cluster_size`` members
    survive; the ``unclustered`` bucket and sub-min clusters do NOT each become a
    child. ``max_depth`` bounds recursion: each named child's own cluster is the
    corpus for its depth pass (``max_depth - 1``), so a child carries its own
    ``children`` only while depth remains. Returns ``[]`` when the gate fails, the
    corpus is empty, ``llm_fn`` is missing, or ``max_depth <= 0``.

    Determinism: :func:`zettelkasten.embeddings.propose_clusters` is deterministic
    and the text→slug mapping is deterministic given identical facet text, so
    identical inputs yield identical child ``key``s. As in Stage 3, the agent's
    NAMING itself is not stabilized — a re-run can emit different child titles and
    therefore different slugs (a known limitation).
    """
    if int(max_depth) <= 0 or embed_index is None or llm_fn is None:
        return []

    item_by_id: dict[str, Any] = {}
    seen_ids: set[str] = set()
    views: list[_NoteView] = []
    for item in cell_notes:
        v = _as_view(item)
        if v is None or v.note_id in seen_ids:
            continue
        seen_ids.add(v.note_id)
        views.append(v)
        item_by_id[v.note_id] = item
    if not views:
        return []

    from zettelkasten.embeddings import propose_clusters

    vectors = {v.note_id: _safe_get_vector(embed_index, v.note_id) for v in views}
    if not any(vectors.values()):
        # No usable vectors → no cohesion signal → do NOT fabricate a subtree.
        return []

    clustered = propose_clusters(
        vectors, threshold=min_cohesion, max_k=max(1, int(max_children))
    )
    # Sufficient-clusters GATE: only clusters at/above the size floor count, and we
    # need at least ``min_clusters`` of them — otherwise the column stays FLAT.
    clusters = [
        c for c in (clustered.get("clusters") or []) if len(c) >= max(1, int(min_cluster_size))
    ]
    if len(clusters) < max(1, int(min_clusters)):
        return []

    by_id = {v.note_id: v for v in views}

    # Name every surviving cluster in ONE agent call, reusing the Stage 3 induction
    # seam: the clusters are surfaced as candidate groupings (hints) over a digest
    # of the column's notes, with the parent facet folded into the intent.
    title = str(parent_facet.get("title") or parent_facet.get("key") or "").strip()
    desc = str(parent_facet.get("desc") or "").strip()
    intent = "Sub-divide the analytical facet" + (f" '{title}'" if title else "")
    if desc:
        intent += f" ({desc})"
    intent += (
        " into distinct child facets — name each candidate grouping below as one "
        "child sub-facet."
    )
    digest = _corpus_digest(views)
    # Each surviving cluster is LABELED with an explicit HANDLE in the prompt and
    # the agent ECHOES that handle per child facet, so a child maps to its OWN
    # cluster by the handle it named — NOT by the facet's reply position. This
    # removes the positional assumption entirely: a pure reorder of the naming
    # reply no longer re-targets which cluster a child's grandchildren are induced
    # from. Empty-hint clusters are dropped together with their hint (and handle)
    # so the handle↔cluster alignment stays exact.
    hint_pairs = [
        ([by_id[nid].title for nid in cluster if nid in by_id][:5], cluster)
        for cluster in clusters
    ]
    hint_pairs = [(h, c) for h, c in hint_pairs if h]
    hints = [h for h, _ in hint_pairs]
    hint_clusters = [c for _, c in hint_pairs]
    handles = [_cluster_handle(i) for i in range(len(hint_clusters))]
    handle_to_cluster = {h: c for h, c in zip(handles, hint_clusters)}
    prompt = _build_subfacet_prompt(intent, digest, handles, hints)
    try:
        raw = llm_fn(_INDUCE_CONTRACT, prompt) or ""
    except Exception as exc:  # noqa: BLE001 — a naming hiccup degrades to a flat column
        logger.warning("remine induce_subfacets call failed: %s", exc)
        raw = ""

    raw_facets = _parse_json_array(raw)
    children, src_indices = _facets_to_dimensions_indexed(
        raw_facets, max_cols=max(1, int(max_children))
    )
    if not children:
        return []

    # Read each surviving child's echoed cluster HANDLE off its SOURCE facet
    # (``src_indices`` threads through ``_facets_to_dimensions``' dedup /
    # disambiguation / truncation, so it points at the raw facet the child came
    # from). A handle echoed by more than one facet is AMBIGUOUS — it maps to no
    # cluster. The prompt DEMANDS a handle per child, so reply position is NEVER
    # trusted: a missing/unknown/ambiguous handle — and a reply that echoes NO
    # usable handle AT ALL — yields NO grandchildren for that facet (safe) rather
    # than re-introducing the positional reorder bug.
    child_handles: list[str] = []
    handle_counts: dict[str, int] = {}
    for src_i in src_indices:
        rf = raw_facets[src_i] if 0 <= src_i < len(raw_facets) else None
        h = str(rf.get("cluster") or "").strip().upper() if isinstance(rf, dict) else ""
        child_handles.append(h)
        if h:
            handle_counts[h] = handle_counts.get(h, 0) + 1

    def _cluster_for(idx: int) -> "list[str] | None":
        # Map a child to its cluster ONLY by the unique handle it echoed; a
        # missing, unknown, or duplicate (ambiguous) handle → no cluster → no
        # grandchildren. Reply position is never used as a fallback.
        h = child_handles[idx]
        if h and handle_counts.get(h) == 1:
            return handle_to_cluster.get(h)
        return None

    # Recurse: each named child's OWN cluster (resolved by handle above) is the
    # corpus for its depth pass, bounded by ``max_depth``.
    if int(max_depth) > 1:
        for idx, child in enumerate(children):
            cluster = _cluster_for(idx)
            if not cluster:
                continue
            grand_notes = [item_by_id[nid] for nid in cluster if nid in item_by_id]
            grandchildren = induce_subfacets(
                grand_notes,
                child,
                llm_fn=llm_fn,
                embed_index=embed_index,
                min_clusters=min_clusters,
                min_cluster_size=min_cluster_size,
                min_cohesion=min_cohesion,
                max_children=max_children,
                max_depth=int(max_depth) - 1,
            )
            if grandchildren:
                child["children"] = grandchildren

    return children

build_scope_embed_index

build_scope_embed_index(get_graph: Any, *, project: str = '', graph: str = '', graphs_dir: Any = None, localize: 'Callable[[str], str] | None' = None) -> _ScopeEmbeddingIndex

Build a LAZY scope-spanning embedding index for the Stage 4 depth pass.

The production PROPOSE entry points (the remine MCP tool and the dashboard PROPOSE route) call this on the EMERGENT (intent) path ONLY and ONLY when depth >= 1 so an induced column's cell notes can actually be sub-clustered into children. It reuses each scope graph's own per-box :class:~zettelkasten.embeddings.EmbeddingIndex (the existing mechanism); see :class:_ScopeEmbeddingIndex for the laziness contract. project/graph name the owner scope exactly as :func:propose_remine partitions it; pass the same localize the route uses so ref-form graph names resolve.

Source code in zettelkasten/remine.py

def build_scope_embed_index(
    get_graph: Any,
    *,
    project: str = "",
    graph: str = "",
    graphs_dir: Any = None,
    localize: "Callable[[str], str] | None" = None,
) -> _ScopeEmbeddingIndex:
    """Build a LAZY scope-spanning embedding index for the Stage 4 depth pass.

    The production PROPOSE entry points (the ``remine`` MCP tool and the dashboard
    PROPOSE route) call this on the EMERGENT (intent) path ONLY and ONLY when
    ``depth >= 1`` so an induced column's cell notes can actually be sub-clustered
    into ``children``. It reuses each scope graph's own per-box
    :class:`~zettelkasten.embeddings.EmbeddingIndex` (the existing mechanism); see
    :class:`_ScopeEmbeddingIndex` for the laziness contract. ``project``/``graph``
    name the owner scope exactly as :func:`propose_remine` partitions it; pass the
    same ``localize`` the route uses so ref-form graph names resolve.
    """
    return _ScopeEmbeddingIndex(
        get_graph,
        project=project,
        graph=graph,
        graphs_dir=graphs_dir,
        localize=localize,
    )

read_prior_members

read_prior_members(org: 'dict[str, Any] | None', get_graph: Any) -> dict[str, set[str]]

The prior dimension-node membership of a PROMOTED spine, keyed by column.

For an org already materialized into a spine, each materializable column carries its dimension node id (apex) and that node's synthesis graph (graph); the node's outgoing spine-member edges name the notes routed under it (dim --spine-member--> note@home_graph). This returns {column_key: {member note id, ...}} — the prior map :func:reconcile_ids matches induced facets against by MEMBER OVERLAP so a re-mine reuses the durable column key/node id instead of minting a fresh slug. Member ids are bare note ids, the SAME uid the rest of re-mining uses (cell note_id).

Returns {} for a not-yet-promoted org (lens only — no apex/graph), an unknown org, or any unreadable spine graph → the caller then treats the re-mine as GREENFIELD (every key minted in order, byte-identical to today).

Source code in zettelkasten/remine.py

def read_prior_members(
    org: "dict[str, Any] | None",
    get_graph: Any,
) -> dict[str, set[str]]:
    """The prior dimension-node membership of a PROMOTED spine, keyed by column.

    For an org already materialized into a spine, each materializable column
    carries its dimension node id (``apex``) and that node's synthesis graph
    (``graph``); the node's outgoing ``spine-member`` edges name the notes routed
    under it (``dim --spine-member--> note@home_graph``). This returns
    ``{column_key: {member note id, ...}}`` — the prior map :func:`reconcile_ids`
    matches induced facets against by MEMBER OVERLAP so a re-mine reuses the
    durable column key/node id instead of minting a fresh slug. Member ids are bare
    note ids, the SAME uid the rest of re-mining uses (cell ``note_id``).

    Returns ``{}`` for a not-yet-promoted org (lens only — no apex/graph), an
    unknown org, or any unreadable spine graph → the caller then treats the re-mine
    as GREENFIELD (every key minted in order, byte-identical to today).
    """
    from zettelkasten.spine import MEMBERSHIP_RELATION

    out: dict[str, set[str]] = {}
    if not isinstance(org, dict):
        return out
    lens = org.get("lens_definition") if isinstance(org.get("lens_definition"), dict) else None
    columns = (lens or {}).get("columns") or org.get("columns") or []
    for col in columns:
        if not isinstance(col, dict):
            continue
        key = str(col.get("key") or "").strip()
        node_id = str(col.get("apex") or "").strip()
        gname = str(col.get("graph") or "").strip()
        if not key or not node_id or not gname:
            continue
        try:
            zg = get_graph(gname)
        except Exception as exc:  # noqa: BLE001 — an unreadable spine degrades to greenfield
            logger.debug("remine prior-member read failed for graph '%s': %s", gname, exc)
            continue
        node = zg.notes.get(node_id)
        if node is None:
            continue
        members = {
            str(getattr(link, "target", "") or "")
            for link in (node.links or [])
            if str(getattr(link, "relation", "") or "") == MEMBERSHIP_RELATION
            and str(getattr(link, "target", "") or "")
        }
        if members:
            out[key] = members
    return out

propose_remine

propose_remine(get_graph: Any, *, dimensions: 'list[dict[str, Any]] | None' = None, schema: str = '', columns: 'list[dict[str, Any]] | None' = None, project: str = '', graph: str = '', name: str = '', row_axis: 'dict[str, Any] | None' = None, graphs_dir: Any = None, localize: 'Callable[[str], str] | None' = None, extract_fn: 'Callable[[str, str], str] | None' = None, classify_fn: 'Callable[[Any, dict[str, Any]], list[dict[str, Any]]] | None' = None, embed_index: Any = None, similarity_threshold: float = DEFAULT_SIMILARITY_THRESHOLD, intent: str = '', induced_dimensions: 'list[dict[str, Any]] | None' = None, prior_members: 'dict[str, set[str]] | None' = None, max_cols: int = 8, depth: int = 0, title: str = '', table_id: str = 'remine-proposal') -> dict[str, Any]

PROPOSE phase: build a re-mined grid + residual bucket. NO writes.

Gathers one cell per dimension over the row-axis scope with the agent-semantic classifier enabled, so each cell carries its deterministic members PLUS the classifier's agent-inferred members (provenance='reviewer-inferred' with confidence/verified/quote). The returned payload mirrors a :func:zettelkasten.tables.build_matrix result so it renders through the identical matrix UI, and additionally carries residual (a.k.a. unassigned) — the in-scope notes that landed in NO dimension and were not already deterministically tagged, surfaced rather than silently dropped.

Column resolution, most-fixed first:

1. An explicit dimensions list (classifier descriptors key/tag/title/desc) is used verbatim.
2. A registered schema expands to its fixed dimensions via :func:expand_dimensions (the UNCHANGED default path).
3. With NEITHER, an intent ("make a matrix for X") and/or an explicit induced_dimensions list INDUCES the columns from the row corpus via :func:induce_dimensions (Stage 3 facet induction). Induced columns are flagged induced=True so the UI can show they were synthesized.

columns default to schema_tag columns derived from the dimensions. The classifier is built from extract_fn (the LLM seam — production defaults to the in-process agent; tests inject a fake) via :func:build_classify_fn, or a fully-formed classify_fn can be injected directly. NOTHING is persisted or attached — the caller must explicitly APPLY.

prior_members ({column_key: {member note id, ...}}, e.g. from :func:read_prior_members) is the EXISTING spine's dimension-node membership. On the induced path it lets each induced facet REUSE the prior column key when its classified members overlap a prior dimension (via :func:reconcile_ids), so a re-mine over an existing spine does not churn column keys / orphan dimension nodes when the agent renames or reorders a facet. None/empty (no prior spine) is GREENFIELD — every key is the freshly-minted slug, byte- identical to the no-prior result.

depth (default 0 = OFF) opts into Stage 4 emergent DEPTH: with depth >= 1 AND an EMERGENT (induced) column set AND an embed_index, each induced column's classified cell notes are sub-clustered and, when cohesive, named into child facets via :func:induce_subfacets and attached as that column's children (recursing up to depth levels). The depth pass is READ-ONLY (no new writes) and NEVER touches an enforced/registry column. With depth == 0 the output is byte-identical to the Stage 3 result (no children key on any column).

Source code in zettelkasten/remine.py

def propose_remine(
    get_graph: Any,
    *,
    dimensions: "list[dict[str, Any]] | None" = None,
    schema: str = "",
    columns: "list[dict[str, Any]] | None" = None,
    project: str = "",
    graph: str = "",
    name: str = "",
    row_axis: "dict[str, Any] | None" = None,
    graphs_dir: Any = None,
    localize: "Callable[[str], str] | None" = None,
    extract_fn: "Callable[[str, str], str] | None" = None,
    classify_fn: "Callable[[Any, dict[str, Any]], list[dict[str, Any]]] | None" = None,
    embed_index: Any = None,
    similarity_threshold: float = DEFAULT_SIMILARITY_THRESHOLD,
    intent: str = "",
    induced_dimensions: "list[dict[str, Any]] | None" = None,
    prior_members: "dict[str, set[str]] | None" = None,
    max_cols: int = 8,
    depth: int = 0,
    title: str = "",
    table_id: str = "remine-proposal",
) -> dict[str, Any]:
    """PROPOSE phase: build a re-mined grid + residual bucket. NO writes.

    Gathers one cell per dimension over the row-axis scope with the agent-semantic
    classifier enabled, so each cell carries its deterministic members PLUS the
    classifier's agent-inferred members (``provenance='reviewer-inferred'`` with
    ``confidence``/``verified``/``quote``). The returned payload mirrors a
    :func:`zettelkasten.tables.build_matrix` result so it renders through the
    identical matrix UI, and additionally carries ``residual`` (a.k.a.
    ``unassigned``) — the in-scope notes that landed in NO dimension and were not
    already deterministically tagged, surfaced rather than silently dropped.

    Column resolution, most-fixed first:

    1. An explicit ``dimensions`` list (classifier descriptors
       ``key``/``tag``/``title``/``desc``) is used verbatim.
    2. A registered ``schema`` expands to its fixed dimensions via
       :func:`expand_dimensions` (the UNCHANGED default path).
    3. With NEITHER, an ``intent`` ("make a matrix for X") and/or an explicit
       ``induced_dimensions`` list INDUCES the columns from the row corpus via
       :func:`induce_dimensions` (Stage 3 facet induction). Induced columns are
       flagged ``induced=True`` so the UI can show they were synthesized.

    ``columns`` default to schema_tag columns derived from the dimensions. The
    classifier is built from ``extract_fn`` (the LLM seam — production defaults to
    the in-process agent; tests inject a fake) via :func:`build_classify_fn`, or a
    fully-formed ``classify_fn`` can be injected directly. NOTHING is persisted or
    attached — the caller must explicitly APPLY.

    ``prior_members`` (``{column_key: {member note id, ...}}``, e.g. from
    :func:`read_prior_members`) is the EXISTING spine's dimension-node membership.
    On the induced path it lets each induced facet REUSE the prior column key when
    its classified members overlap a prior dimension (via :func:`reconcile_ids`),
    so a re-mine over an existing spine does not churn column keys / orphan
    dimension nodes when the agent renames or reorders a facet. ``None``/empty (no
    prior spine) is GREENFIELD — every key is the freshly-minted slug, byte-
    identical to the no-prior result.

    ``depth`` (default ``0`` = OFF) opts into Stage 4 emergent DEPTH: with
    ``depth >= 1`` AND an EMERGENT (induced) column set AND an ``embed_index``, each
    induced column's classified cell notes are sub-clustered and, when cohesive,
    named into child facets via :func:`induce_subfacets` and attached as that
    column's ``children`` (recursing up to ``depth`` levels). The depth pass is
    READ-ONLY (no new writes) and NEVER touches an enforced/registry column. With
    ``depth == 0`` the output is byte-identical to the Stage 3 result (no
    ``children`` key on any column).
    """
    from zettelkasten import tables

    base = Path(graphs_dir) if graphs_dir is not None else tables.GRAPHS_DIR
    loc = localize or (lambda s: s)
    axis = tables.normalize_row_axis(row_axis)

    # Index the scope's notes by id so each residual note id can be resolved to a
    # ``ResidualNote`` (title + home graph). Built from the SAME partition
    # ``gather_rows`` uses below, so every id resolves. A scope build hiccup must
    # not fail the propose — the residual just degrades to bare ids. The scopes are
    # also the corpus that facet induction (below) reads.
    scopes: list[Any] = []
    note_index: dict[str, tuple[Any, Any]] = {}
    try:
        scopes = tables._build_row_scopes(
            get_graph, axis, project=project, graph=graph, base=base, loc=loc
        )
        for scope in scopes:
            for zg, note in getattr(scope, "notes", []) or []:
                nid = str(getattr(note, "id", "") or "")
                if nid:
                    note_index.setdefault(nid, (zg, note))
    except Exception as exc:  # noqa: BLE001 — residual resolution is best-effort
        logger.debug("remine propose scope index failed: %s", exc)

    # Resolve the column dimensions. A registry schema keeps the exact historical
    # path; only the no-schema/no-dimensions case opts into facet induction so
    # existing callers are byte-for-byte unaffected.
    induced = False
    if dimensions is None:
        if schema:
            dimensions = expand_dimensions(schema)
        elif induced_dimensions is not None or str(intent or "").strip():
            induced = True
            if induced_dimensions is not None:
                # Slug + de-dupe an explicitly-supplied facet list through the SAME
                # path the agent induction uses, so the keys are deterministic.
                dimensions = _facets_to_dimensions(induced_dimensions, max_cols=max_cols)
            else:
                scope_notes = [
                    pair for scope in scopes for pair in (getattr(scope, "notes", []) or [])
                ]
                dimensions = induce_dimensions(
                    scope_notes,
                    intent,
                    llm_fn=extract_fn,
                    embed_index=embed_index,
                    max_cols=max_cols,
                )
    dims = _normalize_remine_dimensions(dimensions or [])

    cols = tables.normalize_columns(
        columns if columns is not None else _columns_from_dimensions(dims)
    )
    if induced and columns is None:
        # Flag synthesized columns so the UI can mark them as agent-induced. Done
        # AFTER ``normalize_columns`` (which emits a fixed key set) so the flag
        # survives, and only on the induction path so registry columns are
        # untouched.
        for col in cols:
            col["induced"] = True

    if classify_fn is None:
        classify_fn = build_classify_fn(
            dims,
            extract_fn=extract_fn,
            embed_index=embed_index,
            similarity_threshold=similarity_threshold,
            schema=schema,
        )

    rows = tables.gather_rows(
        get_graph,
        cols,
        project=project,
        graph=graph,
        row_axis=axis,
        graphs_dir=base,
        localize=loc,
        classify_fn=classify_fn,
    )

    # COLUMN identity reconcile (induced path only): when a PRIOR spine exists for
    # this scope (``prior_members`` non-empty), match each induced facet to a prior
    # dimension node by MEMBER OVERLAP and REUSE the prior column key, so an agent
    # rename/reorder of a facet does not orphan the durable dimension node. Done
    # here — AFTER columns are built and rows classified (the only place an induced
    # facet's member set is known) — rather than in the greenfield-pure
    # ``induce_dimensions``. Greenfield (no prior) mints every key via the slug
    # logic in order → byte-identical to Stage 3.
    if induced and columns is None:
        _reconcile_induced_columns(cols, rows, dims, prior_members=prior_members or {})

    # Stage 4 emergent DEPTH (opt-in): only when columns are EMERGENT (induced),
    # an embed index is available, and ``depth >= 1``. Each induced column's
    # classified cell notes are sub-clustered and, when cohesive, named into child
    # facets attached as the column's ``children``. An enforced/registry column is
    # NEVER given a depth pass, and with ``depth == 0`` ``cols`` is left untouched
    # so the Stage 3 output is byte-identical.
    if induced and columns is None and int(depth) >= 1 and embed_index is not None:
        dim_by_key = {d["key"]: d for d in dims}
        for col in cols:
            if not col.get("induced"):
                continue
            key = str(col.get("key") or "")
            cell_ids: list[str] = []
            seen_cell: set[str] = set()
            for row in rows:
                cell = (row.get("cells") or {}).get(key) or {}
                for m in cell.get("members") or []:
                    nid = str(m.get("note_id") or "")
                    if nid and nid not in seen_cell:
                        seen_cell.add(nid)
                        cell_ids.append(nid)
            cell_notes = [note_index[nid] for nid in cell_ids if nid in note_index]
            parent_facet = dim_by_key.get(key) or {
                "key": key,
                "tag": key,
                "title": col.get("label") or key,
                "desc": "",
            }
            children = induce_subfacets(
                cell_notes,
                parent_facet,
                llm_fn=extract_fn,
                embed_index=embed_index,
                max_children=max_cols,
                max_depth=int(depth),
            )
            if children:
                col["children"] = children

    # Residual = in-scope notes that landed in NO cell (so neither deterministic
    # nor agent routing claimed them). Computed straight off the gathered grid, so
    # it is correct regardless of whether the classifier was built or injected.
    rows_by_id = {r["id"]: r for r in rows}
    residual: list[dict[str, Any]] = []
    seen: set[str] = set()
    for scope in scopes:
        row = rows_by_id.get(getattr(scope, "id", ""))
        assigned: set[str] = set()
        if row:
            for cell in (row.get("cells") or {}).values():
                for m in cell.get("members") or []:
                    assigned.add(str(m.get("note_id") or ""))
        for zg, note in getattr(scope, "notes", []) or []:
            nid = str(getattr(note, "id", "") or "")
            if not nid or nid in assigned or nid in seen:
                continue
            seen.add(nid)
            residual.append(_residual_note(zg, note))

    return {
        "table_id": table_id,
        "title": title,
        "columns": cols,
        "row_axis": axis,
        "rows": rows,
        "synthesis": {},
        "signature": "",
        "schema_version": 1,
        "spine_id": tables.DEFAULT_SPINE_ID,
        # Materialized at view-time and never persisted — the empty/Generate states
        # never apply, so it reads as "populated" for the matrix UI.
        "exists": True,
        "ephemeral": True,
        # The residual bucket, under BOTH keys the frontend may read.
        "residual": residual,
        "unassigned": residual,
    }

apply_remine

apply_remine(owner_type: str, owner_name: str, org_id: str, *, grid: 'dict[str, Any] | list[dict[str, Any]]', get_graph: Any, graphs_dir: Any = None, tag_stamp: bool = False, attach_relation: 'str | None' = None, row_relation: 'str | None' = None, link_relation: 'str | None' = None) -> dict[str, Any]

APPLY phase: materialize an approved proposed grid via the attach seam.

Reuses :func:zettelkasten.organizations.promote_organization with proposed_grid= — the ONLY edge-writer — so the approved grid promotes through the identical scaffold/attach/reconcile machinery (membership is reconciled, never add-only, so a re-run prunes members no longer classified and adds new ones; idempotent). grid is the PROPOSE payload (its rows are used) or a bare rows list. tag_stamp (default OFF) is the one opt-in that mutates base notes. Returns an apply summary the routes/tool echo.

APPLY-GATE: induced (intent-proposed) grids are preview-only — their columns are synthesized from the corpus, not the org's fixed schema, so promoting one onto an org with a different schema is a corruption vector. The guard lives HERE, where the ROW cells are consumed, so it cannot be bypassed by stripping grid["columns"] (a flag-only check upstream can be): a grid carrying induced columns is refused, AND a grid whose row cells carry any key absent from the target org's FIXED schema is refused (this catches a columns-stripped bare-rows induced grid). A legitimate registry-schema grid, whose cell keys are exactly the org's column keys, applies unchanged.

Source code in zettelkasten/remine.py

def apply_remine(
    owner_type: str,
    owner_name: str,
    org_id: str,
    *,
    grid: "dict[str, Any] | list[dict[str, Any]]",
    get_graph: Any,
    graphs_dir: Any = None,
    tag_stamp: bool = False,
    attach_relation: "str | None" = None,
    row_relation: "str | None" = None,
    link_relation: "str | None" = None,
) -> dict[str, Any]:
    """APPLY phase: materialize an approved proposed ``grid`` via the attach seam.

    Reuses :func:`zettelkasten.organizations.promote_organization` with
    ``proposed_grid=`` — the ONLY edge-writer — so the approved grid promotes
    through the identical scaffold/attach/reconcile machinery (membership is
    reconciled, never add-only, so a re-run prunes members no longer classified
    and adds new ones; idempotent). ``grid`` is the PROPOSE payload (its ``rows``
    are used) or a bare rows list. ``tag_stamp`` (default OFF) is the one opt-in
    that mutates base notes. Returns an apply summary the routes/tool echo.

    APPLY-GATE: induced (intent-proposed) grids are preview-only — their columns
    are synthesized from the corpus, not the org's fixed schema, so promoting one
    onto an org with a different schema is a corruption vector. The guard lives
    HERE, where the ROW cells are consumed, so it cannot be bypassed by stripping
    ``grid["columns"]`` (a flag-only check upstream can be): a grid carrying
    ``induced`` columns is refused, AND a grid whose row cells carry any key absent
    from the target org's FIXED schema is refused (this catches a columns-stripped
    bare-rows induced grid). A legitimate registry-schema grid, whose cell keys are
    exactly the org's column keys, applies unchanged.
    """
    from zettelkasten import organizations, tables

    # Refuse an induced grid up front when the columns flag survived (defense in
    # depth alongside the cell-key check below, which catches a stripped grid).
    if isinstance(grid, dict) and any(
        isinstance(c, dict) and c.get("induced") for c in (grid.get("columns") or [])
    ):
        raise ValueError(
            "apply refuses an induced (intent-proposed) grid — induced columns are "
            "preview-only and not promotable."
        )

    rows = list((grid.get("rows") if isinstance(grid, dict) else grid) or [])

    # The grid's row cell-keys MUST be a subset of the org's FIXED schema column
    # keys — otherwise the rows describe an induced/foreign structure that was
    # never the org's, so refuse before any edge is written.
    org = organizations.load_organization(
        owner_type, owner_name, org_id, graphs_dir=graphs_dir, migrate=False
    )
    if org is None:
        raise ValueError(f"organization '{org_id}' not found for {owner_type} '{owner_name}'")
    lens = org.get("lens_definition") if isinstance(org.get("lens_definition"), dict) else None
    allowed_keys = {
        str(c.get("key") or "").strip()
        for c in tables.normalize_columns((lens or {}).get("columns") or org.get("columns") or [])
    }
    for row in rows:
        cells = row.get("cells") if isinstance(row, dict) else None
        if not isinstance(cells, dict):
            continue
        for cell_key in cells:
            if str(cell_key).strip() not in allowed_keys:
                raise ValueError(
                    "apply refuses a grid whose row cells carry keys absent from the "
                    "org's fixed schema — induced/preview grids are not promotable."
                )

    result = organizations.promote_organization(
        owner_type,
        owner_name,
        org_id,
        get_graph=get_graph,
        graphs_dir=graphs_dir,
        proposed_grid=rows,
        tag_stamp=tag_stamp,
        attach_relation=attach_relation,
        row_relation=row_relation,
        link_relation=link_relation,
    )

    # Strong usage signal: re-mining reclassifies/attaches these notes — an
    # explicit synthesis action — so boost their access-recency (weight 3.0).
    # Best-effort and never fatal: a usage-DB hiccup must not fail an apply.
    try:
        from zettelkasten import usage

        for m in result.get("attached_members") or []:
            g = str(m.get("graph") or "")
            nid = str(m.get("note_id") or "")
            if g and nid:
                usage.record_access(g, nid, weight=3.0)
    except Exception:
        pass

    return {
        "applied": True,
        "edges_written": int(result.get("attached_edges") or 0),
        "spine_ref": result.get("spine_ref", ""),
        "dimension_nodes": result.get("dimension_nodes", {}),
        "org_id": org_id,
    }

resolve_source_scope

resolve_source_scope(org: dict[str, Any], get_graph: Any, base: Any) -> list[str]

The graphs whose notes are this spine's classifier scope (source→persona).

Resolution, most-precise first:

1. The spine apex's per-source rollup edges (:func:_apex_source_graphs) — the persona's OWN source corpora, even when the project hosts several personas over different subsets.
2. The owner scope: a graph-owned org classifies its own graph; a project-owned org classifies the project's registered sources (framing.frame_search_sources), minus the spine graph itself and the _ system graphs (e.g. _cross).

Never includes the spine graph itself (its nodes are structure, not evidence).

Source code in zettelkasten/remine.py

def resolve_source_scope(org: dict[str, Any], get_graph: Any, base: Any) -> list[str]:
    """The graphs whose notes are this spine's classifier scope (source→persona).

    Resolution, most-precise first:

    1. The spine apex's per-source rollup edges (:func:`_apex_source_graphs`) —
       the persona's OWN source corpora, even when the project hosts several
       personas over different subsets.
    2. The owner scope: a graph-owned org classifies its own graph; a
       project-owned org classifies the project's registered sources
       (``framing.frame_search_sources``), minus the spine graph itself and the
       ``_`` system graphs (e.g. ``_cross``).

    Never includes the spine graph itself (its nodes are structure, not evidence).
    """
    spine_ref = str(org.get("spine_ref") or "").strip()
    if spine_ref:
        rollup = _apex_source_graphs(get_graph, spine_ref)
        if rollup:
            return rollup

    owner = org.get("owner") or {}
    ot = str(owner.get("type") or "")
    on = str(owner.get("name") or "")
    if ot == "graph" and on:
        return [on] if on != spine_ref else []
    if ot == "project" and on:
        from zettelkasten.framing import frame_search_sources

        return [
            s
            for s in frame_search_sources(on, graphs_dir=base)
            if s and s != spine_ref and not str(s).startswith("_")
        ]
    return []

backfill_spine

backfill_spine(owner_type: str, owner_name: str, org_id: str, *, get_graph: Any, classify_fn: 'Callable[[Any, dict[str, Any]], list[dict[str, Any]]]', graphs_dir: Any = None, source_graphs: 'list[str] | None' = None, apply: bool = True, tag_stamp: bool = False, row_id: str = '', row_label: str = '') -> dict[str, Any]

Backfill evidence members onto a persona spine's FIXED dimension nodes.

The degenerate re-mine: rows + columns are already fixed (the persona and its reverse-engineered schema), so this runs classify_fn over the persona's OWN source-note scope (:func:resolve_source_scope) and attaches the results onto the EXISTING dimension nodes via the promote-attach path (promote_organization(..., proposed_grid=...)) — additive (a node's ported synthesized body is preserved), reconcile-based (a re-run prunes/adds), and idempotent.

classify_fn(scope, column) -> [{note_id, quote, confidence}] is the per-column seam (:func:build_classify_fn in production; a canned callable in tests). The classified note ids are mapped back to their HOME graph from the scope so the attach edge points at the right base note. With apply=False the proposed grid is returned WITHOUT writing (the propose-first preview).

Source code in zettelkasten/remine.py

def backfill_spine(
    owner_type: str,
    owner_name: str,
    org_id: str,
    *,
    get_graph: Any,
    classify_fn: "Callable[[Any, dict[str, Any]], list[dict[str, Any]]]",
    graphs_dir: Any = None,
    source_graphs: "list[str] | None" = None,
    apply: bool = True,
    tag_stamp: bool = False,
    row_id: str = "",
    row_label: str = "",
) -> dict[str, Any]:
    """Backfill evidence members onto a persona spine's FIXED dimension nodes.

    The degenerate re-mine: rows + columns are already fixed (the persona and its
    reverse-engineered schema), so this runs ``classify_fn`` over the persona's
    OWN source-note scope (:func:`resolve_source_scope`) and attaches the results
    onto the EXISTING dimension nodes via the promote-attach path
    (``promote_organization(..., proposed_grid=...)``) — additive (a node's ported
    synthesized body is preserved), reconcile-based (a re-run prunes/adds), and
    idempotent.

    ``classify_fn(scope, column) -> [{note_id, quote, confidence}]`` is the
    per-column seam (:func:`build_classify_fn` in production; a canned callable in
    tests). The classified note ids are mapped back to their HOME graph from the
    scope so the attach edge points at the right base note. With ``apply=False``
    the proposed grid is returned WITHOUT writing (the propose-first preview).
    """
    from pathlib import Path

    from zettelkasten import organizations as orgs
    from zettelkasten import tables
    from zettelkasten.graph import GRAPHS_DIR

    base = Path(graphs_dir) if graphs_dir is not None else GRAPHS_DIR
    org = orgs.load_organization(owner_type, owner_name, org_id, graphs_dir=base, migrate=False)
    if org is None:
        raise ValueError(f"organization '{org_id}' not found for {owner_type} '{owner_name}'")

    # FIXED columns. On a promoted spine the live columns are in link-form, so the
    # durable schema lives in the stashed ``lens_definition`` (its keys back the
    # dimension node ids); fall back to the live columns for a not-yet-promoted org.
    lens = org.get("lens_definition") if isinstance(org.get("lens_definition"), dict) else None
    columns = (lens or {}).get("columns") or org.get("columns") or []
    # The classify-into dimensions are the MATERIALIZABLE member-bearing columns —
    # the exact set ``_promote_locked`` will key dimension nodes on.
    member_cols = [
        c
        for c, _tag in tables.derive_dimension_tags(columns)
        if str(c.get("backing") or "") in orgs._MEMBER_BEARING_BACKINGS
    ]

    srcs = (
        list(source_graphs)
        if source_graphs is not None
        else resolve_source_scope(org, get_graph, base)
    )

    # Build the persona's source-note scope + a note_id → (home graph, note) map so
    # a classified id can be shaped into a member edge pointing at its base note.
    scope_notes: list[Any] = []
    pair_by_id: dict[str, tuple[Any, Any]] = {}
    for gname in srcs:
        try:
            zg = get_graph(gname)
        except Exception:  # noqa: BLE001 — a missing source graph is skipped, not fatal
            logger.warning("backfill: could not load source graph '%s'", gname, exc_info=True)
            continue
        for note in zg.notes.values():
            nid = str(getattr(note, "id", "") or "")
            if not nid:
                continue
            scope_notes.append((zg, note))
            pair_by_id.setdefault(nid, (zg, note))

    scope = _BackfillScope(
        id=row_id or org_id,
        label=row_label or str(org.get("title") or org_id),
        notes=scope_notes,
    )

    # Classify per FIXED column; shape each assignment into a reviewer-inferred
    # cell member through the SAME builder the propose/apply path uses
    # (:func:`tables._agent_member`), so ``verified`` (and every other member
    # field) is computed by the one verbatim rule — the two materialization paths
    # cannot drift. A classified id absent from the scope (no resolvable home
    # graph/note) is dropped rather than attached blind.
    cells: dict[str, Any] = {}
    member_count = 0
    for col in member_cols:
        key = str(col.get("key") or "")
        members: list[dict[str, Any]] = []
        for entry in classify_fn(scope, col) or []:
            nid = str(entry.get("note_id") or "")
            pair = pair_by_id.get(nid)
            if not nid or pair is None:
                continue
            zg, note = pair
            members.append(
                tables._agent_member(
                    zg,
                    note,
                    quote=entry.get("quote"),
                    confidence=entry.get("confidence"),
                )
            )
        cells[key] = {"members": members, "summary": "", "value": ""}
        member_count += len(members)

    proposed_grid = [{"id": scope.id, "label": scope.label, "cells": cells}]

    if not apply:
        return {
            "applied": False,
            "source_graphs": srcs,
            "member_count": member_count,
            "proposed_grid": proposed_grid,
        }

    result = orgs.promote_organization(
        owner_type,
        owner_name,
        org_id,
        get_graph=get_graph,
        graphs_dir=base,
        proposed_grid=proposed_grid,
        tag_stamp=tag_stamp,
    )
    result["applied"] = True
    result["source_graphs"] = srcs
    result["member_count"] = member_count
    return result