zettelkasten.tables

zettelkasten.tables

Synthesis-matrix engine: corpus → entity-row comparison table.

The Workshop's "Synthesis Matrix" is the read-side projection of grounded extraction. Where :mod:zettelkasten.outline assembles a narrative scaffold, this module assembles a grid: one row per source (entity) in the review's corpus, and user-defined, typed columns. Each cell carries an interpretive value backed by verified evidence (a verbatim quote note + citation), with a provenance tag (self-stated / cross-source / reviewer-inferred). A cell with no supporting note is an explicit [GAP].

Like :mod:zettelkasten.outline this follows the engine-gathers / agent-drafts split, but inverts the priority: most cells fill deterministically with NO LLM. A column backed by a schema dimension tag, a note type, or a CCC slot is populated by querying the row entity's own notes (and their linked quote evidence); the injectable agent EXTRACT pass runs only for free-text prompt columns and cells the deterministic pass left empty.

Persistence mirrors the outline artifact discipline: table definitions + the generated grid live in a single regenerable derived artifact _reviews/<review>.tables.json written through the M3a substrate (review_write_lock → atomic_write_text → _schedule_zettel_commit), keyed by a generation signature so navigation peek is cost-free.

generate_column_synthesis

generate_column_synthesis(column: dict[str, Any], rows: list[dict[str, Any]], *, summarize_fn: 'Callable[[str, str], str] | None' = None) -> str

On-demand one-line AI narrative for a column (consensus / disagreement).

Injectable like the outline DRAFT: summarize_fn(system, prompt) -> str defaults to the in-process dashboard agent. Returns "" when the column is empty so the caller can fall back to the mechanical summary.

Source code in zettelkasten/tables.py

def generate_column_synthesis(
    column: dict[str, Any],
    rows: list[dict[str, Any]],
    *,
    summarize_fn: "Callable[[str, str], str] | None" = None,
) -> str:
    """On-demand one-line AI narrative for a column (consensus / disagreement).

    Injectable like the outline DRAFT: ``summarize_fn(system, prompt) -> str``
    defaults to the in-process dashboard agent. Returns ``""`` when the column is
    empty so the caller can fall back to the mechanical summary.
    """
    key = column["key"]
    items = []
    for r in rows:
        cell = r["cells"].get(key) or {}
        if cell.get("gap"):
            continue
        items.append({"entity": r["label"], "value": cell.get("value", "")})
    if not items:
        return ""
    fn = summarize_fn or _default_summarize_fn
    system = (
        "You summarize one column of a research comparison matrix in a SINGLE "
        "sentence. State the consensus and name the notable disagreements or "
        "outliers by entity. Be precise and do not invent values."
    )
    prompt = (
        f"Column: {column['label']}\n\n"
        f"Per-entity values:\n{json.dumps(items, indent=2, ensure_ascii=False)}\n\n"
        "Write one sentence summarizing this column across the entities."
    )
    return (fn(system, prompt) or "").strip()

suggest_from_description

suggest_from_description(description: str, *, doc_type: str = '', suggest_fn: 'Callable[[str, str], str] | None' = None, get_graph: 'GetGraph | None' = None, project: str = '', graph: str = '', graphs_dir: 'Path | None' = None, localize: 'Callable[[str], str] | None' = None, corpus_limit: int = 120) -> dict[str, Any]

Map a natural-language comparison request to a matrix config.

Returns {"row_axis": {...}, "columns": [...], "themes": [...], "rationale": str}. The agent picks a row axis and an ordered column list from the deterministic catalog (schema dimensions, note types, CCC slots, note/source identity fields); anything the catalog can't cover becomes a free-text prompt column. Deterministic-first by construction: grounded catalog columns are preferred so as much of the grid as possible fills without the LLM.

CORPUS-ANALYSIS PASS (design §8.1). When a get_graph + scope is supplied, this reads a bounded sample of the in-scope notes and asks the agent to propose BOTH (a) the relevant dimensions → columns and (b) the recurring themes → row-hubs (a theme groups sources; the default single-spine row unit stays source, so themes are an OPTIONAL row-hub layer above the source rows, not a separate flat axis). Each returned theme is normalized to {"label": str, "note_ids": [<graph::id>, …]}. When no get_graph is supplied the historical catalog-only mapping runs unchanged (themes is then []), so existing callers are unaffected.

A blank description or an unparseable reply degrades to the source axis with no columns and no themes, so the caller always receives a valid (if empty) config.

