zettelkasten.synapse.matrix

zettelkasten.synapse.matrix

Cross-store matrix lens: the link overlay projected as practice x canon.

A read-only VIEW over the bipartite connection overlay — it materializes nothing into either store. Rows are memory (practice) nodes, columns are ZK (canon) nodes, and a cell is the typed edge between them (relation + confidence). This is the natural matrix shape of a bipartite graph, so no spine/table machinery is needed.

On top of the grid, :func:synthesize_axes rolls each axis up to a summary:

• row synthesis — how one practice node relates across the canon it touches;
• column synthesis — how one canon node is exercised across practice;
• apex — a single synthesis over the row + column summaries.

Synthesis is injectable (default = a tool-free LLM pass) so the lens builds and tests without an LLM.

build_matrix_lens

build_matrix_lens(zk_get_graph: GetGraph, projects: list[str] | None = None, graphs_dir: 'Path | None' = None, overlay: dict[str, Any] | None = None, min_confidence: float = 0.0, max_rows: int = _MAX_AXIS, max_columns: int = _MAX_AXIS) -> dict[str, Any]

Project the connection overlay into a practice x canon matrix (read-only).

Rows = memory nodes, columns = ZK nodes, cells = typed edges. Titles are resolved best-effort via the stores. Axes are capped (busiest nodes first) with a truncation flag.

Source code in zettelkasten/synapse/matrix.py

def build_matrix_lens(
    zk_get_graph: GetGraph,
    projects: list[str] | None = None,
    graphs_dir: "Path | None" = None,
    overlay: dict[str, Any] | None = None,
    min_confidence: float = 0.0,
    max_rows: int = _MAX_AXIS,
    max_columns: int = _MAX_AXIS,
) -> dict[str, Any]:
    """Project the connection overlay into a practice x canon matrix (read-only).

    Rows = memory nodes, columns = ZK nodes, cells = typed edges. Titles are
    resolved best-effort via the stores. Axes are capped (busiest nodes first)
    with a truncation flag.
    """
    from zettelkasten.synapse import overlay as _overlay

    if overlay is None:
        overlay = _overlay.load_overlay()
    edges = [e for e in overlay.get("edges", []) if float(e.get("confidence", 0.0)) >= min_confidence]

    _sources, get_graph = resolve_scope(zk_get_graph, projects=projects, graphs_dir=graphs_dir)

    def _memory_title(mid: str) -> tuple[str, str]:
        try:
            note = get_graph(MEMORY_SOURCE).notes.get(mid)
        except Exception:
            note = None
        return (getattr(note, "title", "") if note else ""), (getattr(note, "type", "") if note else "")

    def _zk_title(zid: str, source: str) -> str:
        try:
            note = get_graph(source).notes.get(zid)
        except Exception:
            note = None
        return getattr(note, "title", "") if note else ""

    # Degree-order the axes so a cap keeps the most-connected nodes.
    row_deg: dict[str, int] = {}
    col_deg: dict[str, int] = {}
    zk_source_of: dict[str, str] = {}
    for e in edges:
        row_deg[e["memory_id"]] = row_deg.get(e["memory_id"], 0) + 1
        col_deg[e["zk_id"]] = col_deg.get(e["zk_id"], 0) + 1
        zk_source_of.setdefault(e["zk_id"], e.get("zk_source", ""))

    row_ids = [rid for rid, _ in sorted(row_deg.items(), key=lambda kv: (-kv[1], kv[0]))]
    col_ids = [cid for cid, _ in sorted(col_deg.items(), key=lambda kv: (-kv[1], kv[0]))]
    rows_truncated = len(row_ids) > max_rows
    cols_truncated = len(col_ids) > max_columns
    row_ids = row_ids[:max_rows]
    col_ids = col_ids[:max_columns]
    row_set, col_set = set(row_ids), set(col_ids)

    cells = [
        {
            "memory_id": e["memory_id"],
            "zk_id": e["zk_id"],
            # Carry the canon source so a downstream consumer (e.g. the grounded
            # synthesis provider) can resolve the claim by its EXACT (zk_source,
            # zk_id) key — the same note id can exist in multiple in-scope graphs.
            "zk_source": e.get("zk_source", ""),
            "relation": e.get("relation"),
            "confidence": e.get("confidence"),
            "rationale": e.get("rationale", ""),
            # Present only on the claim-aligned overlay (additive; ``None`` on the
            # generic note-note overlay).
            "claim_strength": e.get("claim_strength"),
            "priority": e.get("priority"),
        }
        for e in edges
        if e["memory_id"] in row_set and e["zk_id"] in col_set
    ]

    relation_totals: dict[str, int] = {}
    for c in cells:
        relation_totals[c["relation"]] = relation_totals.get(c["relation"], 0) + 1

    rows = []
    for mid in row_ids:
        title, mtype = _memory_title(mid)
        rows.append({"id": mid, "store": "memory", "tier": "practice", "title": title, "type": mtype})
    columns = []
    for zid in col_ids:
        source = zk_source_of.get(zid, "")
        columns.append({"id": zid, "store": "zk", "tier": "canon",
                        "title": _zk_title(zid, source), "source": source})

    return {
        "rows": rows,
        "columns": columns,
        "cells": cells,
        "relation_totals": relation_totals,
        "row_count": len(rows),
        "column_count": len(columns),
        "edge_count": len(cells),
        "truncated": {"rows": rows_truncated, "columns": cols_truncated},
        "scope": {"projects": overlay.get("manifest", {}).get("projects", [])},
    }

synthesize_axes

synthesize_axes(lens: dict[str, Any], synth: AxisSynthesizer | None = None) -> dict[str, Any]

Add row/column summaries and a rolled-up apex synthesis to a matrix lens.

synth(kind, header, items) returns summary text; defaults to a tool-free LLM pass. Purely additive and read-only — the returned lens gains row_synthesis / column_synthesis / apex fields.

Source code in zettelkasten/synapse/matrix.py

def synthesize_axes(lens: dict[str, Any], synth: AxisSynthesizer | None = None) -> dict[str, Any]:
    """Add row/column summaries and a rolled-up apex synthesis to a matrix lens.

    ``synth(kind, header, items)`` returns summary text; defaults to a tool-free
    LLM pass. Purely additive and read-only — the returned lens gains
    ``row_synthesis`` / ``column_synthesis`` / ``apex`` fields.
    """
    synth = synth or _llm_synth

    row_summaries: dict[str, str] = {}
    for r in lens["rows"]:
        items = _row_items(lens, r["id"])
        if items:
            row_summaries[r["id"]] = synth("row", r["title"] or r["id"], items)

    column_summaries: dict[str, str] = {}
    for c in lens["columns"]:
        items = _col_items(lens, c["id"])
        if items:
            column_summaries[c["id"]] = synth("column", c["title"] or c["id"], items)

    apex_items = (
        [{"counterpart": (r["title"] or r["id"]), "relation": "practice", "rationale": row_summaries[r["id"]]}
         for r in lens["rows"] if r["id"] in row_summaries]
        + [{"counterpart": (c["title"] or c["id"]), "relation": "canon", "rationale": column_summaries[c["id"]]}
           for c in lens["columns"] if c["id"] in column_summaries]
    )
    apex = synth("apex", "cross-store synthesis", apex_items) if apex_items else ""

    lens = dict(lens)
    lens["row_synthesis"] = row_summaries
    lens["column_synthesis"] = column_summaries
    lens["apex"] = apex
    return lens