zettelkasten.citations_ops

zettelkasten.citations_ops

Zettelkasten citation-operations business logic.

Citation entity CRUD, dedup, promotion, and OpenAlex citation-graph expansion — carved out of :mod:zettelkasten.server for clarity. server re-exports every name here so existing importers keep working unchanged. Server-level graph cache helpers (_get_graph, _reloaded_graph, …) are imported at module scope; because server performs its re-export import of this module at end-of-file, those names are fully defined by the time this import runs (no cycle).

create_citation

create_citation(id: str, title: str, authors: list[str], year: int, doc_type: str = '', doi: str = '', zotero_key: str = '', source: str = '') -> str

Create a lightweight citation entity in _citations/.

Use this for referenced works that don't need a full source graph yet. Notes can link to citations via target_graph='_citations'.

When source is given, the citation id (whether freshly created or matched to an existing one via dedup) is recorded on that source's _meta.yaml references list, so the source owns a navigable list of the works it cites — the same field citation(action="expand") populates.

Parameters:

Name	Type	Description	Default
id	str	Unique citation ID (e.g. 'sharpe-1966-mutual-fund')	required
title	str	Full title of the work	required
authors	list[str]	List of author names	required
year	int	Publication year	required
doc_type	str	Document type (e.g. 'journal-article', 'book')	''
doi	str	Digital Object Identifier	''
zotero_key	str	Zotero item key	''
source	str	Optional source graph name that cites this work; the resulting citation id is appended to its references list.	''

Source code in zettelkasten/citations_ops.py

def create_citation(
    id: str,
    title: str,
    authors: list[str],
    year: int,
    doc_type: str = "",
    doi: str = "",
    zotero_key: str = "",
    source: str = "",
) -> str:
    """Create a lightweight citation entity in _citations/.

    Use this for referenced works that don't need a full source graph yet.
    Notes can link to citations via target_graph='_citations'.

    When ``source`` is given, the citation id (whether freshly created or matched
    to an existing one via dedup) is recorded on that source's ``_meta.yaml``
    ``references`` list, so the source owns a navigable list of the works it
    cites — the same field ``citation(action="expand")`` populates.

    Args:
        id: Unique citation ID (e.g. 'sharpe-1966-mutual-fund')
        title: Full title of the work
        authors: List of author names
        year: Publication year
        doc_type: Document type (e.g. 'journal-article', 'book')
        doi: Digital Object Identifier
        zotero_key: Zotero item key
        source: Optional source graph name that cites this work; the resulting
            citation id is appended to its ``references`` list.
    """
    try:
        server._validate_name(id)
    except ValueError as e:
        return json.dumps({"error": str(e), "type": "ValidationError"})

    # Keep the citation id namespace disjoint from sources: a citation must not
    # reuse an existing source's name, or links to "<id>" become ambiguous between
    # the source and the citation (the wang-2023 collision).
    if server._source_name_taken(id):
        return json.dumps({
            "error": (f"'{id}' is already a source graph; choose a different citation id to "
                      "avoid a source/citation name collision."),
            "type": "AlreadyExists",
        })

    if source:
        try:
            server._validate_name(source)
        except ValueError as e:
            return json.dumps({"error": str(e), "type": "ValidationError"})
        if not (server.GRAPHS_DIR / source / "_meta.yaml").is_file():
            return json.dumps({"error": f"Source '{source}' not found.", "type": "NotFound"})

    citations_dir = server.GRAPHS_DIR / "_citations"
    citations_dir.mkdir(parents=True, exist_ok=True)

    import yaml
    citation_data: dict = {
        "id": id,
        "title": title,
        "authors": authors,
        "year": year,
    }
    if doc_type:
        citation_data["doc_type"] = doc_type
    if doi:
        citation_data["doi"] = doi
    if zotero_key:
        citation_data["zotero_key"] = zotero_key

    # Dedup: if a citation for the same work already exists (matched on canonical
    # key — DOI/arXiv/normalized author+year+title), reuse it instead of
    # creating a duplicate node — but still link it to the source.
    fingerprint = server.canonical_citation_key(citation_data)
    for existing_id, existing in server.load_citations().items():
        if server.canonical_citation_key(existing) == fingerprint:
            linked = server._link_source_reference(source, existing_id) if source else False
            return json.dumps({
                "id": existing_id,
                "deduped": True,
                "matched_on": fingerprint,
                "source_linked": linked,
                "message": f"Citation for this work already exists as '{existing_id}'; reusing it.",
            })

    filepath = citations_dir / f"{id}.yaml"
    if filepath.exists():
        return json.dumps({"error": f"Citation '{id}' already exists.", "type": "AlreadyExists"})

    filepath.write_text(
        yaml.dump(citation_data, default_flow_style=False, allow_unicode=True), encoding="utf-8")
    server.logger.info("created citation '%s'", id)
    linked = server._link_source_reference(source, id) if source else False
    return json.dumps({
        "id": id,
        "path": str(filepath),
        "source_linked": linked,
        "message": f"Created citation '{id}'" + (f" and linked it to '{source}'" if linked else ""),
    })

dedup_citations

dedup_citations(dry_run: bool = True) -> str

Merge duplicate citation entities in _citations/.

Groups citations by canonical key (DOI > arXiv > normalized author+year+ title). For each group with more than one member, keeps a canonical record (prefers one carrying a DOI, then the shortest id), backfills missing metadata fields from the duplicates, repoints all note links from the duplicate ids to the canonical id, and deletes the duplicate files.