Source code in zettelkasten/tables.py

def suggest_from_description(
    description: str,
    *,
    doc_type: str = "",
    suggest_fn: "Callable[[str, str], str] | None" = None,
    get_graph: "GetGraph | None" = None,
    project: str = "",
    graph: str = "",
    graphs_dir: "Path | None" = None,
    localize: "Callable[[str], str] | None" = None,
    corpus_limit: int = 120,
) -> dict[str, Any]:
    """Map a natural-language comparison request to a matrix config.

    Returns ``{"row_axis": {...}, "columns": [...], "themes": [...], "rationale":
    str}``. The agent picks a row axis and an ordered column list from the
    deterministic catalog (schema dimensions, note types, CCC slots, note/source
    identity fields); anything the catalog can't cover becomes a free-text
    ``prompt`` column. Deterministic-first by construction: grounded catalog
    columns are preferred so as much of the grid as possible fills without the LLM.

    CORPUS-ANALYSIS PASS (design §8.1). When a ``get_graph`` + scope is supplied,
    this reads a bounded sample of the in-scope notes and asks the agent to propose
    BOTH (a) the relevant **dimensions → columns** and (b) the recurring **themes →
    row-hubs** (a theme groups sources; the default single-spine row unit stays
    ``source``, so themes are an OPTIONAL row-hub layer above the source rows, not a
    separate flat axis). Each returned theme is normalized to
    ``{"label": str, "note_ids": [<graph::id>, …]}``. When no ``get_graph`` is
    supplied the historical catalog-only mapping runs unchanged (``themes`` is then
    ``[]``), so existing callers are unaffected.

    A blank description or an unparseable reply degrades to the source axis with no
    columns and no themes, so the caller always receives a valid (if empty) config.
    """
    base = Path(graphs_dir) if graphs_dir is not None else GRAPHS_DIR
    loc = localize or (lambda s: s)
    desc = (description or "").strip()
    blank = {"row_axis": normalize_row_axis({"kind": "source"}), "columns": [], "themes": [], "rationale": ""}
    if not desc:
        return blank

    # Corpus-analysis input: a bounded digest of the in-scope notes (empty when no
    # grapher/scope is supplied → the catalog-only mapping below).
    corpus = (
        _corpus_digest_for_suggest(
            get_graph, project=project, graph=graph, base=base, loc=loc, limit=corpus_limit
        )
        if get_graph is not None and (project or graph)
        else []
    )
    valid_note_ids = {str(d.get("id") or "") for d in corpus}

    catalog = suggest_columns(doc_type=doc_type)
    # Flat ``id → column-def`` lookup the agent selects from. ids are stable and
    # typed (``<backing>:<ref>``) so a returned id maps straight back to a grounded
    # column definition.
    by_id: dict[str, dict[str, Any]] = {}
    listing: list[dict[str, str]] = []

    def _offer(group: str, cid: str, col: dict[str, Any]) -> None:
        if cid in by_id:
            return
        by_id[cid] = col
        listing.append({"id": cid, "label": col["label"], "group": group})

    for col in catalog["identity"]:
        _offer("note/source field", f"identity:{col['ref']}", col)
    for sch in catalog["schemas"]:
        group = f"{'spine' if sch.get('synthesis') else 'template'}: {sch['name']}"
        for col in sch["columns"]:
            _offer(group, f"schema_tag:{col['ref']}", col)
    for col in catalog["note_types"]:
        _offer("note type", f"note_type:{col['ref']}", col)
    for col in catalog["ccc"]:
        _offer("CCC slot", f"ccc:{col['ref']}", col)

    axes = [
        {"kind": "source", "desc": "one row per source document"},
        {"kind": "note", "desc": "one row per note / concept (NO aggregation — each note is its own row)"},
        {"kind": "tag", "desc": "one row per tag value (spans sources)"},
        {"kind": "note_type", "desc": "one row per note type"},
        {
            "kind": "group",
            "desc": (
                "AGGREGATE many notes into one row per emergent entity. Use this "
                "whenever the user wants 'a row per <thing>' where <thing> "
                "consolidates several notes (e.g. 'a row per persona', 'one row per "
                "company'). Pick a 'strategy': 'semantic' (you/the agent read the "
                "notes and partition them per an 'instruction' — use when the entity "
                "is implied by content, e.g. personas), 'link' (group by the apex a "
                "note links to via a 'relation' — use when an explicit spine/links "
                "already connect members to a hub), or 'field' (group by a "
                "frontmatter field named in 'ref')."
            ),
        },
    ]
    # The corpus-analysis pass reads the notes and proposes themes (row-hubs) on
    # top of the column/axis mapping; the catalog-only pass (no corpus) keeps the
    # historical single-job instruction so existing callers behave identically.
    if corpus:
        system = (
            "You configure a research comparison matrix by ANALYZING a sample of "
            "the corpus notes for a short description. Do TWO jobs: (1) choose ONE "
            "row axis + an ordered list of COLUMNS = the relevant DIMENSIONS to "
            "compare per row; (2) identify the recurring THEMES in the corpus — "
            "each theme groups several sources and becomes a row-hub over the "
            "default source rows. When the user wants rows that CONSOLIDATE many "
            "notes into higher-level entities ('a row per persona'), choose the "
            "'group' axis (NOT 'note'). STRONGLY prefer catalog columns — they fill "
            "deterministically with no guessing; only add a free-text column when "
            "nothing in the catalog fits. Ground every theme's members in the "
            "provided note ids. Return STRICT JSON and nothing else."
        )
    else:
        system = (
            "You configure a research comparison matrix from a short description. "
            "Choose ONE row axis (what each row represents) and an ordered list of "
            "columns (what is compared per row). When the user wants rows that "
            "CONSOLIDATE many notes into higher-level entities ('a row per persona'), "
            "choose the 'group' axis (NOT 'note', which makes one row per raw note). "
            "STRONGLY prefer catalog columns — they fill deterministically with no "
            "guessing; only add a free-text column when nothing in the catalog fits. "
            "Return STRICT JSON and nothing else."
        )
    corpus_block = (
        f"Corpus sample (ground THEMES' members in these note ids):\n"
        f"{json.dumps(corpus, indent=2, ensure_ascii=False)}\n\n"
        if corpus
        else ""
    )
    themes_shape = (
        '  "themes": [\n'
        '    {"label": "RCT designs", "note_ids": ["src-a::n1", "src-b::n4"]}\n'
        "  ],\n"
        if corpus
        else ""
    )
    themes_help = (
        "'themes' (OPTIONAL) are recurring groupings of SOURCES that become "
        "row-hubs above the default source rows; each carries a short 'label' and "
        "the 'note_ids' (from the corpus sample) that belong to it. Omit 'themes' "
        "or return [] when no clear grouping emerges."
        if corpus
        else ""
    )
    prompt = (
        f"Description:\n{desc}\n\n"
        + corpus_block
        + f"Row axes (choose one):\n{json.dumps(axes, indent=2)}\n\n"
        f"Column catalog (reference by 'id'):\n"
        f"{json.dumps(listing, indent=2, ensure_ascii=False)}\n\n"
        "Return JSON of the form:\n"
        "{\n"
        '  "row_axis": {"kind": "group", "strategy": "semantic", '
        '"instruction": "one row per persona, consolidating that persona\'s notes"},\n'
        '  "columns": [\n'
        '    {"id": "schema_tag:claim"},\n'
        '    {"prompt": "the concept\'s core formula", "label": "Formula", "type": "numeric"}\n'
        "  ],\n"
        + themes_shape
        + '  "rationale": "one short sentence"\n'
        "}\n"
        "'row_axis' may be a bare kind string (e.g. \"note\") for the simple axes, "
        "or an object for 'group' carrying 'strategy' and ('instruction' for "
        "semantic | 'relation'[+'direction'/'apex'] for link | 'ref' for field). "
        "Use a catalog 'id' for grounded columns; use 'prompt'+'label' (+optional "
        "'type': extractive|numeric|categorical) only for free-text columns. "
        + themes_help
    )
    fn = suggest_fn or _default_extract_fn
    data = _parse_json_object(fn(system, prompt) or "")
    if not isinstance(data, dict):
        return blank

    raw_axis = data.get("row_axis")
    if isinstance(raw_axis, dict):
        axis_in = dict(raw_axis)
        if str(axis_in.get("kind") or "source") not in _ROW_AXES:
            axis_in["kind"] = "source"
    else:
        kind = str(raw_axis or "source")
        if kind not in _ROW_AXES:
            kind = "source"
        axis_in = {"kind": kind}
    raw_cols: list[dict[str, Any]] = []
    for item in data.get("columns") or []:
        if not isinstance(item, dict):
            continue
        cid = str(item.get("id") or "").strip()
        if cid and cid in by_id:
            raw_cols.append(dict(by_id[cid]))
            continue
        instruction = str(item.get("prompt") or item.get("instruction") or "").strip()
        if not instruction:
            continue
        raw_cols.append(
            {
                "label": str(item.get("label") or instruction[:40]).strip(),
                "type": str(item.get("type") or "extractive"),
                "backing": "prompt",
                "ref": "",
                "prompt": instruction,
            }
        )

    # Themes → proposed row-hubs. Normalize each to ``{label, note_ids}`` and keep
    # ONLY note ids the corpus sample actually contains (so a hallucinated member
    # never leaks into the config). A theme with a label but no valid members is
    # still kept (label-only hub) so the user can curate it; a labelless theme is
    # dropped. Themes are always [] on the catalog-only path (no corpus).
    themes: list[dict[str, Any]] = []
    for item in data.get("themes") or []:
        if not isinstance(item, dict):
            continue
        label = str(item.get("label") or "").strip()
        if not label:
            continue
        note_ids = [
            nid
            for nid in (str(x).strip() for x in (item.get("note_ids") or []))
            if nid and (not valid_note_ids or nid in valid_note_ids)
        ]
        themes.append({"label": label, "note_ids": note_ids})

    return {
        "row_axis": normalize_row_axis(axis_in),
        "columns": normalize_columns(raw_cols),
        "themes": themes,
        "rationale": str(data.get("rationale") or "").strip(),
    }

