zettelkasten.opportunities

zettelkasten.opportunities

Negative-space opportunity finder — the six PROPOSE-ONLY gap families.

Where :func:zettelkasten.claims.analyze_gaps surfaces weaknesses attached to existing claims (evidential / dialectical / structural / temporal / coverage / comprehensiveness), this module scans the NEGATIVE SPACE — the missing edges and unmatched nodes across the knowledge graph and across the memory↔zettelkasten boundary. It adds six new gap families as new gap_type strings on the SAME engine:

• asymmetry — a practice memory-node with no aligned canon claim, or a canon claim with no aligned practice node (a two-sided family read off the cross-store claim overlay).
• bridge — two dense clusters with (near-)zero connecting edges but high inter-cluster embedding similarity (a synthesis opportunity).
• crux — an unresolved central debate: a contested camp from the debate map, ranked by strength × centrality.
• orphaned_question — a question note with no incoming answering edge (no supports / responds-to backlink).
• void — a sparse-but-surrounded region of embedding space (an unasked question / underexplored area amid density).
• transfer — a method/model note applied within one thematic cluster that structurally fits an adjacent cluster where it has not been tried.

Every detector is a PURE function of an already-built :class:~zettelkasten.claims.ClaimIndex plus explicit inputs (clusters, vectors, the debate map, the cross-store overlay) — it returns list[dict] of gaps built via :func:zettelkasten.claims._gap, and NEVER writes to any store. Each gap's action only DESCRIBES the suggested next step (the same propose-only posture the rest of the claim engine keeps). Optional model/embedder dependencies are injected as parameters (default None) so tests run deterministically without a model.

This is a LEAF module: nothing in the package imports it at load time. :func:zettelkasten.claims.analyze_gaps does a LAZY from zettelkasten import opportunities inside its body only when one of the six families is requested, so claims.py never imports synapse (which imports claims) at module load and the circular import is avoided.

find_asymmetry

find_asymmetry(index: ClaimIndex, *, overlay: 'dict[str, Any] | None', memory_source: Any, kc_by_key: 'dict[Key, Any] | None' = None, max_sal: float = 1.0, practice_types: 'tuple[str, ...] | None' = None, practice_salience: float = ASYMMETRY_PRACTICE_SALIENCE) -> 'list[dict[str, Any]]'

Two-sided cross-store asymmetry: unmatched practice ↔ unmatched canon.

Reads the claim-aligned cross-store overlay (memory --relation--> zk edges). A practice memory-node is unmatched when its id never appears as an overlay memory_id; a canon claim is unmatched when its (source, id) never appears as an overlay (zk_source, zk_id) endpoint. Each gap's anchor identifies the unmatched node and which side is missing.

Practice nodes are restricted to the overlay's own practice registers (practice_types) so ordinary annotations/notes — which were never connection candidates — do not flood the report. This is a TWO-SIDED family, so it returns [] when EITHER cross-store side is unavailable: an absent/empty overlay (no practice↔canon edges) OR a missing memory_source (the practice side cannot be read). Never crashes.

Source code in zettelkasten/opportunities.py

