zettelkasten.backfill

zettelkasten.backfill

Retroactive quote capture for already-extracted sources.

The extraction protocol now asks the agent to mint verbatim quote notes from the user's Zotero highlights (see get_extraction_protocol / _EVIDENCE_PROTOCOL in :mod:zettelkasten.server). Sources extracted before that change carry paraphrased claim/finding notes but no quote evidence, even though the user's highlights still live in Zotero. This module walks those sources, pulls their saved highlights, and materializes them as quote notes linked supports to the claim/finding each one best matches.

Design (deterministic, no LLM — mirrors the engine/agent split): - Highlights are retrieved, never generated: the body of every quote note is the verbatim annotation text and source.page comes from the Zotero pageLabel. We never fabricate a quote. - Matching a highlight to the note it backs is a cosine-similarity lookup against the source's own embedding index (the same index that powers semantic_search). A highlight links supports -> the best claim/finding above link_floor; below the floor it is still preserved as an unlinked quote note (the source folder is itself the provenance) rather than mis-attributed. - Idempotent: a highlight whose verbatim text already exists as a quote note in the source is skipped, so re-running never duplicates. - Dry-run by default (apply=False): reports exactly what it would write.

backfill_quotes

backfill_quotes(get_graph: Callable[[str], 'ZettelGraph'], *, source: str = '', apply: bool = False, link_floor: float = 0.45, link_types: tuple[str, ...] = DEFAULT_LINK_TYPES, graphs_dir: Path | None = None) -> dict[str, Any]

Mint verbatim quote notes from Zotero highlights for existing sources.

Parameters:

Name	Type	Description	Default
get_graph	Callable[[str], 'ZettelGraph']	Callable returning a live ZettelGraph by name (the server's _get_graph), so writes go through the shared cache + index.	required
source	str	A single source folder to backfill; empty = every source that has a zotero_key.	''
apply	bool	When False (default) this is a dry run that reports the plan; when True it writes the quote notes (with their supports links).	False
link_floor	float	Minimum cosine similarity to link a highlight to a claim; below it the quote is created unlinked.	0.45
link_types	tuple[str, ...]	Note types a highlight may be linked to.	DEFAULT_LINK_TYPES
graphs_dir	Path | None	Override the graphs root (tests / federation).	None

Returns a summary dict (per-source plans + totals).

Source code in zettelkasten/backfill.py

def backfill_quotes(
    get_graph: Callable[[str], "ZettelGraph"],
    *,
    source: str = "",
    apply: bool = False,
    link_floor: float = 0.45,
    link_types: tuple[str, ...] = DEFAULT_LINK_TYPES,
    graphs_dir: Path | None = None,
) -> dict[str, Any]:
    """Mint verbatim quote notes from Zotero highlights for existing sources.

    Args:
        get_graph: Callable returning a live ``ZettelGraph`` by name (the server's
            ``_get_graph``), so writes go through the shared cache + index.
        source: A single source folder to backfill; empty = every source that has
            a ``zotero_key``.
        apply: When False (default) this is a dry run that reports the plan; when
            True it writes the quote notes (with their ``supports`` links).
        link_floor: Minimum cosine similarity to link a highlight to a claim;
            below it the quote is created unlinked.
        link_types: Note types a highlight may be linked to.
        graphs_dir: Override the graphs root (tests / federation).

    Returns a summary dict (per-source plans + totals).
    """
    from zettelkasten.graph import Link, Note, load_source_meta
    from zettelkasten.zotero import fetch_pdf

    if source:
        names = [source]
    else:
        names = _source_names_with_zotero(graphs_dir)

    sources_report: list[dict[str, Any]] = []
    totals = {
        "annotations": 0,
        "linked": 0,
        "unlinked": 0,
        "skipped_existing": 0,
        "quotes": 0,  # created (apply) or planned (dry run)
    }

    for name in names:
        meta = load_source_meta(name, graphs_dir)
        key = (meta.get("zotero_key") or "").strip()
        entry: dict[str, Any] = {"source": name, "zotero_key": key}
        if not key:
            entry["error"] = "no zotero_key (only Zotero sources carry highlights)"
            sources_report.append(entry)
            continue

        fetched = fetch_pdf(key, include_text=False, include_annotations=True)
        if "error" in fetched:
            entry["error"] = fetched["error"]
            sources_report.append(entry)
            continue

        annotations = [
            a
            for a in fetched.get("annotations", [])
            if a.get("type") in _QUOTE_ANNOTATION_KINDS
            and len((a.get("text") or "").strip()) >= _MIN_QUOTE_CHARS
        ]
        entry["annotations"] = len(annotations)
        totals["annotations"] += len(annotations)

        zg = get_graph(name)
        existing = _existing_quote_norms(zg)
        source_label = meta.get("title") or name

        planned: list[dict[str, Any]] = []
        skipped = 0
        for ann in annotations:
            text = (ann.get("text") or "").strip()
            norm = _normalize(text)
            if _already_captured(norm, existing):
                skipped += 1
                continue
            # Reserve this text so two near-identical highlights in one run don't
            # both get written.
            existing.append(norm)

            match = _best_match(zg, text, link_types, link_floor)
            page = str(ann.get("page") or "")
            plan: dict[str, Any] = {
                "text": text[:160] + ("…" if len(text) > 160 else ""),
                "page": page,
                "matched": (
                    {"id": match[0], "title": match[1], "score": round(match[2], 3)}
                    if match
                    else None
                ),
            }

            if match:
                totals["linked"] += 1
            else:
                totals["unlinked"] += 1
            totals["quotes"] += 1

            if apply:
                title = _quote_title(text)
                note_id = zg.generate_unique_id(title)
                links = []
                if match:
                    links.append(
                        Link(target=match[0], relation="supports", direction="outgoing")
                    )
                # Backfilled quotes are copied verbatim from a Zotero highlight,
                # so they are grounded by construction (method "annotation").
                grounding = {"verified": True, "method": "annotation", "score": 1.0}
                if page:
                    grounding["page"] = page
                note = Note(
                    id=note_id,
                    title=title,
                    type="quote",
                    source={"book": source_label, "page": page} if page else {"book": source_label},
                    body=_quote_body(text, ann.get("comment") or ""),
                    tags=["quote", "backfill"],
                    links=links,
                    grounding=grounding,
                )
                zg.save_note(note)
                plan["id"] = note_id

            planned.append(plan)

        entry["skipped_existing"] = skipped
        totals["skipped_existing"] += skipped
        entry["created" if apply else "planned"] = planned
        sources_report.append(entry)

    return {
        "apply": apply,
        "link_floor": link_floor,
        "sources_scanned": len(names),
        "totals": totals,
        "sources": sources_report,
    }

backfill_project_cross

backfill_project_cross(get_graph: Callable[[str], 'ZettelGraph'], *, project: str = '', apply: bool = False, expand_closure: bool = True, graphs_dir: Path | None = None) -> dict[str, Any]

Write explicit cross: membership for projects from the old inference.

For every project (or the single project given): compute the cross ids the old read-path would have shown, union them with any ids already on the manifest, drop a literal _cross entry from sources (the legacy hack), and persist. apply=False (default) reports the plan without writing.

Parameters:

Name	Type	Description	Default
get_graph	Callable[[str], 'ZettelGraph']	Server _get_graph (writes/reads go through the shared cache).	required
project	str	A single project to migrate; empty = every project.	''
apply	bool	When False (default) dry-run; when True rewrite the manifests.	False
expand_closure	bool	Pull in synthesis spines wired only with intra-_cross links (recommended; otherwise a persona/profile spine can be dropped).	True
graphs_dir	Path | None	Override the graphs root (tests).	None

Returns a summary dict (per-project plans + totals).

Source code in zettelkasten/backfill.py

def backfill_project_cross(
    get_graph: Callable[[str], "ZettelGraph"],
    *,
    project: str = "",
    apply: bool = False,
    expand_closure: bool = True,
    graphs_dir: Path | None = None,
) -> dict[str, Any]:
    """Write explicit ``cross:`` membership for projects from the old inference.

    For every project (or the single ``project`` given): compute the cross ids
    the old read-path would have shown, union them with any ids already on the
    manifest, drop a literal ``_cross`` entry from ``sources`` (the legacy hack),
    and persist. ``apply=False`` (default) reports the plan without writing.

    Args:
        get_graph: Server ``_get_graph`` (writes/reads go through the shared cache).
        project: A single project to migrate; empty = every project.
        apply: When False (default) dry-run; when True rewrite the manifests.
        expand_closure: Pull in synthesis spines wired only with intra-``_cross``
            links (recommended; otherwise a persona/profile spine can be dropped).
        graphs_dir: Override the graphs root (tests).

    Returns a summary dict (per-project plans + totals).
    """
    import yaml

    from zettelkasten.graph import (
        GRAPHS_DIR,
        list_projects,
        load_project,
        project_cross_ids,
    )

    base = graphs_dir or GRAPHS_DIR
    cross_graph = get_graph("_cross") if (base / "_cross").is_dir() else None
    total_cross = len(cross_graph.notes) if cross_graph is not None else 0

    if project:
        names = [project]
    else:
        names = [str(p.get("name")) for p in list_projects(graphs_dir=base) if p.get("name")]

    projects_report: list[dict[str, Any]] = []
    totals = {"projects": 0, "cross_added": 0, "sources_dehacked": 0}

    for name in names:
        project_data = load_project(name, graphs_dir=base)
        if not project_data:
            projects_report.append({"project": name, "error": "not found"})
            continue

        raw_sources = list(project_data.get("sources") or [])
        had_cross_source = any(str(s) == "_cross" for s in raw_sources)
        clean_sources = [s for s in raw_sources if not str(s).startswith("_")]

        existing = set(project_cross_ids(project_data))
        inferred = _inferred_project_cross(
            get_graph, project_data, cross_graph,
            graphs_dir=base, expand_closure=expand_closure,
        )
        final = existing | inferred
        added = sorted(final - existing)

        entry: dict[str, Any] = {
            "project": name,
            "existing_cross": len(existing),
            "inferred_cross": len(inferred),
            "added": added,
            "final_cross": len(final),
            "dehacked_sources": had_cross_source,
        }
        # The hack showed EVERY cross note; flag how many are now intentionally
        # NOT claimed so a human can review the dry run before applying.
        if had_cross_source:
            entry["previously_shown_all"] = total_cross
            entry["dropped_from_all"] = max(0, total_cross - len(final))

        totals["projects"] += 1
        totals["cross_added"] += len(added)
        if had_cross_source:
            totals["sources_dehacked"] += 1

        if apply and (added or had_cross_source):
            project_data["sources"] = clean_sources
            project_data["cross"] = sorted(final)
            filepath = base / "_projects" / f"{name}.yaml"
            filepath.write_text(
                yaml.dump(project_data, default_flow_style=False, sort_keys=False),
                encoding="utf-8",
            )
            entry["written"] = str(filepath)

        projects_report.append(entry)

    return {
        "apply": apply,
        "expand_closure": expand_closure,
        "total_cross_notes": total_cross,
        "projects_scanned": len(names),
        "totals": totals,
        "projects": projects_report,
    }