stream_table async

stream_table(get_graph: GetGraph, *, project: str = '', graph: str = '', name: str = '', table_id: str = '', title: str = '', columns: 'list[dict[str, Any]] | None' = None, row_axis: 'dict[str, Any] | None' = None, force: bool = False, extract_fn: 'Callable[[str, str], str] | None' = None, summarize_fn: 'Callable[[str, str], str] | None' = None, graphs_dir: 'Path | None' = None, localize: 'Callable[[str], str] | None' = None)

Streaming twin of :func:build_matrix: the same GATHER → deterministic-fill → EXTRACT → synthesize → persist pipeline, but an async generator that yields PROGRESS so a matrix build is never an opaque, crash-looking spinner.

Yields plain {"event": <name>, "data": {...}} dicts (the route formats them as SSE — this stays framework-agnostic):

• stage — a phase marker (gather → extract → synthesize) with a short human message.
• grid — the deterministic grid the instant GATHER finishes, so the WHOLE table renders immediately (every non-prompt column filled, prompt cells still [GAP]) while the agent works.
• cells — one row's freshly EXTRACTed prompt cells, emitted per row so the AI columns fill in live rather than all-at-once at the end.
• synthesis — one column's AI consensus/disagreement narrative, emitted per column during the synthesize phase. When summarize_fn is supplied the build AUTO-synthesizes every filled column (no per-column click), folding the narrative into synthesis[key]["ai"] so it persists and survives reload.
• result — the terminal payload, identical in shape to :func:build_matrix's return (rows/synthesis/signature/ cached/…), so the frontend reconciles it exactly like the JSON build.