def find_asymmetry(
    index: ClaimIndex,
    *,
    overlay: "dict[str, Any] | None",
    memory_source: Any,
    kc_by_key: "dict[Key, Any] | None" = None,
    max_sal: float = 1.0,
    practice_types: "tuple[str, ...] | None" = None,
    practice_salience: float = ASYMMETRY_PRACTICE_SALIENCE,
) -> "list[dict[str, Any]]":
    """Two-sided cross-store asymmetry: unmatched practice ↔ unmatched canon.

    Reads the claim-aligned cross-store overlay (``memory --relation--> zk``
    edges). A practice memory-node is unmatched when its id never appears as an
    overlay ``memory_id``; a canon claim is unmatched when its ``(source, id)``
    never appears as an overlay ``(zk_source, zk_id)`` endpoint. Each gap's anchor
    identifies the unmatched node and which side is missing.

    Practice nodes are restricted to the overlay's own practice registers
    (``practice_types``) so ordinary annotations/notes — which were never
    connection candidates — do not flood the report. This is a TWO-SIDED family,
    so it returns ``[]`` when EITHER cross-store side is unavailable: an
    absent/empty overlay (no practice↔canon edges) OR a missing ``memory_source``
    (the practice side cannot be read). Never crashes.
    """
    if memory_source is None:
        return []
    edges = (overlay or {}).get("edges", []) if overlay else []
    if not edges:
        return []
    if practice_types is None:
        try:
            from zettelkasten.synapse.synthesis import PRACTICE_TYPES as _PT
            practice_types = _PT
        except Exception:
            practice_types = ("decision", "experiment", "checkpoint")

    linked_memory: set[str] = {
        str(e.get("memory_id")) for e in edges if e.get("memory_id")
    }
    linked_canon: set[Key] = {
        (str(e.get("zk_source")), str(e.get("zk_id")))
        for e in edges
        if e.get("zk_source") and e.get("zk_id")
    }

    gaps: list[dict[str, Any]] = []

    # ── practice memory-nodes with no aligned canon claim ──────────────────────
    notes = getattr(memory_source, "notes", {}) or {}
    for mid, mnote in notes.items():
        mtype = getattr(mnote, "type", "") or ""
        if practice_types and mtype not in practice_types:
            continue
        if str(mid) in linked_memory:
            continue
        title = getattr(mnote, "title", "") or str(mid)
        anchor = {
            "id": str(mid),
            "graph": _claims._MEMORY_STORE,
            "title": title,
            "type": mtype or "practice",
            "store": "memory",
            "side": "practice",
            "missing": "canon",
        }
        gaps.append(_claims._gap(
            ASYMMETRY, salience=practice_salience, action_tool="synapse",
            action={"tool": "claim", "action": "test", "args": {"text": title},
                    "note": ("Practice entry has no aligned canon claim; test it as "
                             "a hypothesis or materialize it.")},
            predicate="unaligned practice (no canon claim)",
            detail=f"memory entry {mid} ({mtype or 'practice'}) has no claim-overlay edge",
            anchor=anchor,
        ))

    # ── canon claims with no aligned practice node ─────────────────────────────
    for key in index.claims:
        if key in linked_canon:
            continue
        kc = (kc_by_key or {}).get(key)
        norm_sal = (kc.salience / max_sal) if (kc and max_sal) else 0.0
        anchor = _claims._claim_ref(index, key)
        anchor["side"] = "canon"
        anchor["missing"] = "practice"
        gaps.append(_claims._gap(
            ASYMMETRY, salience=norm_sal, action_tool="synapse",
            action={"tool": "synapse", "action": "build_claim_connections",
                    "args": {"zk_source": key[0], "zk_id": key[1]},
                    "note": ("Canon claim has no aligned practice; connect or test it "
                             "against the memory tree.")},
            predicate="unaligned canon (no practice node)",
            detail=f"claim {key[0]}/{key[1]} has no claim-overlay edge",
            anchor=anchor,
        ))
    return gaps

find_bridges

find_bridges(index: ClaimIndex, *, clusters: 'list[list[Key]]', vectors: 'dict[Key, list[float]]', centrality: 'dict[Key, float]', min_similarity: float = BRIDGE_MIN_SIMILARITY, max_cross_edges: int = BRIDGE_MAX_CROSS_EDGES, min_cluster_size: int = BRIDGE_MIN_CLUSTER_SIZE, hub_max_fanout: int = BRIDGE_HUB_MAX_FANOUT, hub_min_shared_hubs: int = BRIDGE_MIN_SHARED_HUBS) -> 'list[dict[str, Any]]'

Pairs of dense clusters with few connecting edges but similar centroids.

A synthesis opportunity: two thematically-close regions the corpus has never linked. The anchor carries a representative from EACH cluster so the frontend can draw a ghost edge. O(k²) over the (small) number of clusters. Returns [] when fewer than two clusters exist.

"Unconnected" means unconnected DIRECTLY and INDIRECTLY: a pair linked only through a shared excluded hub note (an evidence/organizational note that is never clustered) is treated as connected — those clusters already share a one-hop path, so they are not a genuine synthesis gap. Hub suppression tracks real connectivity via hub_max_fanout (a focused hub suppresses on its own) and hub_min_shared_hubs (a diffuse high-degree hub suppresses only when corroborated), so a single mega-hub cannot wipe every bridge it touches yet genuinely co-shared pairs are still suppressed (see _hub_connected_pairs).