Parameters:

Name	Type	Description	Default
dry_run	bool	If true (default), report what would be merged without changing anything. Set false to apply.	True

Source code in zettelkasten/citations_ops.py

def dedup_citations(dry_run: bool = True) -> str:
    """Merge duplicate citation entities in _citations/.

    Groups citations by canonical key (DOI > arXiv > normalized author+year+
    title). For each group with more than one member, keeps a canonical record
    (prefers one carrying a DOI, then the shortest id), backfills missing
    metadata fields from the duplicates, repoints all note links from the
    duplicate ids to the canonical id, and deletes the duplicate files.

    Args:
        dry_run: If true (default), report what would be merged without changing
            anything. Set false to apply.
    """
    import yaml
    citations = server.load_citations()
    if not citations:
        return json.dumps({"groups": [], "message": "No citations found."})

    groups: dict[str, list[str]] = {}
    for cid, data in citations.items():
        groups.setdefault(server.canonical_citation_key(data), []).append(cid)

    plan: list[dict] = []
    remap: dict[str, str] = {}
    merges: list[tuple[str, list[str], dict]] = []
    for key, ids in groups.items():
        if len(ids) < 2:
            continue
        # Choose canonical: prefer a record with a DOI, then the shortest id.
        canonical = sorted(
            ids,
            key=lambda i: (0 if server.normalize_doi(citations[i].get("doi", "")) else 1, len(i), i),
        )[0]
        dups = [i for i in ids if i != canonical]
        merged = dict(citations[canonical])
        for d in dups:
            for k, v in citations[d].items():
                if k != "id" and v and not merged.get(k):
                    merged[k] = v
            remap[d] = canonical
        plan.append({"canonical": canonical, "merged_from": dups, "key": key})
        merges.append((canonical, dups, merged))

    if dry_run:
        return json.dumps({
            "dry_run": True,
            "duplicate_groups": len(plan),
            "plan": plan,
            "message": f"Found {len(plan)} duplicate group(s). Re-run with dry_run=false to apply.",
        })

    citations_dir = server.GRAPHS_DIR / "_citations"
    for canonical, dups, merged in merges:
        (citations_dir / f"{canonical}.yaml").write_text(
            yaml.dump(merged, default_flow_style=False, allow_unicode=True), encoding="utf-8")
        for d in dups:
            dup_path = citations_dir / f"{d}.yaml"
            if dup_path.exists():
                dup_path.unlink()
        # Mirror promote_citation's warm-index hygiene (see _invalidate_warm_citation):
        # a cold reconcile only ghost-prunes deleted ids and only re-embeds a node
        # when its backing note changes. Removed dup ids and a canonical whose
        # title/abstract was backfilled don't touch any note, so already-warm boxes
        # keep stale/ghost vectors for the process lifetime. Drop them eagerly:
        # each removed dup, and the canonical itself when its embedded text changed
        # (reconcile re-adds the canonical with a fresh vector on next access).
        for d in dups:
            server._invalidate_warm_citation(d)
        orig = citations.get(canonical) or {}
        if (merged.get("title") != orig.get("title")
                or merged.get("abstract") != orig.get("abstract")):
            server._invalidate_warm_citation(canonical)

    links_rewritten = server._rewrite_citation_links(remap)
    server.logger.info("dedup_citations merged %d groups, rewrote %d links", len(plan), links_rewritten)
    return json.dumps({
        "dry_run": False,
        "duplicate_groups": len(plan),
        "merged": plan,
        "links_rewritten": links_rewritten,
        "message": f"Merged {len(plan)} group(s); rewrote {links_rewritten} link(s).",
    })

unlink_citation

unlink_citation(citation_id: str, sources: list[str] | None = None, remove_global: bool = False, dry_run: bool = False) -> str

Remove a citation's links from the graph — scoped, or globally.