A cache hit (unchanged signature + extant grid, force false) emits a single result with cached=True and no agent call. peek / preview are NOT streaming concerns — navigation/wizard keep using the cost-free JSON :func:build_matrix paths. Errors are raised; the route wraps them into a terminal error event.

Source code in zettelkasten/tables.py

async def stream_table(
    get_graph: GetGraph,
    *,
    project: str = "",
    graph: str = "",
    name: str = "",
    table_id: str = "",
    title: str = "",
    columns: "list[dict[str, Any]] | None" = None,
    row_axis: "dict[str, Any] | None" = None,
    force: bool = False,
    extract_fn: "Callable[[str, str], str] | None" = None,
    summarize_fn: "Callable[[str, str], str] | None" = None,
    graphs_dir: "Path | None" = None,
    localize: "Callable[[str], str] | None" = None,
):
    """Streaming twin of :func:`build_matrix`: the same GATHER → deterministic-fill
    → EXTRACT → synthesize → persist pipeline, but an async generator that yields
    PROGRESS so a matrix build is never an opaque, crash-looking spinner.

    Yields plain ``{"event": <name>, "data": {...}}`` dicts (the route formats
    them as SSE — this stays framework-agnostic):

    * ``stage`` — a phase marker (``gather`` → ``extract`` → ``synthesize``) with a
      short human ``message``.
    * ``grid`` — the deterministic grid the instant GATHER finishes, so the WHOLE
      table renders immediately (every non-``prompt`` column filled, ``prompt``
      cells still ``[GAP]``) while the agent works.
    * ``cells`` — one row's freshly EXTRACTed ``prompt`` cells, emitted per row so
      the AI columns fill in live rather than all-at-once at the end.
    * ``synthesis`` — one column's AI consensus/disagreement narrative, emitted per
      column during the synthesize phase. When ``summarize_fn`` is supplied the
      build AUTO-synthesizes every filled column (no per-column click), folding the
      narrative into ``synthesis[key]["ai"]`` so it persists and survives reload.
    * ``result`` — the terminal payload, identical in shape to
      :func:`build_matrix`'s return (``rows``/``synthesis``/``signature``/
      ``cached``/…), so the frontend reconciles it exactly like the JSON build.

    A cache hit (unchanged signature + extant grid, ``force`` false) emits a single
    ``result`` with ``cached=True`` and no agent call. ``peek`` / ``preview`` are
    NOT streaming concerns — navigation/wizard keep using the cost-free JSON
    :func:`build_matrix` paths. Errors are raised; the route wraps them into a
    terminal ``error`` event.
    """
    import asyncio

    base = Path(graphs_dir) if graphs_dir is not None else GRAPHS_DIR
    if not name:
        name = graph or project or "all"
    validate_id(name, kind="review name", for_filename=True)
    if not table_id:
        table_id = "matrix"
    validate_id(table_id, kind="table id", for_filename=True)

    cols = normalize_columns(columns or [])
    axis = normalize_row_axis(row_axis)

    # The grid's spine origin, echoed on the result/grid frames so the frontend
    # stamps ``gridSpineId`` cross-session (see :func:`_spine_id_for_org`).
    spine_id = _spine_id_for_org(
        _load_backing_org(name, table_id, project=project, graph=graph, base=base)
    )

    # A ported BLANK-spine axis loads its durable attribution decisions and folds a
    # cross-graph + attribution term into the cache key — see :func:`build_matrix`
    # for the rationale (the signature then moves on a foreign collision or a
    # decision change, so a cache hit is safe and a miss re-scans).
    is_blank_spine = _is_blank_spine_axis(axis)
    attributions = (
        _load_attributions(name, table_id, project=project, graph=graph, graphs_dir=base)
        if is_blank_spine
        else None
    )
    blank_sig_extra = _blank_spine_sig_extra(base, attributions) if is_blank_spine else ""

    store = _load_tables(name, base)
    stored = store["tables"].get(table_id)

    # ── spine/schema axis: deterministic build, no agent (streamed twin) ──────
    # Mirrors :func:`build_matrix`'s spine branch — adopt the schema-matrix, emit
    # the grid + result frames, persist as a normal table. No gather/extract/
    # synthesize phases (the cells come straight from spine membership).
    if axis.get("kind") == "group" and axis.get("strategy") == "spine":
        want_synth = summarize_fn is not None
        table = await asyncio.to_thread(
            _spine_schema_table,
            get_graph,
            axis=axis,
            project=project,
            graph=graph,
            name=name,
            table_id=table_id,
            title=title,
            cols=cols,
            base=base,
            localize=localize,
            spine_id=spine_id,
            with_node_refs=want_synth,
        )
        refs = _extract_node_refs(table)
        if not force and stored is not None and stored.get("signature") == table["signature"]:
            out = dict(stored)
            out.setdefault("row_axis", normalize_row_axis(stored.get("row_axis")))
            out.setdefault("schema_version", _TABLE_SCHEMA_VERSION)
            out["stale"] = False
            out["cached"] = True
            out["exists"] = True
            out["spine_id"] = spine_id
            yield {"event": "result", "data": out}
            return
        yield {"event": "stage", "data": {"phase": "gather", "message": "Reading spine structure…"}}
        # Emit the scaffold grid immediately (member titles fill every cell) so the
        # whole table renders while the per-cell paragraphs are authored.
        yield {"event": "grid", "data": {
            "table_id": table["table_id"],
            "title": table["title"],
            "columns": table["columns"],
            "row_axis": table["row_axis"],
            "rows": table["rows"],
            "synthesis": {},
            "rows_count": len(table["rows"]),
            "columns_count": len(table["columns"]),
            "prompt_columns": 0,
            "spine_id": spine_id,
        }}
        # ── per-cell summaries: author + materialize one row at a time (streamed) ─
        # Mirrors the outline scaffold's progressive build — each row's authored
        # paragraphs are emitted as a ``cells`` frame the instant they're written,
        # and persisted onto the backing spine dimension nodes.
        if want_synth and refs:
            yield {"event": "stage", "data": {
                "phase": "synthesize",
                "message": "Writing cell summaries…",
                "rows": len(table["rows"]),
            }}
            loc2 = localize or (lambda s: s)
            cache: dict[str, Any] = {}
            for row in table["rows"]:
                changed = await asyncio.to_thread(
                    _materialize_row_summaries,
                    get_graph, loc2, cache, table["columns"], row, refs,
                    summarize_fn=summarize_fn, force=force,
                )
                if changed:
                    yield {"event": "cells", "data": {"row_id": row["id"], "cells": changed}}
            table["synthesis"] = compute_synthesis(table["columns"], table["rows"])
            table["attached_note_ids"] = _attached_note_ids(table["rows"])
        await asyncio.to_thread(
            _persist_table, name, table, base, project=project, graph=graph
        )
        out = dict(table)
        out["stale"] = False
        out["cached"] = False
        out["exists"] = True
        yield {"event": "result", "data": out}
        return

    signature = _table_signature(
        cols, row_axis=axis, project=project, graph=graph, name=name, base=base,
        extra=blank_sig_extra,
    )

    # ── cache hit: unchanged signature + extant grid ⇒ no gather/agent ────────
    if not force and stored is not None and stored.get("signature") == signature:
        out = dict(stored)
        out.setdefault("row_axis", normalize_row_axis(stored.get("row_axis")))
        out.setdefault("schema_version", _TABLE_SCHEMA_VERSION)
        out["stale"] = False
        out["cached"] = True
        out["exists"] = True
        out["spine_id"] = spine_id
        yield {"event": "result", "data": out}
        return

    # ── GATHER + deterministic fill (off the event loop so the stream stays live) ─
    # Only digest each row's note material when we will actually EXTRACT — it is
    # grounding input for the agent, not grid data, and is stripped before any row
    # is sent on the wire or persisted.
    prompt_cols = [c for c in cols if c["backing"] == "prompt"]
    need_material = bool(prompt_cols) and extract_fn is not None
    # A semantic ``group`` axis needs the agent to partition the corpus into rows;
    # the same injected ``extract_fn`` doubles as the grouping seam.
    group_fn = extract_fn if axis.get("strategy") == "semantic" else None
    yield {"event": "stage", "data": {"phase": "gather", "message": "Gathering material…"}}
    collision_out: dict[str, Any] = {}
    rows = await asyncio.to_thread(
        gather_rows,
        get_graph,
        cols,
        project=project,
        graph=graph,
        row_axis=axis,
        graphs_dir=base,
        localize=localize,
        with_material=need_material,
        group_fn=group_fn,
        attributions=attributions,
        collision_out=collision_out,
    )

    # ── incremental plan: reuse unchanged cells, target the agent at changed ──
    # Diff each freshly-gathered cell's material signature against the prior grid
    # (while ``row["material"]`` is still present): unchanged cells are reused
    # VERBATIM here (so the grid frame already shows their prior AI values), and
    # the EXTRACT/summarize passes below run only over the CHANGED set. ``force``
    # (or no prior grid) marks every cell changed → the full streamed build.
    fresh_sigs, changed = _plan_incremental_cells(rows, cols, stored, force=force)

    # Emit the deterministic grid right away — the table renders in full (every
    # non-``prompt`` column filled) while EXTRACT works. The wire copy strips the
    # transient grounding ``material``; the working ``rows`` keep it for EXTRACT.
    grid_rows = [{k: v for k, v in r.items() if k != "material"} for r in rows]
    yield {"event": "grid", "data": {
        "table_id": table_id,
        "title": title or table_id,
        "columns": cols,
        "row_axis": axis,
        "rows": grid_rows,
        "synthesis": {},
        "rows_count": len(rows),
        "columns_count": len(cols),
        "prompt_columns": len(prompt_cols),
        "spine_id": spine_id,
        "needs_attribution": list(collision_out.get("needs_attribution") or []),
        "unresolved_collision": bool(collision_out.get("unresolved_collision")),
        "collision_scan_failed": bool(collision_out.get("collision_scan_failed")),
    }}

    # ── EXTRACT (agent), streamed per row so the AI columns fill in live ───────
    if need_material:
        yield {"event": "stage", "data": {
            "phase": "extract",
            "message": "Extracting AI columns…",
            "rows": len(rows),
            "columns": len(prompt_cols),
        }}
        for r in rows:
            material = r.pop("material", "")
            row_only = (
                None if force else {ck for (rid, ck) in changed if rid == r["id"]}
            )
            # An unchanged row (no changed prompt cell) keeps the verbatim values
            # already reused into its cells — skip the agent entirely.
            if row_only is not None and not row_only:
                continue
            if not material:
                continue
            filled = await asyncio.to_thread(
                _extract_row_cells, r, material, prompt_cols, extract_fn, only=row_only
            )
            if filled:
                r["cells"].update(filled)
                yield {"event": "cells", "data": {"row_id": r["id"], "cells": filled}}
    else:
        for r in rows:
            r.pop("material", None)

    # ── per-cell summaries: agent synthesis over 2+-member cells (streamed) ────
    # Single-member cells already read as their member title (no call); a 2+-member
    # cell gets one batched call per row, emitted as updated ``cells`` so summaries
    # fill in live.
    if summarize_fn is not None:
        for r in rows:
            row_only = (
                None if force else {ck for (rid, ck) in changed if rid == r["id"]}
            )
            summarized = await asyncio.to_thread(
                _summarize_row_cells, r, cols, summarize_fn, only=row_only
            )
            if summarized:
                yield {"event": "cells", "data": {"row_id": r["id"], "cells": summarized}}

    # ── synthesize: mechanical stats + auto AI narrative (streamed per column) ─
    # The mechanical summary is free; the AI narrative auto-runs for every filled
    # column when ``summarize_fn`` is supplied (no per-column click) and is folded
    # into ``synthesis[key]["ai"]`` so it persists with the grid and survives a
    # reload. Each column streams a ``synthesis`` frame so the footer fills live.
    yield {"event": "stage", "data": {
        "phase": "synthesize", "message": "Synthesizing…", "columns": len(cols),
    }}
    # Stamp fresh signatures onto every cell, then recompute synthesis only for
    # columns with a changed cell — unchanged columns keep their prior mechanical
    # stats AND their prior AI narrative verbatim (so neither is re-run below).
    _stamp_cell_signatures(rows, cols, fresh_sigs)
    dirty_cols = _synthesis_dirty_cols(cols, rows, stored, changed, force=force)
    synthesis = _merge_synthesis(cols, rows, stored, dirty_cols)
    if summarize_fn is not None:
        for col in cols:
            entry = synthesis.get(col["key"])
            if not entry or not entry.get("filled"):
                continue
            if col["key"] not in dirty_cols:
                continue
            try:
                text = await asyncio.to_thread(
                    generate_column_synthesis, col, rows, summarize_fn=summarize_fn
                )
            except Exception:  # noqa: BLE001 — a narrative hiccup never fails the build
                text = ""
            if text:
                entry["ai"] = text
                yield {"event": "synthesis", "data": {"column_key": col["key"], "summary": text}}
    table = {
        "table_id": table_id,
        "title": title or table_id,
        "columns": cols,
        "row_axis": axis,
        "rows": rows,
        "synthesis": synthesis,
        "signature": signature,
        "schema_version": _TABLE_SCHEMA_VERSION,
        "built_signature": signature,
        "attached_note_ids": _attached_note_ids(rows),
        "generated_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        # The grid's spine origin, PERSISTED so the persisted-list endpoint and a
        # later cross-session ``peek`` carry it (old grids lack it → read DEFAULT).
        "spine_id": spine_id,
        # Blank-spine collision bucket (see :func:`build_matrix`): excluded
        # unattributable members + the unresolved flag, persisted for later peeks.
        "needs_attribution": list(collision_out.get("needs_attribution") or []),
        "unresolved_collision": bool(collision_out.get("unresolved_collision")),
        "collision_scan_failed": bool(collision_out.get("collision_scan_failed")),
    }
    await asyncio.to_thread(
        _persist_table, name, table, base, project=project, graph=graph
    )

    out = dict(table)
    out["stale"] = False
    out["cached"] = False
    out["exists"] = True
    yield {"event": "result", "data": out}