Source code in zettelkasten/opportunities.py

def find_bridges(
    index: ClaimIndex,
    *,
    clusters: "list[list[Key]]",
    vectors: "dict[Key, list[float]]",
    centrality: "dict[Key, float]",
    min_similarity: float = BRIDGE_MIN_SIMILARITY,
    max_cross_edges: int = BRIDGE_MAX_CROSS_EDGES,
    min_cluster_size: int = BRIDGE_MIN_CLUSTER_SIZE,
    hub_max_fanout: int = BRIDGE_HUB_MAX_FANOUT,
    hub_min_shared_hubs: int = BRIDGE_MIN_SHARED_HUBS,
) -> "list[dict[str, Any]]":
    """Pairs of dense clusters with few connecting edges but similar centroids.

    A synthesis opportunity: two thematically-close regions the corpus has never
    linked. The anchor carries a representative from EACH cluster so the frontend
    can draw a ghost edge. O(k²) over the (small) number of clusters. Returns
    ``[]`` when fewer than two clusters exist.

    "Unconnected" means unconnected DIRECTLY *and* INDIRECTLY: a pair linked only
    through a shared excluded hub note (an evidence/organizational note that is
    never clustered) is treated as connected — those clusters already share a
    one-hop path, so they are not a genuine synthesis gap. Hub suppression tracks
    real connectivity via ``hub_max_fanout`` (a focused hub suppresses on its own)
    and ``hub_min_shared_hubs`` (a diffuse high-degree hub suppresses only when
    corroborated), so a single mega-hub cannot wipe every bridge it touches yet
    genuinely co-shared pairs are still suppressed (see ``_hub_connected_pairs``).
    """
    if len(clusters) < 2:
        return []
    sims = _cluster_pair_sims(clusters, vectors)
    cross = _cross_edge_counts(index, clusters)
    hub_connected = _hub_connected_pairs(
        index, clusters, max_fanout=hub_max_fanout, min_shared_hubs=hub_min_shared_hubs,
    )
    gaps: list[dict[str, Any]] = []
    for (i, j), sim in sims.items():
        if len(clusters[i]) < min_cluster_size or len(clusters[j]) < min_cluster_size:
            continue
        if sim < min_similarity:
            continue
        if cross.get((i, j), 0) > max_cross_edges:
            continue
        if (i, j) in hub_connected:
            continue
        rep_a = _representative(clusters[i], centrality, index)
        rep_b = _representative(clusters[j], centrality, index)
        anchor = dict(rep_a)
        anchor["target"] = rep_b
        anchor["similarity"] = round(sim, 4)
        anchor["cluster_a_size"] = len(clusters[i])
        anchor["cluster_b_size"] = len(clusters[j])
        gaps.append(_claims._gap(
            BRIDGE, salience=max(0.0, min(1.0, sim)), action_tool="synthesize",
            action={"tool": "synapse", "action": "synthesize",
                    "args": {"a": rep_a.get("id"), "a_graph": rep_a.get("graph"),
                             "b": rep_b.get("id"), "b_graph": rep_b.get("graph")},
                    "note": "Two similar clusters are unconnected — synthesize a bridge."},
            predicate="unconnected but similar clusters (synthesis opportunity)",
            detail=(f"similarity={round(sim, 3)}, connecting_edges="
                    f"{cross.get((i, j), 0)}, sizes={len(clusters[i])}/{len(clusters[j])}"),
            anchor=anchor,
        ))
    return gaps

find_crux

find_crux(index: ClaimIndex, *, debate: 'dict[str, Any] | None', centrality: 'dict[Key, float]', importance_map: 'dict[str, float] | None' = None) -> 'list[dict[str, Any]]'

Unresolved central debates: contested camps ranked by strength × centrality.

Reads the camps from :func:zettelkasten.claims.debate_map and keeps only the contested ones (a live, unresolved debate). Each camp is scored so a strong, load-bearing debate is the crux worth resolving. The action suggests resolving via a supersedes decision. Returns [] when there are no contested camps.

