zettelkasten.dashboard.backend.routes.claims

zettelkasten.dashboard.backend.routes.claims

Claim, claim-anatomy, and citation enrich/expand routes.

Split out of the former flat routes.py; behaviour is unchanged.

enrich_graph_citations

enrich_graph_citations(name: str, dry_run: bool = Query(False)) -> dict

Refresh OpenAlex citation metrics for a single source and its citations.

Source code in zettelkasten/dashboard/backend/routes/claims.py

@router.post("/graphs/{name}/enrich-citations")
def enrich_graph_citations(name: str, dry_run: bool = Query(False)) -> dict:
    """Refresh OpenAlex citation metrics for a single source and its citations."""
    _get_graph_structural(name)  # 404s on an unknown/unsafe name (no embedding build)
    _enrich_local_or_400(name)
    result = enrich.enrich_citation_metrics(graph=name, graphs_dir=GRAPHS_DIR, dry_run=dry_run)
    if not dry_run:
        _papers_cache.clear()
        _claims_cache.clear()
    return result

enrich_project_citations

enrich_project_citations(name: str, dry_run: bool = Query(False)) -> dict

Refresh OpenAlex citation metrics across a project's sources/citations.

Source code in zettelkasten/dashboard/backend/routes/claims.py

@router.post("/projects/{name}/enrich-citations")
def enrich_project_citations(name: str, dry_run: bool = Query(False)) -> dict:
    """Refresh OpenAlex citation metrics across a project's sources/citations."""
    _load_project_or_404(name)
    _enrich_local_or_400(name)
    result = enrich.enrich_citation_metrics(project=name, graphs_dir=GRAPHS_DIR, dry_run=dry_run)
    if not dry_run:
        _papers_cache.clear()
        _claims_cache.clear()
    return result

expand_graph_citations

expand_graph_citations(name: str, direction: str = Query('both', description='backward | forward | both'), limit: int = Query(25), dry_run: bool = Query(False)) -> dict

Grow a source's citation graph via OpenAlex and link it to the source.

Wraps citation(action="expand"): resolves the source's DOI, mints deduped stubs in _citations/, and records their ids on the source's references / cited_by lists — the same lists GET /graphs/{name}/ citations surfaces. LOCAL-ONLY (a federated/read-only source is rejected), and refused when the server runs write-disabled, mirroring the review-action barrier. Clears the papers/claims caches so refreshed scores surface.

Source code in zettelkasten/dashboard/backend/routes/claims.py

@router.post("/graphs/{name}/expand-citations")
def expand_graph_citations(
    name: str,
    direction: str = Query("both", description="backward | forward | both"),
    limit: int = Query(25),
    dry_run: bool = Query(False),
) -> dict:
    """Grow a source's citation graph via OpenAlex and link it to the source.

    Wraps ``citation(action="expand")``: resolves the source's DOI, mints
    deduped stubs in ``_citations/``, and records their ids on the source's
    ``references`` / ``cited_by`` lists — the same lists ``GET /graphs/{name}/
    citations`` surfaces. LOCAL-ONLY (a federated/read-only source is rejected),
    and refused when the server runs write-disabled, mirroring the review-action
    barrier. Clears the papers/claims caches so refreshed scores surface.
    """
    import zettelkasten.server as zk_server

    _get_graph_structural(name)  # 404s on an unknown/unsafe name (no embedding build)
    _enrich_local_or_400(name)
    if zk_server._WRITE_DISABLED:
        raise HTTPException(
            status_code=403,
            detail="Graph writes are disabled (ZK_DISABLE_WRITE); this action is unavailable.",
        )
    result = json.loads(zk_server.expand_citations(
        source=name, direction=direction, limit=limit, dry_run=dry_run,
    ))
    if isinstance(result, dict) and result.get("error"):
        raise HTTPException(status_code=400, detail=result["error"])
    if not dry_run:
        _papers_cache.clear()
        _claims_cache.clear()
    return result

expand_project_citations