The inverse of the various paths that attach a citation (link_notes with target_graph="_citations", create_citation/expand recording it on a source's references/cited_by). Two tiers, selected by remove_global:

• Scoped unlink (remove_global=False): drop every graph="_citations" note link pointing at citation_id from the note graphs named in sources, and remove citation_id from those sources' meta reference lists. The global _citations/<id>.yaml entity is KEPT (it may still be referenced from other scopes). A WRITE, not a delete.
• Global remove (remove_global=True): scrub the links from EVERY note graph (sources + _cross) and EVERY source's meta lists, then delete the _citations/<id>.yaml file and drop the node from warm indexes. A DELETE; callers must gate it like promote/delete_note.

Parameters:

Name	Type	Description	Default
citation_id	str	the citation entity id (_citations/<id>.yaml stem).	required
sources	list[str] | None	note graphs to scrub for scoped unlink. Ignored (all graphs are scanned) when remove_global is true. Empty/None in scoped mode is a no-op for links (nothing in scope to unlink).	None
remove_global	bool	also delete the global citation entity after scrubbing.	False
dry_run	bool	report what WOULD change without writing.	False

Source code in zettelkasten/citations_ops.py

def unlink_citation(
    citation_id: str,
    sources: list[str] | None = None,
    remove_global: bool = False,
    dry_run: bool = False,
) -> str:
    """Remove a citation's links from the graph — scoped, or globally.

    The inverse of the various paths that *attach* a citation (``link_notes``
    with ``target_graph="_citations"``, ``create_citation``/``expand`` recording
    it on a source's ``references``/``cited_by``). Two tiers, selected by
    ``remove_global``:

    * **Scoped unlink** (``remove_global=False``): drop every ``graph="_citations"``
      note link pointing at ``citation_id`` from the note graphs named in
      ``sources``, and remove ``citation_id`` from those sources' meta reference
      lists. The global ``_citations/<id>.yaml`` entity is KEPT (it may still be
      referenced from other scopes). A WRITE, not a delete.
    * **Global remove** (``remove_global=True``): scrub the links from EVERY note
      graph (sources + ``_cross``) and EVERY source's meta lists, then delete the
      ``_citations/<id>.yaml`` file and drop the node from warm indexes. A DELETE;
      callers must gate it like ``promote``/``delete_note``.

    Args:
        citation_id: the citation entity id (``_citations/<id>.yaml`` stem).
        sources: note graphs to scrub for scoped unlink. Ignored (all graphs are
            scanned) when ``remove_global`` is true. Empty/None in scoped mode is
            a no-op for links (nothing in scope to unlink).
        remove_global: also delete the global citation entity after scrubbing.
        dry_run: report what WOULD change without writing.
    """
    try:
        server._validate_name(citation_id)
    except ValueError as e:
        return json.dumps({"error": str(e), "type": "ValidationError"})

    citations_dir = server.GRAPHS_DIR / "_citations"
    citation_path = citations_dir / f"{citation_id}.yaml"
    citation_exists = citation_path.is_file()

    # Resolve the graphs whose note links we scan. Global remove must reach every
    # note graph (incl. _cross) so no dangling edge survives the file deletion;
    # scoped unlink touches only the caller's sources.
    if remove_global:
        scan_graphs = list(server.note_graph_names())
    else:
        scan_graphs = [s for s in (sources or []) if isinstance(s, str) and s]

    links_removed = 0
    notes_changed: list[tuple[str, str]] = []   # (graph, note_id)
    pending_saves: list[tuple[str, object]] = []  # (graph, note) to persist
    for gname in scan_graphs:
        try:
            zg = server._get_graph(gname)
        except Exception:
            continue
        for note in zg.notes.values():
            kept = [
                link for link in note.links
                if not (link.graph == "_citations" and link.target == citation_id)
            ]
            removed = len(note.links) - len(kept)
            if removed:
                links_removed += removed
                notes_changed.append((gname, note.id))
                if not dry_run:
                    note.links = kept
                    pending_saves.append((gname, note))

    # Remove the id from source meta reference lists. For global remove, scan all
    # source folders (a source can reference a citation via meta WITHOUT a note
    # link). For scoped unlink, only the caller's sources.
    if remove_global:
        meta_scope = [n for n in server.note_graph_names() if (server.GRAPHS_DIR / n / "_meta.yaml").is_file()]
    else:
        meta_scope = scan_graphs
    metas_changed: list[str] = []
    for sname in meta_scope:
        meta = server.load_source_meta(sname)
        if citation_id in (meta.get("references") or []) or citation_id in (meta.get("cited_by") or []):
            metas_changed.append(sname)
            if not dry_run:
                server._unlink_source_reference(sname, citation_id)

    if dry_run:
        return json.dumps({
            "dry_run": True,
            "id": citation_id,
            "remove_global": remove_global,
            "links_removed": links_removed,
            "notes_changed": [nid for _g, nid in notes_changed],
            "sources_meta_changed": metas_changed,
            "would_delete_entity": bool(remove_global and citation_exists),
            "message": (
                f"Would remove {links_removed} link(s) across {len(set(g for g, _ in notes_changed))} "
                f"graph(s) and {len(metas_changed)} meta list(s)"
                + ("; would delete the global citation entity." if remove_global and citation_exists else ".")
            ),
        })

    # Persist note edits and reload each affected graph once.
    saved_graphs: set[str] = set()
    for gname, note in pending_saves:
        try:
            server._get_graph(gname).save_note(note)
            saved_graphs.add(gname)
        except Exception as exc:
            server.logger.warning("unlink_citation: failed to save note %s in %s: %s", note.id, gname, exc)
    for gname in saved_graphs:
        server._reload_graph(gname)

    deleted_entity = False
    if remove_global and citation_exists:
        try:
            citation_path.unlink()
            deleted_entity = True
            # Same warm-index hygiene as promote/dedup: the cold reconcile only
            # ghost-prunes on rebuild, so drop the node from warm boxes eagerly.
            server._invalidate_warm_citation(citation_id)
        except OSError as exc:
            return json.dumps({
                "error": f"Removed links but failed to delete citation file: {exc}",
                "type": "IOError",
                "id": citation_id,
                "links_removed": links_removed,
            })

    server.logger.info(
        "unlink_citation '%s' (global=%s): removed %d link(s), %d meta list(s), deleted=%s",
        citation_id, remove_global, links_removed, len(metas_changed), deleted_entity,
    )
    return json.dumps({
        "id": citation_id,
        "remove_global": remove_global,
        "links_removed": links_removed,
        "notes_changed": [nid for _g, nid in notes_changed],
        "sources_meta_changed": metas_changed,
        "deleted_entity": deleted_entity,
        "message": (
            f"Removed {links_removed} link(s) and cleared {len(metas_changed)} meta list(s)"
            + (" and deleted the global citation." if deleted_entity else ".")
        ),
    })

promote_citation

promote_citation(citation_id: str) -> str

Promote a citation to a full source folder.

Reads the citation metadata from _citations/{id}.yaml, creates a source folder with _meta.yaml, and removes the citation file.

Parameters:

Name	Type	Description	Default
citation_id	str	ID of the citation to promote	required

Source code in zettelkasten/citations_ops.py

def promote_citation(citation_id: str) -> str:
    """Promote a citation to a full source folder.

    Reads the citation metadata from _citations/{id}.yaml, creates a source
    folder with _meta.yaml, and removes the citation file.

    Args:
        citation_id: ID of the citation to promote
    """
    try:
        server._validate_name(citation_id)
    except ValueError as e:
        return json.dumps({"error": str(e), "type": "ValidationError"})

    citations_dir = server.GRAPHS_DIR / "_citations"
    citation_path = citations_dir / f"{citation_id}.yaml"
    if not citation_path.exists():
        return json.dumps({"error": f"Citation '{citation_id}' not found.", "type": "NotFound"})

    import yaml
    citation_data = yaml.safe_load(citation_path.read_text(encoding="utf-8"))
    if not citation_data or not isinstance(citation_data, dict):
        return json.dumps({"error": f"Citation '{citation_id}' has invalid data.", "type": "ValidationError"})

    name = citation_id
    graph_path = server.GRAPHS_DIR / name
    if graph_path.exists():
        return json.dumps({"error": f"Source folder '{name}' already exists.", "type": "AlreadyExists"})

    # exist_ok closes the check-then-create race: two near-simultaneous promotes
    # (e.g. the MCP tool and the dashboard route) both passing the exists() check
    # above would otherwise have the loser raise FileExistsError out of the tool.
    graph_path.mkdir(parents=True, exist_ok=True)
    meta_data: dict = {
        "name": name,
        "doc_type": citation_data.get("doc_type", ""),
        "title": citation_data.get("title", ""),
        "authors": citation_data.get("authors", []),
        "year": citation_data.get("year"),
        "venue": citation_data.get("venue", ""),
        "doi": citation_data.get("doi", ""),
        "abstract": citation_data.get("abstract", ""),
        "coverage": {"human": "none", "agent": "none"},
    }
    zotero_key = citation_data.get("zotero_key", "")
    if zotero_key:
        meta_data["zotero_key"] = zotero_key

    meta = graph_path / "_meta.yaml"
    try:
        meta.write_text(yaml.dump(meta_data, default_flow_style=False), encoding="utf-8")
    except OSError as e:
        if graph_path.exists():
            import shutil
            shutil.rmtree(graph_path, ignore_errors=True)
        return json.dumps({"error": f"Failed to write source folder: {e}", "type": "IOError"})

    citation_path.unlink()
    server.logger.info("promoted citation '%s' to source folder", citation_id)
    # The citation yaml is gone, so drop its node from any warm box index now;
    # otherwise reconcile would only prune it on the next cold rebuild and the
    # promoted work would double-count as both a (ghost) citation and a source.
    server._invalidate_warm_citation(citation_id)
    # Cache the new source folder as a FULLY LOADED graph (mirrors _get_graph /
    # _reload_graph). An unloaded ZettelGraph(name) with no _graph_mtimes entry is
    # served by _get_graph WITHOUT load(); touching its .embeddings then reconciles
    # against an empty notes_snapshot, pruning & persisting an EMPTY citation cache.
    # Route the republish through _reloaded_graph so it INHERITS any outgoing
    # instance's _embeddings_lock on name reuse (a source folder previously cached
    # under this name). Building a fresh instance with its own lock here would
    # reintroduce the reload double-build race _reloaded_graph exists to close.
    with server._graph_lock:
        server._reloaded_graph(name)
    return json.dumps({"name": name, "path": str(graph_path), "message": f"Promoted citation '{citation_id}' to source folder"})

expand_corpus

expand_corpus(citation_id: str = '', zotero_key: str = '') -> str

Acquire a cited-but-unowned work into the corpus, end to end.

The executable companion to rank_missing_papers: given an acquisition target (a cited-but-unowned work id in _citations/, optionally carrying a zotero_key), this ORCHESTRATES the existing primitives to bring it in — it does not reimplement any of them. When a zotero_key is available it FETCHES the PDF from the local Zotero install (:func:zotero_fetch_pdf) and INGESTS it for a stable content_hash + cached full text (:func:ingest_source), then PROMOTES the citation into an owned source folder (:func:promote_citation) and wires the full-text pointer onto the new source's meta.

Idempotent and partial-failure tolerant:

• Already owned — if a source folder for the target already exists, or its content is already owned under another source (a content_hash match reported by ingest), this is a safe no-op reporting status="already_owned" rather than a duplicate ingest or folder.
• Degraded — with no zotero_key, no local PDF, or a fetch/ingest error, the work is still promoted as a metadata-only owned source and the failure is reported in the structured result, never raised. Promotion is the only graph mutation and is atomic, so the corpus is never left half-mutated.

Returns a structured result describing each step (fetched / ingested / promoted), the resulting owned source name, and an overall status (acquired | already_owned | failed).

Parameters:

Name	Type	Description	Default
citation_id	str	Id of the cited-but-unowned work to acquire (a _citations/ entry).	''
zotero_key	str	Optional Zotero item key; falls back to the citation's stored zotero_key.	''

Source code in zettelkasten/citations_ops.py

def expand_corpus(citation_id: str = "", zotero_key: str = "") -> str:
    """Acquire a cited-but-unowned work into the corpus, end to end.

    The executable companion to ``rank_missing_papers``: given an acquisition
    target (a cited-but-unowned work id in ``_citations/``, optionally carrying a
    ``zotero_key``), this ORCHESTRATES the existing primitives to bring it in —
    it does not reimplement any of them. When a ``zotero_key`` is available it
    FETCHES the PDF from the local Zotero install (:func:`zotero_fetch_pdf`) and
    INGESTS it for a stable ``content_hash`` + cached full text
    (:func:`ingest_source`), then PROMOTES the citation into an owned source
    folder (:func:`promote_citation`) and wires the full-text pointer onto the
    new source's meta.

    Idempotent and partial-failure tolerant:

    * Already owned — if a source folder for the target already exists, or its
      content is already owned under another source (a ``content_hash`` match
      reported by ingest), this is a safe no-op reporting
      ``status="already_owned"`` rather than a duplicate ingest or folder.
    * Degraded — with no ``zotero_key``, no local PDF, or a fetch/ingest error,
      the work is still promoted as a metadata-only owned source and the failure
      is reported in the structured result, never raised. Promotion is the only
      graph mutation and is atomic, so the corpus is never left half-mutated.

    Returns a structured result describing each step (``fetched`` / ``ingested``
    / ``promoted``), the resulting owned source ``name``, and an overall
    ``status`` (``acquired`` | ``already_owned`` | ``failed``).

    Args:
        citation_id: Id of the cited-but-unowned work to acquire (a ``_citations/`` entry).
        zotero_key: Optional Zotero item key; falls back to the citation's stored ``zotero_key``.
    """
    try:
        server._validate_name(citation_id)
    except ValueError as e:
        return json.dumps({"error": str(e), "type": "ValidationError"})

    # Idempotent: a source folder already named after the target is owned → no-op.
    if (server.GRAPHS_DIR / citation_id).exists():
        return json.dumps({
            "target": citation_id,
            "status": "already_owned",
            "name": citation_id,
            "message": f"'{citation_id}' is already an owned source; nothing to acquire.",
        })

    citation_path = server.GRAPHS_DIR / "_citations" / f"{citation_id}.yaml"
    if not citation_path.exists():
        return json.dumps({
            "target": citation_id,
            "error": f"No cited-but-unowned work '{citation_id}' to acquire.",
            "type": "NotFound",
        })

    import yaml
    citation_data = yaml.safe_load(citation_path.read_text(encoding="utf-8")) or {}
    if not isinstance(citation_data, dict):
        citation_data = {}
    key = (zotero_key or citation_data.get("zotero_key") or "").strip()

    # Canonical-key idempotency: the ownership unit is the DEDUPABLE canonical
    # citation key (DOI/arXiv/identifying metadata), the same predicate
    # build_corpus merges on. If any existing source folder already carries this
    # work's dedupable key — even a metadata-only owner with no content_hash, the
    # exact state rank_missing_papers' actions used to produce — this is a no-op,
    # never a second owned folder for the same work. Runs on ALL paths (incl. the
    # degraded, no-key/fetch-fail metadata-only promote) BEFORE any fetch. A
    # degenerate/non-dedupable key never short-circuits (keeps prior behavior).
    canonical_owner = server._find_source_by_canonical_key(citation_data)
    if canonical_owner:
        return json.dumps({
            "target": citation_id,
            "status": "already_owned",
            "name": canonical_owner,
            "message": (
                f"A source for the same work is already owned as "
                f"'{canonical_owner}' (matched on canonical citation key); "
                "skipped duplicate acquisition."
            ),
        })

    fetched: dict | None = None
    ingested: dict | None = None

    if key:
        try:
            fetch_res = json.loads(
                server.zotero_fetch_pdf(key, include_text=False, include_annotations=False)
            )
        except Exception as e:
            return server._expand_failed(citation_id, "fetch", e, fetched=fetched, ingested=ingested)
        if "error" in fetch_res:
            # No local PDF / network / config error → degrade to metadata-only.
            fetched = {"ok": False, "reason": fetch_res["error"]}
        else:
            fetched = {"ok": True, "path": fetch_res.get("path", "")}
            try:
                ingest_res = json.loads(server.ingest_source(key=key))
            except Exception as e:
                return server._expand_failed(citation_id, "ingest", e, fetched=fetched, ingested=ingested)
            if "error" in ingest_res:
                ingested = {"ok": False, "reason": ingest_res["error"]}
            else:
                ingested = {
                    "ok": True,
                    "content_hash": ingest_res.get("content_hash", ""),
                    "source_path": ingest_res.get("source_path", ""),
                }
                existing = ingest_res.get("existing_source", "")
                if existing:
                    # The content is already owned under another source → no-op,
                    # never a duplicate promotion.
                    ingested["existing_source"] = existing
                    return json.dumps({
                        "target": citation_id,
                        "status": "already_owned",
                        "name": existing,
                        "fetched": fetched,
                        "ingested": ingested,
                        "message": (
                            f"Content already owned as source '{existing}'; "
                            "skipped duplicate promotion."
                        ),
                    })
    else:
        fetched = {"ok": False, "reason": "no zotero_key on target; promoting metadata-only."}

    try:
        promote_res = json.loads(server.promote_citation(citation_id))
    except Exception as e:
        return server._expand_failed(citation_id, "promote", e, fetched=fetched, ingested=ingested)
    if "error" in promote_res:
        return json.dumps({
            "target": citation_id,
            "status": "failed",
            "fetched": fetched,
            "ingested": ingested,
            "promoted": {"ok": False, "reason": promote_res["error"]},
            "message": f"Failed to promote '{citation_id}': {promote_res['error']}",
        })

    name = promote_res.get("name", citation_id)

    # Wire the full-text pointer (content_hash + source_path) onto the freshly
    # promoted source so get_fulltext can re-derive offline. Best-effort: a write
    # failure leaves the source owned (metadata-only), never half-mutated.
    if ingested and ingested.get("ok") and ingested.get("content_hash"):
        try:
            meta_path = server.GRAPHS_DIR / name / "_meta.yaml"
            meta = yaml.safe_load(meta_path.read_text(encoding="utf-8")) or {}
            if isinstance(meta, dict):
                meta["content_hash"] = ingested["content_hash"]
                if ingested.get("source_path"):
                    meta["source_path"] = ingested["source_path"]
                meta_path.write_text(
                    yaml.dump(meta, default_flow_style=False), encoding="utf-8")
                # promote_citation cached a FULLY-LOADED graph BEFORE this meta
                # rewrite and bumped _graph_mtimes to the pre-rewrite mtime. Only
                # bumping the mtime here would leave _check_auto_reload seeing
                # current==cached forever, so the in-memory source would never
                # carry content_hash/source_path. Force-reload so the cache
                # reflects the wired pointer (also refreshes _graph_mtimes).
                server._reload_graph(name)
        except Exception as e:
            ingested["meta_wire_error"] = str(e)

    return json.dumps({
        "target": citation_id,
        "status": "acquired",
        "name": name,
        "fetched": fetched,
        "ingested": ingested,
        "promoted": {"ok": True},
        "message": (
            f"Acquired '{citation_id}' as owned source '{name}'"
            + (" with full text." if ingested and ingested.get("ok") else " (metadata-only).")
        ),
    })

promote_citation_to_graph

promote_citation_to_graph(citation_id: str = '', zotero_key: str = '') -> str

Promote an OpenAlex citation into an owned source, gated on Zotero ownership.

The executable behind the dashboard's per-citation "promote to a graph" button. Where :func:expand_corpus degrades to a metadata-only promote when no PDF is available, this REQUIRES the work to already live in the user's Zotero library: it resolves a Zotero item key by matching the citation's DOI or title against the local install (or the Web API) and only then delegates to :func:expand_corpus to fetch the PDF, ingest it, promote the stub into an owned source folder (a new graph), and wire the full-text pointer. The promoted source keeps the citation id, so every source that already links this work resolves to it as owned — the citation edges are preserved for free.

When the work is NOT yet in Zotero, NOTHING is mutated: the result reports status="not_in_zotero" with the bibliographic detail the UI needs to prompt the user to add it to Zotero first, then retry.

Returns a structured result with status one of: acquired | already_owned | not_in_zotero | zotero_unconfigured | failed.

Parameters:

Name	Type	Description	Default
citation_id	str	Id of the cited-but-unowned work to promote (a _citations/ entry).	''
zotero_key	str	Optional Zotero item key; skips the library lookup when supplied.	''

Source code in zettelkasten/citations_ops.py

def promote_citation_to_graph(citation_id: str = "", zotero_key: str = "") -> str:
    """Promote an OpenAlex citation into an owned source, gated on Zotero ownership.

    The executable behind the dashboard's per-citation "promote to a graph"
    button. Where :func:`expand_corpus` degrades to a metadata-only promote when
    no PDF is available, this REQUIRES the work to already live in the user's
    Zotero library: it resolves a Zotero item key by matching the citation's DOI
    or title against the local install (or the Web API) and only then delegates
    to :func:`expand_corpus` to fetch the PDF, ingest it, promote the stub into
    an owned source folder (a new graph), and wire the full-text pointer. The
    promoted source keeps the citation id, so every source that already links
    this work resolves to it as owned — the citation edges are preserved for free.

    When the work is NOT yet in Zotero, NOTHING is mutated: the result reports
    ``status="not_in_zotero"`` with the bibliographic detail the UI needs to
    prompt the user to add it to Zotero first, then retry.

    Returns a structured result with ``status`` one of: ``acquired`` |
    ``already_owned`` | ``not_in_zotero`` | ``zotero_unconfigured`` | ``failed``.

    Args:
        citation_id: Id of the cited-but-unowned work to promote (a ``_citations/`` entry).
        zotero_key: Optional Zotero item key; skips the library lookup when supplied.
    """
    try:
        server._validate_name(citation_id)
    except ValueError as e:
        return json.dumps({"error": str(e), "type": "ValidationError"})

    # Already owned (a source folder of this name exists) → no-op, irrespective
    # of Zotero state, so we never prompt to re-add an owned work.
    if (server.GRAPHS_DIR / citation_id).exists():
        return json.dumps({
            "target": citation_id,
            "status": "already_owned",
            "name": citation_id,
            "message": f"'{citation_id}' is already an owned source; nothing to acquire.",
        })

    citation_path = server.GRAPHS_DIR / "_citations" / f"{citation_id}.yaml"
    if not citation_path.exists():
        return json.dumps({
            "target": citation_id,
            "error": f"No citation '{citation_id}' to promote.",
            "type": "NotFound",
        })

    import yaml
    citation_data = yaml.safe_load(citation_path.read_text(encoding="utf-8")) or {}
    if not isinstance(citation_data, dict):
        citation_data = {}

    # Same-work ownership under a different id (canonical DOI/arXiv/metadata key,
    # the predicate build_corpus merges on) → no-op, never a duplicate folder.
    canonical_owner = server._find_source_by_canonical_key(citation_data)
    if canonical_owner:
        return json.dumps({
            "target": citation_id,
            "status": "already_owned",
            "name": canonical_owner,
            "message": (
                f"A source for the same work is already owned as "
                f"'{canonical_owner}'; skipped duplicate acquisition."
            ),
        })

    key = (zotero_key or citation_data.get("zotero_key") or "").strip()
    if not key:
        from zettelkasten import zotero

        match = zotero.find_item(
            doi=str(citation_data.get("doi", "")),
            title=str(citation_data.get("title", "")),
            year=citation_data.get("year"),
        )
        if match.get("unconfigured"):
            return json.dumps({
                "target": citation_id,
                "status": "zotero_unconfigured",
                "title": citation_data.get("title", ""),
                "doi": citation_data.get("doi", ""),
                "openalex_id": citation_data.get("openalex_id", ""),
                "message": match.get("error", "Zotero is not configured."),
            })
        key = str(match.get("key", "") or "").strip()

    if not key:
        # In neither the citation's metadata nor the Zotero library → don't
        # promote; hand back the detail the UI needs to prompt the user to add it.
        return json.dumps({
            "target": citation_id,
            "status": "not_in_zotero",
            "title": citation_data.get("title", ""),
            "authors": citation_data.get("authors", []),
            "year": citation_data.get("year"),
            "doi": citation_data.get("doi", ""),
            "openalex_id": citation_data.get("openalex_id", ""),
            "message": (
                "This paper isn't in your Zotero library yet. Add it to Zotero "
                "(e.g. via the browser connector or by DOI), then retry to pull "
                "it in and promote it to a graph."
            ),
        })

    # In Zotero → reuse the full acquisition pipeline (fetch + ingest + promote
    # + wire full text). expand_corpus reports its own structured outcome.
    return server.expand_corpus(citation_id=citation_id, zotero_key=key)

expand_citations

expand_citations(source: str = '', doi: str = '', direction: str = 'both', limit: int = 25, dry_run: bool = False) -> str

Expand a source's citation graph via OpenAlex (free, no API key).

Resolves a work by DOI (from the source's _meta.yaml, or the doi arg), then pulls its references (backward) and/or the works that cite it (forward), minting deduplicated citation entities in _citations/. When a source graph is given, the discovered ids are recorded on its _meta.yaml under references / cited_by so the raw citation graph is queryable (e.g. by analyze_gaps) without flooding the note graph with edges.

Set OPENALEX_MAILTO to join OpenAlex's faster polite pool.

Parameters:

Name	Type	Description	Default
source	str	Source graph name; its DOI is used and results recorded on it.	''
doi	str	DOI to expand (overrides the source's DOI when provided).	''
direction	str	'backward' (references), 'forward' (citing works), or 'both'.	'both'
limit	int	Max works to fetch per direction (capped at 100 backward / 200 forward).	25
dry_run	bool	If true, fetch and report without creating citations or editing meta.	False

Source code in zettelkasten/citations_ops.py

def expand_citations(
    source: str = "",
    doi: str = "",
    direction: str = "both",
    limit: int = 25,
    dry_run: bool = False,
) -> str:
    """Expand a source's citation graph via OpenAlex (free, no API key).

    Resolves a work by DOI (from the source's _meta.yaml, or the `doi` arg),
    then pulls its references (backward) and/or the works that cite it
    (forward), minting deduplicated citation entities in _citations/. When a
    `source` graph is given, the discovered ids are recorded on its _meta.yaml
    under `references` / `cited_by` so the raw citation graph is queryable
    (e.g. by analyze_gaps) without flooding the note graph with edges.

    Set OPENALEX_MAILTO to join OpenAlex's faster polite pool.

    Args:
        source: Source graph name; its DOI is used and results recorded on it.
        doi: DOI to expand (overrides the source's DOI when provided).
        direction: 'backward' (references), 'forward' (citing works), or 'both'.
        limit: Max works to fetch per direction (capped at 100 backward / 200 forward).
        dry_run: If true, fetch and report without creating citations or editing meta.
    """
    from zettelkasten import openalex

    if direction not in {"backward", "forward", "both"}:
        return json.dumps({"error": "direction must be 'backward', 'forward', or 'both'.", "type": "ValidationError"})

    meta: dict = {}
    if source:
        try:
            server._validate_name(source)
        except ValueError as e:
            return json.dumps({"error": str(e), "type": "ValidationError"})
        meta = server.load_source_meta(source)
        if not meta and not (server.GRAPHS_DIR / source).is_dir():
            return json.dumps({"error": f"Source '{source}' not found.", "type": "NotFound"})

    use_doi = (doi or meta.get("doi") or "").strip()
    if not use_doi:
        return json.dumps({"error": "No DOI available. Pass `doi`, or set the source's _meta.yaml doi.", "type": "ValidationError"})

    work = openalex.get_work_by_doi(use_doi)
    if not work:
        return json.dumps({"error": f"OpenAlex could not resolve DOI '{use_doi}'.", "type": "NotFound"})

    discovered: list[tuple[str, dict]] = []  # (direction_label, simplified work)
    if direction in {"backward", "both"}:
        for w in openalex.get_referenced_works(work, limit=limit):
            discovered.append(("backward", w))
    if direction in {"forward", "both"}:
        for w in openalex.get_citing_works(openalex.simplify(work)["openalex_id"], limit=limit):
            discovered.append(("forward", w))

    # Seed dedup index from existing citations.
    existing = server.load_citations()
    fp_to_id: dict[str, str] = {server.canonical_citation_key(d): cid for cid, d in existing.items()}
    # ``taken`` reserves both existing citation ids AND existing source/note-graph
    # names, so a freshly minted citation id never collides with a source of the
    # same key (the wang-2023 bug: a new "Plan-and-Solve" citation grabbing the
    # "wang-2023" key already owned by the "Self-Consistency" source). On a clash
    # _citation_id_for falls through to its title-word / openalex-id suffix.
    taken: set[str] = set(existing.keys()) | set(server.note_graph_names())

    import yaml
    citations_dir = server.GRAPHS_DIR / "_citations"
    references: list[str] = []
    cited_by: list[str] = []
    created = 0
    reused = 0
    report: list[dict] = []

    for label, w in discovered:
        fp = server.canonical_citation_key(w)
        if fp in fp_to_id:
            cid = fp_to_id[fp]
            reused += 1
            is_new = False
            # Backfill fields that postdate an already-stored citation, without
            # clobbering existing data. Each field is keyed by the single citation
            # file, so there is nothing to duplicate. Only fields the freshly
            # fetched work actually carries (non-empty) AND the stored citation
            # lacks are written, so an offline/404/None response is a no-op that
            # never corrupts existing YAML.
            if not dry_run:
                existing_cit = existing.get(cid, {})
                pending = {
                    k: w.get(k)
                    for k in ("counts_by_year", "abstract", "venue")
                    if w.get(k) not in (None, "", []) and not existing_cit.get(k)
                }
                if pending:
                    cpath = citations_dir / f"{cid}.yaml"
                    if cpath.exists():
                        cdata = yaml.safe_load(cpath.read_text(encoding="utf-8")) or {}
                        if isinstance(cdata, dict):
                            changed = False
                            for k, v in pending.items():
                                if not cdata.get(k):
                                    cdata[k] = v
                                    changed = True
                            if changed:
                                cpath.write_text(
                                    yaml.dump(cdata, default_flow_style=False, allow_unicode=True),
                                    encoding="utf-8")
                                existing[cid] = cdata
        else:
            cid = server._citation_id_for(w, taken)
            taken.add(cid)
            fp_to_id[fp] = cid
            is_new = True
            if not dry_run:
                citations_dir.mkdir(parents=True, exist_ok=True)
                data = {k: v for k, v in {
                    "id": cid,
                    "title": w.get("title", ""),
                    "authors": w.get("authors", []),
                    "year": w.get("year"),
                    "doc_type": w.get("doc_type", ""),
                    "doi": w.get("doi", ""),
                    "openalex_id": w.get("openalex_id", ""),
                    "cited_by_count": w.get("cited_by_count", 0),
                    "counts_by_year": w.get("counts_by_year", []),
                    "abstract": w.get("abstract", ""),
                    "venue": w.get("venue", ""),
                }.items() if v not in (None, "", [])}
                (citations_dir / f"{cid}.yaml").write_text(
                    yaml.dump(data, default_flow_style=False, allow_unicode=True),
                    encoding="utf-8")
            created += 1
        (references if label == "backward" else cited_by).append(cid)
        report.append({"direction": label, "id": cid, "title": w.get("title", ""),
                       "cited_by_count": w.get("cited_by_count", 0), "new": is_new})

    if source and not dry_run:
        if references:
            meta["references"] = sorted(set(meta.get("references", []) + references))
        if cited_by:
            meta["cited_by"] = sorted(set(meta.get("cited_by", []) + cited_by))
        # Store the source's own per-year citation history (from the resolved work)
        # so trend/recency scoring has it locally. Backfill-only: never clobber an
        # existing value, since the source meta is the source of truth once set.
        swork = openalex.simplify(work)
        if swork.get("counts_by_year") and not meta.get("counts_by_year"):
            meta["counts_by_year"] = swork["counts_by_year"]
        (server.GRAPHS_DIR / source / "_meta.yaml").write_text(
            yaml.dump(meta, default_flow_style=False), encoding="utf-8")

    report.sort(key=lambda r: r["cited_by_count"], reverse=True)
    return json.dumps({
        "dry_run": dry_run,
        "doi": use_doi,
        "resolved_title": openalex.simplify(work)["title"],
        "backward_count": len(references),
        "forward_count": len(cited_by),
        "citations_created": created,
        "citations_reused": reused,
        "source_updated": bool(source and not dry_run),
        "top_works": report[:20],
        "message": (
            f"Expanded '{use_doi}': {len(references)} reference(s), {len(cited_by)} citing work(s); "
            f"{created} new citation(s), {reused} reused."
        ),
    })