Centrality is counted EXACTLY ONCE, as a single per-camp GLOBAL factor:

• within_strength — the mean of the camp members' claim_strength (how well-evidenced the debate is), with NO centrality weighting, so a genuinely central debate is not attenuated by its own members' centrality.
• global_factor — the camp's most-central member over the global max centrality across ALL claims. A central debate keeps ~1.0; a globally-peripheral lone camp is scaled toward 0 so it cannot inflate to top severity even when its members are internally strong.

Salience is within_strength × global_factor. (A prior version ALSO weighted each member's strength by a within-set–normalized centrality, so centrality entered twice — a ~centrality² attenuation that under-scored a debate central among the contested set but globally peripheral. Folding centrality into the single global factor counts it once.) Salience stays in [0, 1]; the div-by-zero and all-equal guards are preserved (an empty or all-zero centrality map yields a 0 factor, never a crash).

Source code in zettelkasten/opportunities.py

def find_crux(
    index: ClaimIndex,
    *,
    debate: "dict[str, Any] | None",
    centrality: "dict[Key, float]",
    importance_map: "dict[str, float] | None" = None,
) -> "list[dict[str, Any]]":
    """Unresolved central debates: contested camps ranked by strength × centrality.

    Reads the camps from :func:`zettelkasten.claims.debate_map` and keeps only the
    ``contested`` ones (a live, unresolved debate). Each camp is scored so a
    strong, load-bearing debate is the crux worth resolving. The action suggests
    resolving via a supersedes decision. Returns ``[]`` when there are no
    contested camps.

    Centrality is counted EXACTLY ONCE, as a single per-camp GLOBAL factor:

    * ``within_strength`` — the mean of the camp members' ``claim_strength`` (how
      well-evidenced the debate is), with NO centrality weighting, so a genuinely
      central debate is not attenuated by its own members' centrality.
    * ``global_factor`` — the camp's most-central member over the global max
      centrality across ALL claims. A central debate keeps ~1.0; a
      globally-peripheral lone camp is scaled toward 0 so it cannot inflate to top
      severity even when its members are internally strong.

    Salience is ``within_strength × global_factor``. (A prior version ALSO weighted
    each member's strength by a within-set–normalized centrality, so centrality
    entered twice — a ~centrality² attenuation that under-scored a debate central
    among the contested set but globally peripheral. Folding centrality into the
    single global factor counts it once.) Salience stays in [0, 1]; the
    div-by-zero and all-equal guards are preserved (an empty or all-zero centrality
    map yields a 0 factor, never a crash).
    """
    if not debate:
        return []

    # First pass: gather the in-index members of every contested camp.
    contested_camps: list[tuple[dict[str, Any], list[Key]]] = []
    for camp in debate.get("camps", []):
        if camp.get("status") != "contested":
            continue
        member_keys: list[Key] = []
        for uid in camp.get("members", []):
            graph, sep, nid = str(uid).partition("::")
            if not sep:
                continue
            key = (graph, nid)
            if key in index.resolved:
                member_keys.append(key)
        if not member_keys:
            continue
        contested_camps.append((camp, member_keys))

    # The GLOBAL max centrality (over every claim) drives the peripheral-camp
    # down-weight below; guarded so an empty/all-zero map yields a 0 factor.
    global_max = max(centrality.values()) if centrality else 0.0

    gaps: list[dict[str, Any]] = []
    for camp, member_keys in contested_camps:
        scores: list[float] = []
        for key in member_keys:
            rc = index.resolved.get(key)
            if rc is None:
                continue
            scores.append(_claims.claim_strength(rc, importance_map=importance_map))
        # Within-set component: mean member strength ONLY (centrality is applied
        # once, via the global factor below — not here — to avoid double-counting).
        within_strength = (sum(scores) / len(scores)) if scores else 0.0
        # Down-weight a globally-peripheral camp: its most-central member relative
        # to the global max centrality. A central debate keeps ~1.0; a peripheral
        # lone camp is scaled toward 0 so it cannot inflate to top severity.
        camp_global = max((centrality.get(k, 0.0) for k in member_keys), default=0.0)
        global_factor = (camp_global / global_max) if global_max > 0 else 0.0
        salience = within_strength * global_factor
        rep = max(member_keys, key=lambda k: (centrality.get(k, 0.0), _claims.claim_uid(k)))
        anchor = _claims._claim_ref(index, rep)
        anchor["camp"] = camp.get("id")
        anchor["members"] = [_claims.claim_uid(k) for k in member_keys]
        anchor["size"] = len(member_keys)
        gaps.append(_claims._gap(
            CRUX, salience=max(0.0, min(1.0, salience)), action_tool="resolve",
            action={"tool": "find_supersessions", "action": "propose",
                    "args": {"graph": rep[0]},
                    "note": "Central contested debate — resolve via a supersedes decision."},
            predicate="unresolved central debate (contested camp)",
            detail=(f"camp size={len(member_keys)}, "
                    f"mean_strength={round(camp.get('mean_strength', 0.0), 3)}"),
            anchor=anchor,
        ))
    return gaps

find_orphaned_questions

find_orphaned_questions(index: ClaimIndex) -> 'list[dict[str, Any]]'

question notes with no incoming answering edge (supports / responds-to).

An open question nobody has answered. Salience rises with how connected the question is (a well-situated but unanswered question matters more). A question is skipped when it is no longer an open ask, in either of two ways: its status is discarded (soft-deleted), or it carries an incoming supersedes edge (belief-revised by a later note — the edge-based supersession the rest of the engine records, since status="superseded" is never written). Every other status — including the unanswered open / proposed / draft authoring states — is surfaced.

Source code in zettelkasten/opportunities.py

def find_orphaned_questions(index: ClaimIndex) -> "list[dict[str, Any]]":
    """``question`` notes with no incoming answering edge (supports / responds-to).

    An open question nobody has answered. Salience rises with how connected the
    question is (a well-situated but unanswered question matters more). A question
    is skipped when it is no longer an open ask, in either of two ways: its status
    is ``discarded`` (soft-deleted), or it carries an incoming ``supersedes`` edge
    (belief-revised by a later note — the edge-based supersession the rest of the
    engine records, since ``status="superseded"`` is never written). Every other
    status — including the unanswered ``open`` / ``proposed`` / ``draft`` authoring
    states — is surfaced.
    """
    gaps: list[dict[str, Any]] = []
    for key, note in index.notes_by_key.items():
        if (note.type or "") != "question":
            continue
        status = (note.status or "complete").strip().lower()
        if status in CLOSED_QUESTION_STATUSES:
            continue
        incoming = index.by_target.get(key, [])
        # An incoming ``supersedes`` edge closes the question (belief-revised).
        if any(e.relation == SUPERSEDES_RELATION for e in incoming):
            continue
        if any(e.relation in ANSWER_RELATIONS for e in incoming):
            continue
        incident = len(incoming) + len(index.by_source.get(key, []))
        salience = min(1.0, 0.5 + 0.5 * _claims._saturate(incident, 4.0))
        anchor = _claims._claim_ref(index, key)
        title = note.title or key[1]
        gaps.append(_claims._gap(
            ORPHANED_QUESTION, salience=salience, action_tool="answer",
            action={"tool": "claim", "action": "test", "args": {"text": title},
                    "note": "Open question with no answering edge — answer it."},
            predicate="orphaned question (no answering backlink)",
            detail=f"incident_edges={incident}, incoming_answers=0",
            anchor=anchor,
        ))
    return gaps

find_voids

find_voids(index: ClaimIndex, *, clusters: 'list[list[Key]]', vectors: 'dict[Key, list[float]]', centrality: 'dict[Key, float]', min_similarity: float = VOID_MIN_SIMILARITY, void_max_size: int = VOID_MAX_SIZE, dense_min_size: int = VOID_DENSE_MIN_SIZE, labeler: 'Callable[[list[Key]], str] | None' = None) -> 'list[dict[str, Any]]'

Sparse-but-surrounded regions: a tiny cluster hugging a dense one.

A small cluster (<= void_max_size) whose centroid sits close (>= min_similarity) to a DENSE cluster (>= dense_min_size) is an underexplored area amid density — an unasked question. Optionally named by an injected labeler (else a deterministic label). Returns [] when there is no dense cluster to sit beside.

Source code in zettelkasten/opportunities.py

def find_voids(
    index: ClaimIndex,
    *,
    clusters: "list[list[Key]]",
    vectors: "dict[Key, list[float]]",
    centrality: "dict[Key, float]",
    min_similarity: float = VOID_MIN_SIMILARITY,
    void_max_size: int = VOID_MAX_SIZE,
    dense_min_size: int = VOID_DENSE_MIN_SIZE,
    labeler: "Callable[[list[Key]], str] | None" = None,
) -> "list[dict[str, Any]]":
    """Sparse-but-surrounded regions: a tiny cluster hugging a dense one.

    A small cluster (``<= void_max_size``) whose centroid sits close
    (``>= min_similarity``) to a DENSE cluster (``>= dense_min_size``) is an
    underexplored area amid density — an unasked question. Optionally named by an
    injected ``labeler`` (else a deterministic label). Returns ``[]`` when there
    is no dense cluster to sit beside.
    """
    if len(clusters) < 2:
        return []
    centroids = [_centroid(c, vectors) for c in clusters]
    dense = [i for i, c in enumerate(clusters) if len(c) >= dense_min_size]
    if not dense:
        return []
    gaps: list[dict[str, Any]] = []
    for i, members in enumerate(clusters):
        if len(members) > void_max_size:
            continue
        best_j = -1
        best_sim = -1.0
        for j in dense:
            if j == i:
                continue
            sim = _claims._guarded_cosine(centroids[i], centroids[j])
            if sim is not None and sim > best_sim:
                best_sim = sim
                best_j = j
        if best_j < 0 or best_sim < min_similarity:
            continue
        rep = _representative(members, centrality, index)
        near = _representative(clusters[best_j], centrality, index)
        label = labeler(members) if labeler else f"void near {near.get('title') or near.get('id')}"
        anchor = dict(rep)
        anchor["label"] = label
        anchor["near"] = near
        anchor["similarity"] = round(best_sim, 4)
        gaps.append(_claims._gap(
            VOID, salience=max(0.0, min(1.0, best_sim)), action_tool="explore",
            action={"tool": "claim", "action": "propose",
                    "args": {"theme": label},
                    "note": "Sparse region amid density — an underexplored area to develop."},
            predicate="void (sparse region amid density)",
            detail=f"label='{label}', similarity_to_dense={round(best_sim, 3)}",
            anchor=anchor,
        ))
    return gaps

find_transfers

find_transfers(index: ClaimIndex, *, clusters: 'list[list[Key]]', vectors: 'dict[Key, list[float]]', centrality: 'dict[Key, float]', min_similarity: float = TRANSFER_MIN_SIMILARITY, method_types: 'tuple[str, ...] | None' = None) -> 'list[dict[str, Any]]'

A method/model note that structurally fits an untried adjacent cluster.

For each method/model note in a cluster A, find the most similar OTHER cluster B (centroid sim >= min_similarity) to which the note has NO edge — the method has not been tried there. The anchor carries the source method note and the target cluster's representative. Returns [] when fewer than two clusters exist.

Source code in zettelkasten/opportunities.py

def find_transfers(
    index: ClaimIndex,
    *,
    clusters: "list[list[Key]]",
    vectors: "dict[Key, list[float]]",
    centrality: "dict[Key, float]",
    min_similarity: float = TRANSFER_MIN_SIMILARITY,
    method_types: "tuple[str, ...] | None" = None,
) -> "list[dict[str, Any]]":
    """A method/model note that structurally fits an untried adjacent cluster.

    For each method/model note in a cluster A, find the most similar OTHER cluster
    B (centroid sim ``>= min_similarity``) to which the note has NO edge — the
    method has not been tried there. The anchor carries the source method note and
    the target cluster's representative. Returns ``[]`` when fewer than two
    clusters exist.
    """
    if len(clusters) < 2:
        return []
    if method_types is None:
        method_types = METHOD_TYPES
    cluster_of: dict[Key, int] = {}
    for idx, members in enumerate(clusters):
        for key in members:
            cluster_of[key] = idx
    centroids = [_centroid(c, vectors) for c in clusters]
    member_set = [set(c) for c in clusters]

    gaps: list[dict[str, Any]] = []
    for key, cidx in sorted(cluster_of.items()):
        note = index.notes_by_key.get(key)
        if note is None or (note.type or "") not in method_types:
            continue
        # neighbours of this method note (either direction).
        neighbours: set[Key] = set()
        for e in index.by_source.get(key, []):
            neighbours.add(e.dst_key)
        for e in index.by_target.get(key, []):
            neighbours.add(e.src_key)
        best_j = -1
        best_sim = -1.0
        for j in range(len(clusters)):
            if j == cidx:
                continue
            if neighbours & member_set[j]:
                continue  # already tried in that cluster
            sim = _claims._guarded_cosine(centroids[cidx], centroids[j])
            if sim is not None and sim > best_sim:
                best_sim = sim
                best_j = j
        if best_j < 0 or best_sim < min_similarity:
            continue
        target = _representative(clusters[best_j], centrality, index)
        title = note.title or key[1]
        anchor = _claims._claim_ref(index, key)
        anchor["target_cluster"] = target
        anchor["similarity"] = round(best_sim, 4)
        gaps.append(_claims._gap(
            TRANSFER, salience=max(0.0, min(1.0, best_sim)), action_tool="transfer",
            action={"tool": "claim", "action": "test",
                    "args": {"text": f"Apply '{title}' to {target.get('title') or target.get('id')}"},
                    "note": "Method fits an adjacent cluster where it has not been tried."},
            predicate="untried method transfer (fits an adjacent cluster)",
            detail=(f"method '{note.type}' from {key[0]}/{key[1]} → cluster of "
                    f"{target.get('graph')}/{target.get('id')} (sim={round(best_sim, 3)})"),
            anchor=anchor,
        ))
    return gaps

detect_gaps

detect_gaps(*, index: ClaimIndex, ctx: tuple, get_graph: 'Callable[[str], Any]', centrality: 'dict[Key, float]', importance_map: 'dict[str, float] | None', kc_by_key: 'dict[Key, Any] | None', max_sal: float, wanted: 'set[str] | None', localize: 'Callable[[str], str] | None' = None, namespace: 'Callable[[str], str] | None' = None, graphs_dir: Any = None, project: str = '', graph: str = '', embed_fn: 'Callable[[str, str], Any] | None' = None, overlay: 'dict[str, Any] | None' = None, memory_source: Any = None, debate: 'dict[str, Any] | None' = None) -> 'list[dict[str, Any]]'

Run the requested negative-space detectors and return their gaps (flat list).

Wires the production dependencies (vectors + clusters, the cross-store overlay, the debate map) and dispatches to the pure detectors. Only families present in wanted run (wanted=None runs them all). Each dependency is built lazily, so requesting only orphaned_question never touches embeddings or the overlay. overlay / memory_source / debate / embed_fn are injectable for deterministic tests; unset, they are resolved from the stores.

graphs_dir + namespace describe the scan's scope: on a federated / namespaced scan the (unscopeable) asymmetry loaders are skipped so a peer's claims are never paired with the local overlay (see the asymmetry block).

Source code in zettelkasten/opportunities.py

def detect_gaps(
    *,
    index: ClaimIndex,
    ctx: tuple,
    get_graph: "Callable[[str], Any]",
    centrality: "dict[Key, float]",
    importance_map: "dict[str, float] | None",
    kc_by_key: "dict[Key, Any] | None",
    max_sal: float,
    wanted: "set[str] | None",
    localize: "Callable[[str], str] | None" = None,
    namespace: "Callable[[str], str] | None" = None,
    graphs_dir: Any = None,
    project: str = "",
    graph: str = "",
    embed_fn: "Callable[[str, str], Any] | None" = None,
    overlay: "dict[str, Any] | None" = None,
    memory_source: Any = None,
    debate: "dict[str, Any] | None" = None,
) -> "list[dict[str, Any]]":
    """Run the requested negative-space detectors and return their gaps (flat list).

    Wires the production dependencies (vectors + clusters, the cross-store overlay,
    the debate map) and dispatches to the pure detectors. Only families present in
    ``wanted`` run (``wanted=None`` runs them all). Each dependency is built
    lazily, so requesting only ``orphaned_question`` never touches embeddings or
    the overlay. ``overlay`` / ``memory_source`` / ``debate`` / ``embed_fn`` are
    injectable for deterministic tests; unset, they are resolved from the stores.

    ``graphs_dir`` + ``namespace`` describe the scan's scope: on a federated /
    namespaced scan the (unscopeable) asymmetry loaders are skipped so a peer's
    claims are never paired with the local overlay (see the asymmetry block).
    """
    def _want(t: str) -> bool:
        return wanted is None or t in wanted

    gaps: list[dict[str, Any]] = []

    # ── asymmetry (cross-store overlay) ─────────────────────────────────────────
    # The overlay + memory-source loaders read the LOCAL default stores with no
    # scope argument, so on a FEDERATED / namespaced scan pairing them with the
    # peer's ``index.claims`` fabricates asymmetry gaps. Since the loaders cannot
    # be scoped, asymmetry is hard-disabled for non-local scans: the loaders are
    # SKIPPED and the family contributes nothing (injected deps still run, for
    # deterministic tests). See ``_is_local_scope``.
    if _want(ASYMMETRY):
        local_scan = _is_local_scope(graphs_dir, namespace)
        ov = overlay
        if ov is None and local_scan:
            ov = _load_claim_overlay()
        ms = memory_source
        if ms is None and local_scan:
            ms = _load_memory_source()
        if ov and ms is not None:
            try:
                gaps.extend(find_asymmetry(
                    index, overlay=ov, memory_source=ms,
                    kc_by_key=kc_by_key, max_sal=max_sal,
                ))
            except Exception as exc:  # pragma: no cover - defensive
                logger.warning("opportunities: asymmetry detector failed: %s", exc)

    # ── orphaned questions (index only — no embeddings) ─────────────────────────
    if _want(ORPHANED_QUESTION):
        try:
            gaps.extend(find_orphaned_questions(index))
        except Exception as exc:  # pragma: no cover - defensive
            logger.warning("opportunities: orphaned_question detector failed: %s", exc)

    # ── crux (debate map) ───────────────────────────────────────────────────────
    if _want(CRUX):
        dm = debate
        if dm is None:
            try:
                dm = _claims.debate_map(get_graph, _context=ctx)
            except Exception as exc:  # pragma: no cover - defensive
                logger.warning("opportunities: debate map failed: %s", exc)
                dm = None
        if dm:
            try:
                gaps.extend(find_crux(
                    index, debate=dm, centrality=centrality, importance_map=importance_map,
                ))
            except Exception as exc:  # pragma: no cover - defensive
                logger.warning("opportunities: crux detector failed: %s", exc)

    # ── cluster-based families (bridge / void / transfer) ───────────────────────
    if _want(BRIDGE) or _want(VOID) or _want(TRANSFER):
        provider = _claims._discover_vector_provider(get_graph, embed_fn)
        if embed_fn is not None:
            vectors: dict[Key, list[float]] = {}
            for key, note in index.notes_by_key.items():
                if (note.type or "") not in CLUSTERABLE_TYPES:
                    continue
                vec = provider(key[0], key[1])
                if vec:
                    vectors[key] = vec
        else:
            vectors = _collect_vectors(get_graph, index, localize)
        clusters = _build_clusters(index, vectors)
        if len(clusters) >= 2:
            if _want(BRIDGE):
                try:
                    gaps.extend(find_bridges(
                        index, clusters=clusters, vectors=vectors, centrality=centrality,
                    ))
                except Exception as exc:  # pragma: no cover - defensive
                    logger.warning("opportunities: bridge detector failed: %s", exc)
            if _want(VOID):
                try:
                    gaps.extend(find_voids(
                        index, clusters=clusters, vectors=vectors, centrality=centrality,
                    ))
                except Exception as exc:  # pragma: no cover - defensive
                    logger.warning("opportunities: void detector failed: %s", exc)
            if _want(TRANSFER):
                try:
                    gaps.extend(find_transfers(
                        index, clusters=clusters, vectors=vectors, centrality=centrality,
                    ))
                except Exception as exc:  # pragma: no cover - defensive
                    logger.warning("opportunities: transfer detector failed: %s", exc)

    return gaps