expand_project_citations(name: str, direction: str = Query('both', description='backward | forward | both'), limit: int = Query(25), dry_run: bool = Query(False)) -> dict

Bulk-grow the citation graph for EVERY owned source in a project.

A scope-level wrapper over citation(action="expand"): it runs the same per-source OpenAlex expansion (mint deduped _citations/ stubs, record them on each source's references / cited_by) across all of the project's local sources, then aggregates the per-source outcomes. Sources without a resolvable DOI are skipped (not an error). LOCAL-ONLY and refused when the server runs write-disabled, mirroring the single-source route. Clears the papers/claims caches once at the end so refreshed scores surface.

Source code in zettelkasten/dashboard/backend/routes/claims.py

@router.post("/projects/{name}/expand-citations")
def expand_project_citations(
    name: str,
    direction: str = Query("both", description="backward | forward | both"),
    limit: int = Query(25),
    dry_run: bool = Query(False),
) -> dict:
    """Bulk-grow the citation graph for EVERY owned source in a project.

    A scope-level wrapper over ``citation(action="expand")``: it runs the same
    per-source OpenAlex expansion (mint deduped ``_citations/`` stubs, record them
    on each source's references / cited_by) across all of the project's local
    sources, then aggregates the per-source outcomes. Sources without a resolvable
    DOI are skipped (not an error). LOCAL-ONLY and refused when the server runs
    write-disabled, mirroring the single-source route. Clears the papers/claims
    caches once at the end so refreshed scores surface.
    """
    import zettelkasten.server as zk_server

    _load_project_or_404(name)  # 404s on an unknown/unsafe/corrupt project
    _enrich_local_or_400(name)
    if zk_server._WRITE_DISABLED:
        raise HTTPException(
            status_code=403,
            detail="Graph writes are disabled (ZK_DISABLE_WRITE); this action is unavailable.",
        )

    sources = enrich._scope_sources(name, "", GRAPHS_DIR)
    per_source: list[dict] = []
    sources_expanded = 0
    sources_skipped = 0
    citations_added = 0
    for sname in sources:
        res = json.loads(zk_server.expand_citations(
            source=sname, direction=direction, limit=limit, dry_run=dry_run,
        ))
        if isinstance(res, dict) and res.get("error"):
            # A DOI-less / unresolvable source is a skip, not a hard failure —
            # one bad source must not abort the whole bulk run.
            sources_skipped += 1
            per_source.append({"source": sname, "skipped": res["error"]})
            continue
        added = int(res.get("citations_created", 0) or 0) if isinstance(res, dict) else 0
        citations_added += added
        sources_expanded += 1
        per_source.append({"source": sname, **(res if isinstance(res, dict) else {})})

    if not dry_run and sources_expanded:
        _papers_cache.clear()
        _claims_cache.clear()

    return {
        "dry_run": dry_run,
        "scope": {"project": name},
        "sources_in_scope": len(sources),
        "sources_expanded": sources_expanded,
        "sources_skipped": sources_skipped,
        "citations_created": citations_added,
        "per_source": per_source,
        "message": (
            f"Expanded citations for {sources_expanded} of {len(sources)} source(s) "
            f"via OpenAlex; {citations_added} new citation(s) added; "
            f"{sources_skipped} source(s) skipped (no resolvable DOI)."
        ),
    }

graph_claims

graph_claims(name: str, limit: int = Query(0, ge=0, description='Max key claims to return; 0 = all')) -> dict

Claims view for a single source graph (its claims, scored and situated).

Source code in zettelkasten/dashboard/backend/routes/claims.py

@router.get("/graphs/{name}/claims")
def graph_claims(
    name: str,
    limit: int = Query(0, ge=0, description="Max key claims to return; 0 = all"),
) -> dict:
    """Claims view for a single source graph (its claims, scored and situated)."""
    _get_graph_structural(name)  # 404s on an unknown/unsafe name (claim engine is structural)
    local, repo = _resolve_ref(name)
    base = repo.zettel_dir if repo is not None else GRAPHS_DIR
    ns = (lambda s: federation.namespace_id(repo.id, s)) if repo is not None else (lambda s: s)
    loc = lambda d: _resolve_ref(d)[0]

    cov_sig = _coverage_signature([local], base)
    meta_sig = _meta_signature([local], base)
    # Graph-scope engine runs build_corpus(project="") = GLOBAL, so the key must
    # also track the citation pool and every other graph's signature.
    cit_sig = _citations_signature(base)
    corpus_sig = _global_corpus_signature(base)
    cache_key = (
        f"graph:{name}:{limit}:"
        f"{hash(_graph_sigs.get(name, ()))}:{cov_sig}:{meta_sig}:{cit_sig}:{corpus_sig}"
    )
    cached = _claims_cache_get(cache_key)
    if cached is not None:
        return cached

    # Pass the STRUCTURAL loader into the engine: assemble_claims →
    # _build_context → build_corpus/build_claim_index call ``get_graph(display)``
    # for every in-scope source, and a graph-scope claims request runs
    # build_corpus(project="") = GLOBAL, so it walks EVERY note graph. With the
    # eager ``_get_graph`` that would build (and persist) each box's kglite index
    # on the request thread — the exact freeze/SIGSEGV-prone work
    # ``_get_graph_structural`` exists to avoid. The claim/corpus engine is purely
    # structural (it reads only ``zg.notes`` + links, never ``.embeddings``), so
    # the structural loader is sufficient — mirrors the papers routes.
    payload = claims.assemble_claims(
        _get_graph_structural,
        graph=local,
        limit=limit if limit > 0 else None,
        scope_name=name,
        graphs_dir=base,
        namespace=ns,
        localize=loc,
    )
    _claims_cache_put(cache_key, payload)
    return payload

project_claims

project_claims(name: str, limit: int = Query(0, ge=0, description='Max key claims to return; 0 = all')) -> dict

Claims view across a project: the claims asserted by any of its sources.

Source code in zettelkasten/dashboard/backend/routes/claims.py

@router.get("/projects/{name}/claims")
def project_claims(
    name: str,
    limit: int = Query(0, ge=0, description="Max key claims to return; 0 = all"),
) -> dict:
    """Claims view across a project: the claims asserted by any of its sources."""
    local, repo = _resolve_ref(name)
    project_data = _load_project_or_404(name)
    base = repo.zettel_dir if repo is not None else GRAPHS_DIR
    ns = (lambda s: federation.namespace_id(repo.id, s)) if repo is not None else (lambda s: s)
    loc = lambda d: _resolve_ref(d)[0]

    sources = list(project_data.get("sources", []))
    if (base / "_cross").is_dir():
        sources.append("_cross")
    scoped_display = frame_search_sources(local, graphs_dir=base, namespace=ns)

    # Reload each scoped source BEFORE keying the cache so an out-of-band note
    # edit refreshes its `_graph_sigs` entry and changes the cache key — mirrors
    # project_papers exactly (without this the key stays frozen and serves a
    # stale payload after an MCP/editor edit).
    for display in scoped_display:
        _try_get_graph_structural(display)  # refresh _graph_sigs; no embedding build

    cov_sig = _coverage_signature(sources, base)
    meta_sig = _meta_signature([loc(d) for d in scoped_display], base)
    sig = "|".join(f"{d}:{hash(_graph_sigs.get(d, ()))}" for d in scoped_display)
    # Project scope is covered by the per-source sigs above, but the referenced
    # citations come from the global store — fold its fingerprint in too.
    cit_sig = _citations_signature(base)
    cache_key = f"project:{name}:{limit}:{sig}:{cov_sig}:{meta_sig}:{cit_sig}"
    cached = _claims_cache_get(cache_key)
    if cached is not None:
        return cached

    # Structural loader into the engine (see graph_claims): the claim/corpus
    # engine reads only ``zg.notes`` + links, so a project-scope claims request
    # must not eager-build each source's kglite index on the request thread.
    payload = claims.assemble_claims(
        _get_graph_structural,
        project=local,
        limit=limit if limit > 0 else None,
        scope_name=name,
        graphs_dir=base,
        namespace=ns,
        localize=loc,
    )
    _claims_cache_put(cache_key, payload)
    return payload

get_claim_anatomy

get_claim_anatomy(name: str, note_id: str) -> dict

Full anatomy of a single claim/finding in a source graph.

Claim ids are unique only within a graph, so the claim is identified by (name, note_id). Mirrors the get_note single-graph resolution: 404s when the note doesn't exist, isn't a claim/finding, or isn't an active (scorable) claim. The index is built over this one graph — its own evidence plus whatever cross-source links the claim's notes carry. No cache (like get_note / get_citation): a single anatomy is a one-pass read.

Source code in zettelkasten/dashboard/backend/routes/claims.py

@router.get("/graphs/{name}/claims/{note_id}")
def get_claim_anatomy(name: str, note_id: str) -> dict:
    """Full anatomy of a single claim/finding in a source graph.

    Claim ids are unique only within a graph, so the claim is identified by
    (``name``, ``note_id``). Mirrors the get_note single-graph resolution: 404s
    when the note doesn't exist, isn't a claim/finding, or isn't an active
    (scorable) claim. The index is built over this one graph — its own evidence
    plus whatever cross-source links the claim's notes carry. No cache (like
    get_note / get_citation): a single anatomy is a one-pass read.
    """
    zg = _get_graph_structural(name)  # 404s on an unknown/unsafe name (anatomy read is structural)
    note = zg.notes.get(note_id)
    if note is None or note.type not in claims.CLAIM_TYPES:
        raise HTTPException(
            status_code=404, detail=f"Claim '{note_id}' not found in '{name}'."
        )

    local, repo = _resolve_ref(name)
    base = repo.zettel_dir if repo is not None else GRAPHS_DIR
    ns = (lambda s: federation.namespace_id(repo.id, s)) if repo is not None else (lambda s: s)
    loc = lambda d: _resolve_ref(d)[0]

    # Cache the assembled anatomy in the shared claims LRU. The anatomy's paper
    # signals come from build_corpus(project="") = GLOBAL (via claim_anatomy →
    # _build_context), so — exactly like graph_claims — the key must track the
    # target graph's own note signature, its coverage/_meta, the citation pool,
    # AND every graph's signature (corpus_sig), plus the claim id. A note/coverage/
    # citation edit anywhere the corpus reads changes the key and recomputes; an
    # enrich/expand/coverage write clears _claims_cache outright. This turns a
    # repeat anatomy view of the same claim into a cache HIT instead of re-walking
    # every source graph for the importance map.
    cov_sig = _coverage_signature([local], base)
    meta_sig = _meta_signature([local], base)
    cit_sig = _citations_signature(base)
    corpus_sig = _global_corpus_signature(base)
    cache_key = (
        f"anatomy:graph:{name}:{note_id}:"
        f"{hash(_graph_sigs.get(name, ()))}:{cov_sig}:{meta_sig}:{cit_sig}:{corpus_sig}"
    )
    cached = _claims_cache_get(cache_key)
    if cached is not None:
        return cached

    # Structural loader into the engine (see graph_claims): claim_anatomy →
    # build_claim_index reads only ``zg.notes`` + links, so resolving the claim's
    # source (and any cross-linked source) must not eager-build kglite indexes on
    # the request thread.
    payload = claims.claim_anatomy(
        _get_graph_structural,
        claim_id=note_id,
        source_graph=local,
        graph=local,
        graphs_dir=base,
        namespace=ns,
        localize=loc,
    )
    # A draft (non-active) claim note passes the type check above but is excluded
    # from the scorable claim index, so the engine reports it not found → 404.
    if "error" in payload:
        raise HTTPException(status_code=404, detail=payload["error"])
    _claims_cache_put(cache_key, payload)
    return payload