zettelkasten.graph_io

zettelkasten.graph_io

Loader/writer free functions for the Zettelkasten store.

Split out of :mod:zettelkasten.graph along the data-model / IO seam: the model (Note/Link/ZettelGraph and serialization) stays in graph.py, while the atomic-write helpers and the special-folder loaders/writers (_meta.yaml, _citations/, _projects/, _reviews/) live here.

Every name defined here is also re-exported from zettelkasten.graph (with identity preserved) so existing from zettelkasten.graph import ... callers keep working unchanged.

Circular-import note: this module must NOT import zettelkasten.graph at top level. graph.py imports this module at the bottom (after the model classes are defined) to build its re-export shims; if this module imported graph at top level, the graph-defines-shims step would race a half-initialized graph_io in the import graph_io-first order. Instead we reach the model-owning module lazily at CALL time via :func:_graph — which also means GRAPHS_DIR monkeypatched on zettelkasten.graph is observed here, preserving the previous single-module behaviour.

ExtractionContextNotFound

Bases: Exception

A FULL extraction context was supplied but no keyed entry matched.

Raised by :func:resolve_extraction_context when a FULL context key is demanded (BOTH project AND synthesis_graph supplied) AND the source carries a NON-EMPTY contexts list yet none of its entries match the requested (project, synthesis_graph) pair. With a populated contexts list, silently falling back to the flat block would enforce the write against ANOTHER context's rubric (the original multi-context P1 we must not regress), so this fails loud instead. Callers (add_note / attach_to_dimension / coverage) catch this and surface a clean, structured tool error rather than crashing the run.

This is NOT raised for a PARTIAL pair (exactly one of project / synthesis_graph supplied): a partial pair can NEVER match a keyed entry, so failing loud would buy nothing and only break the legitimate no-structure call (which passes NEITHER key) when it accidentally carries one half. A partial pair degrades to the flat block instead -- treated exactly like a no-context read. It is likewise NOT raised for a source with NO contexts list (pure-legacy / flat-only meta, or a best-effort keyed write that never landed): such a source degrades gracefully to its flat block exactly as before. A miss with NO context supplied is likewise not this error.

Source code in zettelkasten/graph_io.py

class ExtractionContextNotFound(Exception):
    """A FULL extraction context was supplied but no keyed entry matched.

    Raised by :func:`resolve_extraction_context` when a FULL context key is
    demanded (BOTH ``project`` AND ``synthesis_graph`` supplied) AND the source
    carries a NON-EMPTY ``contexts`` list yet none of its entries match the
    requested ``(project, synthesis_graph)`` pair. With a populated ``contexts``
    list, silently falling back to the flat block would enforce the write against
    ANOTHER context's rubric (the original multi-context P1 we must not regress),
    so this fails loud instead. Callers (``add_note`` / ``attach_to_dimension`` /
    coverage) catch this and surface a clean, structured tool error rather than
    crashing the run.

    This is NOT raised for a PARTIAL pair (exactly one of ``project`` /
    ``synthesis_graph`` supplied): a partial pair can NEVER match a keyed entry,
    so failing loud would buy nothing and only break the legitimate no-structure
    call (which passes NEITHER key) when it accidentally carries one half. A
    partial pair degrades to the flat block instead -- treated exactly like a
    no-context read. It is likewise NOT raised for a source with NO ``contexts``
    list (pure-legacy / flat-only meta, or a best-effort keyed write that never
    landed): such a source degrades gracefully to its flat block exactly as
    before. A miss with NO context supplied is likewise not this error.
    """

    # Cap the rendered/surfaced available-context list so an absurdly long
    # ``contexts`` list cannot bloat the error message or the structured tool error.
    _AVAILABLE_CAP = 20

    def __init__(
        self,
        source: str,
        project: str,
        synthesis_graph: str,
        available: list[dict] | None = None,
    ):
        self.source = source
        self.project = project
        self.synthesis_graph = synthesis_graph
        # The (project, synthesis_graph) keys this source DOES carry, so a mis-keyed
        # scribe/operator can self-correct. Capped to ``_AVAILABLE_CAP`` (a populated
        # contexts list is the only way to reach this error, so this is non-empty).
        self.available: list[dict] = [
            {"project": str(a.get("project")),
             "synthesis_graph": str(a.get("synthesis_graph"))}
            for a in (available or [])
            if isinstance(a, dict)
        ][: self._AVAILABLE_CAP]
        self.available_truncated = len(available or []) > self._AVAILABLE_CAP
        avail_msg = ""
        if self.available:
            rendered = ", ".join(
                f"(project='{a['project']}', synthesis_graph='{a['synthesis_graph']}')"
                for a in self.available
            )
            if self.available_truncated:
                rendered += f", … (+{len(available or []) - len(self.available)} more)"
            avail_msg = f" Available contexts for this source: {rendered}."
        super().__init__(
            f"Requested extraction context (project='{project}', "
            f"synthesis_graph='{synthesis_graph}') not found for source "
            f"'{source}'. The source has extraction meta but no entry for this "
            f"context; refusing to silently fall back to another context's rubric."
            f"{avail_msg}"
        )

atomic_write_text

atomic_write_text(path: Path, text: str) -> None

Write text to path crash-safely (temp file + fsync + rename).

The temp file is created in the SAME directory as path so the final os.replace is an atomic rename on the same filesystem. The file's contents are flushed and os.fsync'd before the rename, and the PARENT DIRECTORY is fsync'd after, so a crash can never leave a half-written manifest/note nor lose the directory entry of a newly-created file. The temp file is cleaned up on failure. Mirrors the temp+rename precedent in config.write_config and embeddings.save_cache, adding the fsyncs for durability.

Source code in zettelkasten/graph_io.py

def atomic_write_text(path: Path, text: str) -> None:
    """Write ``text`` to ``path`` crash-safely (temp file + fsync + rename).

    The temp file is created in the SAME directory as ``path`` so the final
    ``os.replace`` is an atomic rename on the same filesystem. The file's
    contents are flushed and ``os.fsync``'d before the rename, and the PARENT
    DIRECTORY is fsync'd after, so a crash can never leave a half-written
    manifest/note nor lose the directory entry of a newly-created file. The temp
    file is cleaned up on failure. Mirrors the temp+rename precedent in
    ``config.write_config`` and ``embeddings.save_cache``, adding the fsyncs for
    durability.
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(f"{path.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp")
    try:
        with open(tmp, "w", encoding="utf-8") as fh:
            fh.write(text)
            fh.flush()
            os.fsync(fh.fileno())
        os.replace(tmp, path)
        # Best-effort: fsync the parent directory so the new directory entry
        # created by the rename is itself durable. Windows cannot fsync a
        # directory (and O_RDONLY on a dir may fail there), so swallow any error
        # — this is a pure no-op on those platforms and must never raise.
        try:
            dir_fd = os.open(str(path.parent), os.O_RDONLY)
            try:
                os.fsync(dir_fd)
            finally:
                os.close(dir_fd)
        except (OSError, AttributeError):
            pass
    except Exception:
        try:
            tmp.unlink(missing_ok=True)
        except OSError:
            pass
        raise

write_yaml_atomic

write_yaml_atomic(path: Path, data) -> None

Dump data to YAML and write it atomically via :func:atomic_write_text.

Uses default_flow_style=False, sort_keys=False to match the existing manifest/meta YAML style in this package.

Source code in zettelkasten/graph_io.py

def write_yaml_atomic(path: Path, data) -> None:
    """Dump ``data`` to YAML and write it atomically via :func:`atomic_write_text`.

    Uses ``default_flow_style=False, sort_keys=False`` to match the existing
    manifest/meta YAML style in this package.
    """
    atomic_write_text(
        path,
        yaml.dump(data, default_flow_style=False, allow_unicode=True, sort_keys=False),
    )

load_source_meta

load_source_meta(name: str, graphs_dir: Path | None = None) -> dict

Load a source graph's _meta.yaml as a dict (empty if absent).

Centralizes the bibliographic/coverage metadata read that several callers previously did inline. Returns {} for a missing file or malformed YAML so callers can treat metadata as best-effort.

Source code in zettelkasten/graph_io.py

def load_source_meta(name: str, graphs_dir: Path | None = None) -> dict:
    """Load a source graph's ``_meta.yaml`` as a dict (empty if absent).

    Centralizes the bibliographic/coverage metadata read that several callers
    previously did inline. Returns ``{}`` for a missing file or malformed YAML
    so callers can treat metadata as best-effort.
    """
    base = graphs_dir or _graphs_dir()
    meta_path = safe_join(base / name, "_meta.yaml")
    if not meta_path.exists():
        return {}
    data = yaml.safe_load(meta_path.read_text(encoding="utf-8"))
    return data if isinstance(data, dict) else {}

parse_date_int

parse_date_int(value: object, *, end: bool = False) -> int

Coerce a source date value to an inclusive-comparable YYYYMMDD int.

Accepts a datetime.date/datetime.datetime (YAML parses date: 2026-01-15 to a date), an ISO-ish string (2026-01-15, 2026/01, 2026), or an int (20260115 passed through; a bare 2026 treated as a year). Returns 0 when nothing parseable is present.

Coarse values fill their missing components with the START of the period by default (year 2026 -> 20260101, month 2026-03 -> 20260301). Pass end=True to fill with the END instead (20261231 / 20260331), so a date-range upper bound built from a year/month is inclusive of the whole period. (Day is clamped to 31 for an end bound; the store's dates are real calendar days, so a numeric <= YYYYMM31 comparison is correct.)

Source code in zettelkasten/graph_io.py

def parse_date_int(value: object, *, end: bool = False) -> int:
    """Coerce a source date value to an inclusive-comparable ``YYYYMMDD`` int.

    Accepts a ``datetime.date``/``datetime.datetime`` (YAML parses ``date:
    2026-01-15`` to a ``date``), an ISO-ish string (``2026-01-15``, ``2026/01``,
    ``2026``), or an int (``20260115`` passed through; a bare ``2026`` treated as
    a year). Returns ``0`` when nothing parseable is present.

    Coarse values fill their missing components with the START of the period by
    default (year ``2026`` -> ``20260101``, month ``2026-03`` -> ``20260301``).
    Pass ``end=True`` to fill with the END instead (``20261231`` / ``20260331``),
    so a date-range *upper* bound built from a year/month is inclusive of the
    whole period. (Day is clamped to ``31`` for an end bound; the store's dates
    are real calendar days, so a numeric ``<= YYYYMM31`` comparison is correct.)
    """
    import datetime as _dt

    if value is None:
        return 0
    if isinstance(value, _dt.datetime):
        value = value.date()
    if isinstance(value, _dt.date):
        return value.year * 10000 + value.month * 100 + value.day

    hi_month, hi_day = (12, 31) if end else (1, 1)
    if isinstance(value, int):
        # A full YYYYMMDD passes through; a bare year fills to period start/end.
        if value >= 10000000:
            return value
        if 1000 <= value <= 9999:
            return value * 10000 + hi_month * 100 + hi_day
        return 0
    text = str(value).strip()
    if not text:
        return 0
    nums = re.findall(r"\d+", text)
    if not nums:
        return 0
    year = int(nums[0][:4]) if nums[0] else 0
    if not (1000 <= year <= 9999):
        return 0
    if len(nums) > 1 and 1 <= int(nums[1]) <= 12:
        month = int(nums[1])
    else:
        month = hi_month
    if len(nums) > 2 and 1 <= int(nums[2]) <= 31:
        day = int(nums[2])
    else:
        day = 31 if end else 1
    return year * 10000 + month * 100 + day

source_date_int

source_date_int(name: str, graphs_dir: Path | None = None) -> int

A source box's date as a YYYYMMDD int, 0 when unknown.

Reads the box's _meta.yaml and prefers a fine date field (finer than the coarse year), falling back to year so existing corpora that only carry a publication year still get a usable (year-granular) date. Used both to stamp the global ANN index and to drive the calendar-recency rerank channel; all notes in a box share the source date, so this is one lookup per box.

Source code in zettelkasten/graph_io.py

def source_date_int(name: str, graphs_dir: Path | None = None) -> int:
    """A source box's date as a ``YYYYMMDD`` int, ``0`` when unknown.

    Reads the box's ``_meta.yaml`` and prefers a fine ``date`` field (finer than
    the coarse ``year``), falling back to ``year`` so existing corpora that only
    carry a publication year still get a usable (year-granular) date. Used both to
    stamp the global ANN index and to drive the calendar-recency rerank channel;
    all notes in a box share the source date, so this is one lookup per box.
    """
    meta = load_source_meta(name, graphs_dir=graphs_dir)
    if not meta:
        return 0
    if meta.get("date") is not None:
        d = parse_date_int(meta.get("date"))
        if d:
            return d
    return parse_date_int(meta.get("year"))

write_coverage

write_coverage(source: str, *, agent: str = '', human: str = '', graphs_dir: Path | None = None) -> dict

Set a source's coverage.{agent,human} levels in its _meta.yaml.

The single coverage-write seam, shared by the set_coverage MCP tool and the dashboard's coverage endpoint so the two can never diverge. An empty agent/human leaves that axis unchanged; both default to "none" when the source has no recorded coverage yet. Returns the resulting coverage dict.

Raises ValueError for a level outside :data:COVERAGE_LEVELS and FileNotFoundError for a source with no _meta.yaml (i.e. not a real source folder). The source name is path-jailed via :func:safe_join.

Source code in zettelkasten/graph_io.py

def write_coverage(
    source: str,
    *,
    agent: str = "",
    human: str = "",
    graphs_dir: Path | None = None,
) -> dict:
    """Set a source's ``coverage.{agent,human}`` levels in its ``_meta.yaml``.

    The single coverage-write seam, shared by the ``set_coverage`` MCP tool and
    the dashboard's coverage endpoint so the two can never diverge. An empty
    ``agent``/``human`` leaves that axis unchanged; both default to ``"none"``
    when the source has no recorded coverage yet. Returns the resulting coverage
    dict.

    Raises ``ValueError`` for a level outside :data:`COVERAGE_LEVELS` and
    ``FileNotFoundError`` for a source with no ``_meta.yaml`` (i.e. not a real
    source folder). The source name is path-jailed via :func:`safe_join`.
    """
    coverage_levels = _graph().COVERAGE_LEVELS
    valid = {"", *coverage_levels}
    if agent not in valid or human not in valid:
        raise ValueError(f"coverage must be one of: {', '.join(coverage_levels)}.")
    base = graphs_dir or _graphs_dir()
    meta_path = safe_join(base / source, "_meta.yaml")
    if not meta_path.exists():
        raise FileNotFoundError(f"Source '{source}' not found.")
    meta = yaml.safe_load(meta_path.read_text(encoding="utf-8")) or {}
    coverage = meta.get("coverage") or {"human": "none", "agent": "none"}
    if agent:
        coverage["agent"] = agent
    if human:
        coverage["human"] = human
    meta["coverage"] = coverage
    meta_path.write_text(yaml.dump(meta, default_flow_style=False), encoding="utf-8")
    return coverage

write_extraction_meta

write_extraction_meta(source: str, *, project: str | None = None, synthesis_graph: str | None = None, schema: str = '', strict: bool | None = None, grounded: bool | None = None, tags: list[str] | None = None, structure: dict | None = None, graphs_dir: Path | None = None) -> dict

Persist a grounded-extraction run's rubric onto a source's _meta.yaml.

Writes (and shallow-merges into) an extraction block so the zettelkasten server can enforce the run's contract at WRITE time -- the seam that makes strict-tag enforcement (add_note), deterministic spine attachment (attach_to_dimension), and the coverage matrix (source coverage) model-independent rather than reliant on the agent doing the right thing.

A single source can be extracted under MULTIPLE contexts -- several schemas in one project (each minting its own synthesis spine) and/or several projects over the SAME source graph. A single flat extraction block keyed only by the source name made these contexts clobber each other last-write-wins. So the block is now a COLLECTION keyed by (project, synthesis_graph):

extraction:
  # legacy flat keys (the no-context write target + read fallback)
  schema: ... / strict: ... / grounded: ... / tags: [...] / structure: {...}
  # one record per (project, synthesis_graph) context
  contexts:
    - {project, synthesis_graph, schema, strict, grounded, tags, structure}
    - ...

Behaviour:

• When BOTH project and synthesis_graph are given, the fields are written/merged into the matching contexts record (matched on the pair). A record created for the first time starts EMPTY (only its identity key) and is NOT seeded from the flat block: a keyed entry must carry ITS OWN schema's rubric, written by this run's keyed writes, not whatever the (possibly another context's) flat block currently holds. An EXISTING record is merged in place (no duplicate appended). When the MERGED POST-WRITE entry HOLDS a structure (it carried one before, or this write added one), it is mirrored into the flat block (the no-context fallback view) with PERMISSIVE per-leaf semantics: structure LAST-WRITE-WINS (so a no-context attach_to_dimension resolves the most recent spine); tags are UNIONED into the flat block's existing tags (order-preserving dedup) so the flat block ACCUMULATES the full vocabulary across every context and a context-less off-spine claim LANDS rather than being rejected; schema / strict / grounded are PRESERVED when the flat block already holds them (only set from the keyed entry when absent), keeping the named-schema fallback rubric prep_sources wrote. Gating on the merged entry (rather than on this call passing structure) keeps the promoted-spine TWO-STEP write working: a later rubric-only write (reconcile_meta_tags, no structure arg) over a context whose earlier write already landed a structure RE-FIRES the mirror so the spine's tags join the flat union. A rubric-only keyed write to a context that NEVER had a structure leaves the flat block untouched, preserving a still-flat context's own tags. The union only ever GROWS the flat allowed set, so it can never reject a previously-valid claim (the original multi-schema coherence bug stays fixed).
• When context is ABSENT (either is falsy) the LEGACY FLAT block is written/ merged EXACTLY as before -- mandatory back-compat for existing on-disk meta and the no-context callers/tests.

Each call only overwrites the keys it is given (a None/empty value leaves the existing key untouched), within whichever record it targets. Returns the record that was written (the flat block, or the context entry).

Raises FileNotFoundError for a source with no _meta.yaml (i.e. not a real source folder). The source name is path-jailed via :func:safe_join, mirroring :func:write_coverage.

Source code in zettelkasten/graph_io.py

def write_extraction_meta(
    source: str,
    *,
    project: str | None = None,
    synthesis_graph: str | None = None,
    schema: str = "",
    strict: bool | None = None,
    grounded: bool | None = None,
    tags: list[str] | None = None,
    structure: dict | None = None,
    graphs_dir: Path | None = None,
) -> dict:
    """Persist a grounded-extraction run's rubric onto a source's ``_meta.yaml``.

    Writes (and shallow-merges into) an ``extraction`` block so the zettelkasten
    server can enforce the run's contract at WRITE time -- the seam that makes
    strict-tag enforcement (``add_note``), deterministic spine attachment
    (``attach_to_dimension``), and the coverage matrix (``source coverage``)
    model-independent rather than reliant on the agent doing the right thing.

    A single source can be extracted under MULTIPLE contexts -- several schemas
    in one project (each minting its own synthesis spine) and/or several projects
    over the SAME source graph. A single flat ``extraction`` block keyed only by
    the source name made these contexts clobber each other last-write-wins. So
    the block is now a COLLECTION keyed by ``(project, synthesis_graph)``:

        extraction:
          # legacy flat keys (the no-context write target + read fallback)
          schema: ... / strict: ... / grounded: ... / tags: [...] / structure: {...}
          # one record per (project, synthesis_graph) context
          contexts:
            - {project, synthesis_graph, schema, strict, grounded, tags, structure}
            - ...

    Behaviour:

    * When BOTH ``project`` and ``synthesis_graph`` are given, the fields are
      written/merged into the matching ``contexts`` record (matched on the pair).
      A record created for the first time starts EMPTY (only its identity key) and
      is NOT seeded from the flat block: a keyed entry must carry ITS OWN schema's
      rubric, written by this run's keyed writes, not whatever the (possibly
      another context's) flat block currently holds. An EXISTING record is merged
      in place (no duplicate appended). When the MERGED POST-WRITE entry HOLDS a
      ``structure`` (it carried one before, or this write added one), it is
      mirrored into the flat block (the no-context fallback view) with PERMISSIVE
      per-leaf semantics: ``structure`` LAST-WRITE-WINS (so a no-context
      ``attach_to_dimension`` resolves the most recent spine); ``tags`` are UNIONED
      into the flat block's existing tags (order-preserving dedup) so the flat
      block ACCUMULATES the full vocabulary across every context and a context-less
      off-spine claim LANDS rather than being rejected; ``schema`` / ``strict`` /
      ``grounded`` are PRESERVED when the flat block already holds them (only set
      from the keyed entry when absent), keeping the named-schema fallback rubric
      ``prep_sources`` wrote. Gating on the merged entry (rather than on this call
      passing ``structure``) keeps the promoted-spine TWO-STEP write working: a
      later rubric-only write (``reconcile_meta_tags``, no ``structure`` arg) over a
      context whose earlier write already landed a structure RE-FIRES the mirror so
      the spine's tags join the flat union. A rubric-only keyed write to a context
      that NEVER had a structure leaves the flat block untouched, preserving a
      still-flat context's own tags. The union only ever GROWS the flat allowed
      set, so it can never reject a previously-valid claim (the original
      multi-schema coherence bug stays fixed).
    * When context is ABSENT (either is falsy) the LEGACY FLAT block is written/
      merged EXACTLY as before -- mandatory back-compat for existing on-disk meta
      and the no-context callers/tests.

    Each call only overwrites the keys it is given (a ``None``/empty value leaves
    the existing key untouched), within whichever record it targets. Returns the
    record that was written (the flat block, or the context entry).

    Raises ``FileNotFoundError`` for a source with no ``_meta.yaml`` (i.e. not a
    real source folder). The source name is path-jailed via :func:`safe_join`,
    mirroring :func:`write_coverage`.
    """
    base = graphs_dir or _graphs_dir()
    meta_path = safe_join(base / source, "_meta.yaml")
    if not meta_path.exists():
        raise FileNotFoundError(f"Source '{source}' not found.")
    meta = yaml.safe_load(meta_path.read_text(encoding="utf-8")) or {}
    block = meta.get("extraction")
    if not isinstance(block, dict):
        block = {}

    fields = dict(schema=schema, strict=strict, grounded=grounded,
                  tags=tags, structure=structure)

    if project and synthesis_graph:
        contexts = block.get("contexts")
        if not isinstance(contexts, list):
            contexts = []
        target: dict | None = None
        for entry in contexts:
            if (isinstance(entry, dict)
                    and str(entry.get("project")) == str(project)
                    and str(entry.get("synthesis_graph")) == str(synthesis_graph)):
                target = entry
                break
        if target is None:
            # A NEW context starts EMPTY (only its identity key). It must carry
            # ITS OWN rubric -- written explicitly by this run's keyed writes
            # (``prep_spine`` / ``reconcile_meta_tags``) -- and must NOT
            # inherit whatever the flat block currently holds: in a multi-schema
            # run the flat block holds the FIRST schema's tags, so seeding from it
            # would give every later context the wrong vocabulary and REJECT valid
            # claims. The flat block is left intact for any legacy no-context
            # reader; it is no longer a seed source for keyed entries.
            target = {"project": str(project),
                      "synthesis_graph": str(synthesis_graph)}
            contexts.append(target)
        _apply_extraction_fields(target, **fields)
        block["contexts"] = contexts
        # Mirror this keyed entry into the FLAT block so the no-context fallback
        # view stays USABLE for every context that shares this source -- but mirror
        # each leaf with the semantics that keep the flat block PERMISSIVE rather
        # than collapsing it to one spine's narrow rubric. The mirror is gated on
        # the MERGED POST-WRITE entry HOLDING a ``structure`` (``"structure" in
        # target``), NOT on this call passing a ``structure`` arg, so the
        # promoted-spine TWO-STEP write still works: step 1
        # (``_persist_structure_meta``) writes structure-only, so the merged entry
        # gains ``structure`` and the mirror fires; step 2 (``reconcile_meta_tags``)
        # writes tags+rubric with NO ``structure`` arg, but the merged entry STILL
        # holds the step-1 structure, so the mirror RE-FIRES. A rubric-only keyed
        # write to a context that NEVER had a structure leaves the flat block alone
        # (``"structure"`` absent from the merged entry), preserving a still-flat
        # context's tags/structure.
        #
        # Per-leaf semantics (the permissive-flat fix):
        #
        # * ``structure`` -- LAST-WRITE-WINS mirror (as before): the flat block is
        #   the no-context fallback view, so a no-context ``attach_to_dimension``
        #   resolves the most recently materialized spine.
        # * ``tags`` -- UNION the keyed entry's tags INTO the flat block's existing
        #   tags (order-preserving dedup: keep flat's existing order, append the new
        #   ones). Flat thus ACCUMULATES the FULL vocabulary across every
        #   spine/schema that touched this source and NEVER shrinks to one spine's
        #   narrow set, so a context-less off-spine claim against the flat block
        #   LANDS instead of being rejected. Because the union only ever GROWS the
        #   allowed set, it can never reject a previously-valid claim -- the original
        #   multi-schema coherence bug (schema-A's narrow flat tags rejecting a
        #   schema-B claim) stays fixed, since both schemas' tags are now present.
        # * ``schema`` / ``strict`` / ``grounded`` -- PRESERVE what the flat block
        #   ALREADY holds (the named-schema rubric ``prep_sources`` wrote); only SET
        #   them from the keyed entry when the flat block LACKS them (a flat-only
        #   source that never had a prep write). Overwriting them would replace the
        #   named-schema fallback rubric with one spine's narrow rubric.
        if "structure" in target:
            block["structure"] = target["structure"]
            keyed_tags = target.get("tags")
            if isinstance(keyed_tags, list):
                existing = block.get("tags")
                merged = list(existing) if isinstance(existing, list) else []
                seen = set(merged)
                for t in keyed_tags:
                    if t not in seen:
                        merged.append(t)
                        seen.add(t)
                block["tags"] = merged
            for k in ("schema", "strict", "grounded"):
                if k in target and k not in block:
                    block[k] = target[k]
        meta["extraction"] = block
        write_yaml_atomic(meta_path, meta)
        return target

    # No context → legacy flat block, byte-for-byte as before (the ``contexts``
    # list, if any, is left untouched alongside the flat leaves).
    _apply_extraction_fields(block, **fields)
    meta["extraction"] = block
    write_yaml_atomic(meta_path, meta)
    return block

prune_extraction_context

prune_extraction_context(source: str, *, project: str, synthesis_graph: str, graphs_dir: Path | None = None) -> bool

Remove an ORPHANED non-enforcing keyed (project, synthesis_graph) context.

Cleans up the structure-only keyed block that :func:coordinator.extraction.materialize_structure writes OPTIMISTICALLY, before reconcile decides membership. When a spine ends up UNUSED (no source joined its enforcing rubric) -- or a single source is DEMOTED to flat for a surviving spine -- that keyed context entry is dead weight; the run's reconcile step prunes it here.

Scope is deliberately narrow and SAFE:

• Removes ONLY the matching contexts entry, and ONLY when it is a NON-ENFORCING orphan -- i.e. it carries NO truthy strict rubric. An entry holding strict was enrolled by some run's ENFORCING keyed write (possibly a PRIOR run over the same source); removing it would strand a real enrollment, so it is PRESERVED and this returns False. (Unused spines only arise on strict/enforcing runs, so this guard is what makes cross-run pruning safe.)
• The PERMISSIVE flat tag union is LEFT INTACT -- it only ever grows by design (so a context-less off-spine claim still lands flat); pruning a dead spine's bookkeeping must never silently tighten the flat substrate.
• The flat structure mirror (the no-context attach_to_dimension fallback) is re-pointed to the most-recent REMAINING structure-bearing context, or dropped when none remains, so a no-context attach never resolves against the pruned spine.

Best-effort: a missing _meta.yaml, no extraction block, an absent context, or an enforcing match is a silent no-op. Returns True iff an entry was removed.

Source code in zettelkasten/graph_io.py

def prune_extraction_context(
    source: str,
    *,
    project: str,
    synthesis_graph: str,
    graphs_dir: Path | None = None,
) -> bool:
    """Remove an ORPHANED non-enforcing keyed ``(project, synthesis_graph)`` context.

    Cleans up the structure-only keyed block that
    :func:`coordinator.extraction.materialize_structure` writes OPTIMISTICALLY,
    before reconcile decides membership. When a spine ends up UNUSED (no source
    joined its enforcing rubric) -- or a single source is DEMOTED to flat for a
    surviving spine -- that keyed context entry is dead weight; the run's
    reconcile step prunes it here.

    Scope is deliberately narrow and SAFE:

    * Removes ONLY the matching ``contexts`` entry, and ONLY when it is a
      NON-ENFORCING orphan -- i.e. it carries NO truthy ``strict`` rubric. An
      entry holding ``strict`` was enrolled by some run's ENFORCING keyed write
      (possibly a PRIOR run over the same source); removing it would strand a
      real enrollment, so it is PRESERVED and this returns ``False``. (Unused
      spines only arise on strict/enforcing runs, so this guard is what makes
      cross-run pruning safe.)
    * The PERMISSIVE flat tag union is LEFT INTACT -- it only ever grows by
      design (so a context-less off-spine claim still lands flat); pruning a dead
      spine's bookkeeping must never silently tighten the flat substrate.
    * The flat ``structure`` mirror (the no-context ``attach_to_dimension``
      fallback) is re-pointed to the most-recent REMAINING structure-bearing
      context, or dropped when none remains, so a no-context attach never
      resolves against the pruned spine.

    Best-effort: a missing ``_meta.yaml``, no ``extraction`` block, an absent
    context, or an enforcing match is a silent no-op. Returns ``True`` iff an
    entry was removed.
    """
    if not project or not synthesis_graph:
        return False
    base = graphs_dir or _graphs_dir()
    try:
        meta_path = safe_join(base / source, "_meta.yaml")
    except Exception:  # noqa: BLE001 — an unvalidatable source name has nothing to prune
        return False
    if not meta_path.exists():
        return False
    meta = yaml.safe_load(meta_path.read_text(encoding="utf-8")) or {}
    block = meta.get("extraction")
    if not isinstance(block, dict):
        return False
    contexts = block.get("contexts")
    if not isinstance(contexts, list) or not contexts:
        return False

    matched: dict | None = None
    kept: list = []
    for entry in contexts:
        if (isinstance(entry, dict)
                and str(entry.get("project")) == str(project)
                and str(entry.get("synthesis_graph")) == str(synthesis_graph)):
            matched = entry
        else:
            kept.append(entry)
    if matched is None:
        return False
    # SAFETY GUARD: never remove an ENFORCING enrollment. A truthy ``strict``
    # means some run's ``reconcile_meta_tags`` successfully wrote this context's
    # rubric (this run, or a PRIOR run over the same source). Only the
    # structure-only ``materialize_structure`` orphan (no ``strict``) is dead
    # weight, so only that is pruned.
    if matched.get("strict"):
        return False

    if kept:
        block["contexts"] = kept
    else:
        block.pop("contexts", None)

    # Re-point (or drop) the flat structure mirror only if it pointed at the
    # pruned spine. The mirror is last-write-wins, so fall back to the
    # most-recent REMAINING structure-bearing context. Flat ``tags`` are NOT
    # touched -- the union is intentionally monotonic.
    flat_struct = block.get("structure")
    if (isinstance(flat_struct, dict)
            and str(flat_struct.get("synthesis_graph")) == str(synthesis_graph)):
        replacement: dict | None = None
        for entry in kept:
            if isinstance(entry, dict) and isinstance(entry.get("structure"), dict):
                replacement = entry["structure"]
        if replacement is not None:
            block["structure"] = replacement
        else:
            block.pop("structure", None)

    meta["extraction"] = block
    write_yaml_atomic(meta_path, meta)
    return True

resolve_extraction_context

resolve_extraction_context(source: str, project: str | None = None, synthesis_graph: str | None = None, graphs_dir: Path | None = None) -> dict | None

Resolve the extraction record governing a write in a given context.

Returns the (project, synthesis_graph) context entry when one matches. The returned dict carries the leaf fields (schema/strict/grounded/ tags/structure) callers read.

The resolution rule -- which reconciles the no-silent-loss requirement with back-compat -- keys on whether the source carries a NON-EMPTY contexts list and whether a FULL context key was demanded. A key is "full" only when BOTH project AND synthesis_graph are supplied; a PARTIAL pair (exactly one of the two) can never match a keyed entry, so it degrades to the flat block exactly like a no-context read rather than failing loud (failing loud on a partial buys nothing -- it would only break the legitimate no-structure call that accidentally carries one half).

• Non-empty contexts list + FULL key whose (project, synthesis_graph) IS present -> return that entry.
• Non-empty contexts list + FULL key that is ABSENT -> RAISE :class:ExtractionContextNotFound. Falling back to the flat block here would silently enforce the WRONG rubric (the multi-context P1 we must not regress).
• PARTIAL key (exactly one half supplied) -> FALL BACK to the flat block (it can never match a keyed entry; treated as a no-context read), even against a populated contexts list. This is what lets a no-structure run that passes a stray project over a multi-context source still enforce flat.
• NO contexts list (empty or absent) -> pure-legacy / flat-only source (or a best-effort keyed write that failed): FALL BACK to the flat block, even under a full key. This degrades to flat enforcement exactly like pre-keyed behaviour rather than hard-rejecting every claim.
• NO context supplied (legacy no-context read) -> fall back to the flat leaves so pre-keyed meta and no-context callers behave exactly as before.

Returns None when the source has no extraction meta at all (or only an empty block) -- including under a full key, so a graph with no extraction rubric (e.g. _cross) correctly SKIPS enforcement rather than failing loud.

Source code in zettelkasten/graph_io.py

def resolve_extraction_context(
    source: str,
    project: str | None = None,
    synthesis_graph: str | None = None,
    graphs_dir: Path | None = None,
) -> dict | None:
    """Resolve the extraction record governing a write in a given context.

    Returns the ``(project, synthesis_graph)`` context entry when one matches. The
    returned dict carries the leaf fields (``schema``/``strict``/``grounded``/
    ``tags``/``structure``) callers read.

    The resolution rule -- which reconciles the no-silent-loss requirement with
    back-compat -- keys on whether the source carries a NON-EMPTY ``contexts``
    list and whether a FULL context key was demanded. A key is "full" only when
    BOTH ``project`` AND ``synthesis_graph`` are supplied; a PARTIAL pair (exactly
    one of the two) can never match a keyed entry, so it degrades to the flat
    block exactly like a no-context read rather than failing loud (failing loud on
    a partial buys nothing -- it would only break the legitimate no-structure call
    that accidentally carries one half).

    * Non-empty ``contexts`` list + FULL key whose ``(project, synthesis_graph)``
      IS present -> return that entry.
    * Non-empty ``contexts`` list + FULL key that is ABSENT -> RAISE
      :class:`ExtractionContextNotFound`. Falling back to the flat block here would
      silently enforce the WRONG rubric (the multi-context P1 we must not regress).
    * PARTIAL key (exactly one half supplied) -> FALL BACK to the flat block (it
      can never match a keyed entry; treated as a no-context read), even against a
      populated ``contexts`` list. This is what lets a no-structure run that
      passes a stray ``project`` over a multi-context source still enforce flat.
    * NO ``contexts`` list (empty or absent) -> pure-legacy / flat-only source (or
      a best-effort keyed write that failed): FALL BACK to the flat block, even
      under a full key. This degrades to flat enforcement exactly like pre-keyed
      behaviour rather than hard-rejecting every claim.
    * NO context supplied (legacy no-context read) -> fall back to the flat leaves
      so pre-keyed meta and no-context callers behave exactly as before.

    Returns ``None`` when the source has no extraction meta at all (or only an
    empty block) -- including under a full key, so a graph with no extraction
    rubric (e.g. ``_cross``) correctly SKIPS enforcement rather than failing loud.
    """
    meta = load_source_meta(source, graphs_dir=graphs_dir)
    block = meta.get("extraction") if isinstance(meta, dict) else None
    if not isinstance(block, dict) or not block:
        return None
    contexts = block.get("contexts")
    has_contexts = isinstance(contexts, list) and len(contexts) > 0
    # Only a FULL key (BOTH halves) demands a keyed entry. A PARTIAL pair can
    # never match a keyed entry, so it falls through to the flat fallback below.
    full_key = bool(project) and bool(synthesis_graph)
    if full_key and has_contexts:
        for entry in contexts:
            if (isinstance(entry, dict)
                    and str(entry.get("project")) == str(project)
                    and str(entry.get("synthesis_graph")) == str(synthesis_graph)):
                return entry
        # FULL key against a populated contexts list, no match: fail loud (a
        # genuine mis-route / unprepared context). Do NOT degrade to the flat
        # block, which may hold another context's rubric (no-silent-loss). Surface
        # the keys this source DOES carry so a mis-keyed caller can self-correct.
        available = [
            {"project": entry.get("project"),
             "synthesis_graph": entry.get("synthesis_graph")}
            for entry in contexts
            if isinstance(entry, dict)
        ]
        raise ExtractionContextNotFound(
            source, str(project), str(synthesis_graph), available=available,
        )
    # TRACEABILITY (no-silent-routing): a source that DOES carry keyed contexts is
    # being resolved WITHOUT a full key (no context, or a partial pair), so it
    # degrades to the flat block -- which holds the LAST structure-bearing keyed
    # write (last-write-wins). For the trio this never happens (scribe/auditor pass
    # both keys or neither against a single-context source), so a degrade here means
    # a NON-trio / manual / future caller is routing a multi-context source via the
    # arbitrary flat fallback. The resolution itself is correct (the north-star
    # no-context path enforces against the flat/semantic default), but the silent
    # last-write-wins pick should be greppable rather than invisible. WARNING-level
    # because it is rare by construction; if it ever fires hot, a caller is failing
    # to thread its (project, synthesis_graph) context.
    if has_contexts:
        logger.warning(
            "resolve_extraction_context: source '%s' carries %d keyed extraction "
            "context(s) but was resolved with %s (project=%r, synthesis_graph=%r); "
            "degrading to the flat block (last-write-wins). Thread the run's "
            "(project, synthesis_graph) to route deterministically.",
            source,
            len(contexts),
            "a partial key" if (bool(project) or bool(synthesis_graph)) else "no context",
            project,
            synthesis_graph,
        )
    # Fall back to the flat leaves (drop the bookkeeping ``contexts`` list). This
    # covers the legacy no-context read, a PARTIAL key (one half supplied), AND a
    # full key against a source with NO contexts list (pure-legacy / flat-only /
    # best-effort-failed keyed write) -- a graceful degrade to flat enforcement,
    # never a hard reject.
    flat = {k: v for k, v in block.items() if k != "contexts"}
    return flat or None

mark_slices_extracted

mark_slices_extracted(source: str, slices, *, project: str | None = None, synthesis_graph: str | None = None, schema: str = '', content_hash: str = '', run_id: str = '', graphs_dir: Path | None = None) -> dict

Record that slices of source have been extracted, on its _meta.yaml.

Appends the given spans to a completed_slices list inside the source's extraction block — the per-slice incremental-extraction ledger the efficiency pipeline reads (via :func:read_completed_slices) to skip already-mined regions. Each stored entry is a dict {char_start, char_end, page_start, page_end, label, kind, schema, content_hash, at, run_id} (at = ISO-8601 UTC timestamp).

Keying mirrors :func:write_extraction_meta: when BOTH project AND synthesis_graph are given the ledger is written onto the matching keyed (project, synthesis_graph) contexts entry (created empty if absent); otherwise it is written onto the legacy flat block. A partial pair (only one half) degrades to the flat block, exactly like a no-context write.

schema is stored on each row so the ledger is SCHEMA-AWARE: stamping a source done under schema A must not make a later schema-B run skip it. It is ALSO part of the span identity (below), so the same span extracted under two schemas keeps two independent rows rather than one clobbering the other (which matters when both land on the flat block). A blank schema reproduces the legacy schema-less row exactly. kind (e.g. "full") is persisted from the span so a whole-document slice round-trips even if the stamping agent drops its char_start=0 (see :func:coordinator.extraction._span_bounds).

Dedup is on SPAN IDENTITY (see :func:_slice_span_identity): re-marking a span already in the ledger REFRESHES that row's provenance (content_hash / at / run_id) in place rather than appending a duplicate — so a region re-extracted under a changed content_hash updates to the new hash instead of leaving a stale row that :func:read_completed_slices would filter out. Insertion order is preserved.

slices entries that are not dicts are ignored. Returns the record written (the flat block, or the keyed context entry). Raises FileNotFoundError when the source has no _meta.yaml; the source name is path-jailed via :func:safe_join, mirroring :func:write_extraction_meta.

Source code in zettelkasten/graph_io.py

def mark_slices_extracted(
    source: str,
    slices,
    *,
    project: str | None = None,
    synthesis_graph: str | None = None,
    schema: str = "",
    content_hash: str = "",
    run_id: str = "",
    graphs_dir: Path | None = None,
) -> dict:
    """Record that ``slices`` of ``source`` have been extracted, on its ``_meta.yaml``.

    Appends the given spans to a ``completed_slices`` list inside the source's
    ``extraction`` block — the per-slice incremental-extraction ledger the
    efficiency pipeline reads (via :func:`read_completed_slices`) to skip
    already-mined regions. Each stored entry is a dict
    ``{char_start, char_end, page_start, page_end, label, kind, schema,
    content_hash, at, run_id}`` (``at`` = ISO-8601 UTC timestamp).

    Keying mirrors :func:`write_extraction_meta`: when BOTH ``project`` AND
    ``synthesis_graph`` are given the ledger is written onto the matching keyed
    ``(project, synthesis_graph)`` ``contexts`` entry (created empty if absent);
    otherwise it is written onto the legacy flat block. A partial pair (only one
    half) degrades to the flat block, exactly like a no-context write.

    ``schema`` is stored on each row so the ledger is SCHEMA-AWARE: stamping a
    source done under schema A must not make a later schema-B run skip it. It is
    ALSO part of the span identity (below), so the same span extracted under two
    schemas keeps two independent rows rather than one clobbering the other (which
    matters when both land on the flat block). A blank ``schema`` reproduces the
    legacy schema-less row exactly. ``kind`` (e.g. ``"full"``) is persisted from
    the span so a whole-document slice round-trips even if the stamping agent drops
    its ``char_start=0`` (see :func:`coordinator.extraction._span_bounds`).

    Dedup is on SPAN IDENTITY (see :func:`_slice_span_identity`): re-marking a
    span already in the ledger REFRESHES that row's provenance
    (``content_hash`` / ``at`` / ``run_id``) in place rather than appending a
    duplicate — so a region re-extracted under a changed ``content_hash`` updates
    to the new hash instead of leaving a stale row that
    :func:`read_completed_slices` would filter out. Insertion order is preserved.

    ``slices`` entries that are not dicts are ignored. Returns the record written
    (the flat block, or the keyed context entry). Raises ``FileNotFoundError``
    when the source has no ``_meta.yaml``; the source name is path-jailed via
    :func:`safe_join`, mirroring :func:`write_extraction_meta`.
    """
    base = graphs_dir or _graphs_dir()
    meta_path = safe_join(base / source, "_meta.yaml")
    if not meta_path.exists():
        raise FileNotFoundError(f"Source '{source}' not found.")
    meta = yaml.safe_load(meta_path.read_text(encoding="utf-8")) or {}
    block = meta.get("extraction")
    if not isinstance(block, dict):
        block = {}

    record = _extraction_ledger_record(
        block, project=project, synthesis_graph=synthesis_graph, create=True,
    )

    existing = record.get("completed_slices")
    if not isinstance(existing, list):
        existing = []
    # Index current rows by span identity so a re-mark refreshes provenance in
    # place; ``ordered`` preserves first-seen order for a stable on-disk ledger.
    by_identity: dict = {}
    ordered: list = []
    for entry in existing:
        if not isinstance(entry, dict):
            continue
        ident = _slice_span_identity(entry)
        if ident not in by_identity:
            ordered.append(ident)
        by_identity[ident] = entry

    at = _now_iso()
    for span in slices or []:
        if not isinstance(span, dict):
            continue
        row = {
            "char_start": span.get("char_start"),
            "char_end": span.get("char_end"),
            "page_start": span.get("page_start"),
            "page_end": span.get("page_end"),
            "label": span.get("label", ""),
            # ``kind`` (e.g. "full") lets a whole-document slice round-trip even
            # when char_start=0 is dropped by the stamping agent; stored only when
            # present so a plain char/page row stays byte-identical to legacy.
            **({"kind": span.get("kind")} if span.get("kind") else {}),
            # ``schema`` isolates one schema's ledger from another's (and blank
            # reproduces the legacy schema-less row).
            "schema": schema,
            "content_hash": content_hash,
            "at": at,
            "run_id": run_id,
        }
        ident = _slice_span_identity(row)
        if ident not in by_identity:
            ordered.append(ident)
        by_identity[ident] = row

    record["completed_slices"] = [by_identity[ident] for ident in ordered]
    meta["extraction"] = block
    write_yaml_atomic(meta_path, meta)
    return record

read_completed_slices

read_completed_slices(source: str, *, project: str | None = None, synthesis_graph: str | None = None, schema: str = '', content_hash: str = '', graphs_dir: Path | None = None) -> list[dict]

Return the completed-slice ledger for (source, context, schema).

The read counterpart of :func:mark_slices_extracted, keyed identically: a FULL (project, synthesis_graph) pair reads the matching keyed context entry; no context (or a partial pair) reads the flat block.

schema makes the read SCHEMA-AWARE so one schema never consumes another's completed slices. The match rule preserves back-compat with legacy (schema-less) ledgers:

• A blank schema (caller doesn't care) returns every row regardless of its stored schema — identical to the pre-schema behaviour.
• A non-blank schema returns rows whose stored schema equals it, PLUS rows with NO stored schema (legacy rows written before schema-awareness are shared by every schema so existing ledgers keep skipping). A row stamped with a DIFFERENT non-blank schema is filtered out — that is the isolation fix.

When content_hash is non-empty, ONLY entries whose stored content_hash matches are returned — a changed source hash invalidates stale ledger rows, so a re-hashed source is correctly seen as not-yet-extracted. An empty content_hash returns every recorded span regardless of hash.

Returns [] for a missing _meta.yaml, a source with no extraction block, a keyed context with no matching entry, or an empty/absent ledger.

Source code in zettelkasten/graph_io.py

def read_completed_slices(
    source: str,
    *,
    project: str | None = None,
    synthesis_graph: str | None = None,
    schema: str = "",
    content_hash: str = "",
    graphs_dir: Path | None = None,
) -> list[dict]:
    """Return the completed-slice ledger for ``(source, context, schema)``.

    The read counterpart of :func:`mark_slices_extracted`, keyed identically: a
    FULL ``(project, synthesis_graph)`` pair reads the matching keyed context
    entry; no context (or a partial pair) reads the flat block.

    ``schema`` makes the read SCHEMA-AWARE so one schema never consumes another's
    completed slices. The match rule preserves back-compat with legacy
    (schema-less) ledgers:

    * A blank ``schema`` (caller doesn't care) returns every row regardless of its
      stored schema — identical to the pre-schema behaviour.
    * A non-blank ``schema`` returns rows whose stored ``schema`` equals it, PLUS
      rows with NO stored schema (legacy rows written before schema-awareness are
      shared by every schema so existing ledgers keep skipping). A row stamped with
      a DIFFERENT non-blank schema is filtered out — that is the isolation fix.

    When ``content_hash`` is non-empty, ONLY entries whose stored ``content_hash``
    matches are returned — a changed source hash invalidates stale ledger rows, so
    a re-hashed source is correctly seen as not-yet-extracted. An empty
    ``content_hash`` returns every recorded span regardless of hash.

    Returns ``[]`` for a missing ``_meta.yaml``, a source with no ``extraction``
    block, a keyed context with no matching entry, or an empty/absent ledger.
    """
    meta = load_source_meta(source, graphs_dir=graphs_dir)
    block = meta.get("extraction") if isinstance(meta, dict) else None
    if not isinstance(block, dict):
        return []
    record = _extraction_ledger_record(
        block, project=project, synthesis_graph=synthesis_graph, create=False,
    )
    if not isinstance(record, dict):
        return []
    slices = record.get("completed_slices")
    if not isinstance(slices, list):
        return []
    rows = [s for s in slices if isinstance(s, dict)]
    if schema:
        want = str(schema)
        # A row matches when its schema equals ``want`` OR it has no stored schema
        # (legacy/shared row). A different non-blank schema is excluded.
        rows = [
            s for s in rows
            if str(s.get("schema", "")) in ("", want)
        ]
    if content_hash:
        rows = [s for s in rows if str(s.get("content_hash", "")) == str(content_hash)]
    return rows

write_source_doi

write_source_doi(source: str, doi: str, *, graphs_dir: Path | None = None) -> str

Set a promoted source's bibliographic doi in its _meta.yaml.

The single DOI-write seam used by the dashboard's source-doi endpoint. The value is normalized to bare 10.xxxx/... form (URL/doi: prefixes stripped) via :func:normalize_doi; an empty/whitespace input clears the field. Returns the stored (normalized) value.

Raises FileNotFoundError for a source with no _meta.yaml (i.e. not a real source folder). The source name is path-jailed via :func:safe_join, mirroring :func:write_coverage.

Source code in zettelkasten/graph_io.py

def write_source_doi(
    source: str,
    doi: str,
    *,
    graphs_dir: Path | None = None,
) -> str:
    """Set a promoted source's bibliographic ``doi`` in its ``_meta.yaml``.

    The single DOI-write seam used by the dashboard's source-doi endpoint. The
    value is normalized to bare ``10.xxxx/...`` form (URL/``doi:`` prefixes
    stripped) via :func:`normalize_doi`; an empty/whitespace input clears the
    field. Returns the stored (normalized) value.

    Raises ``FileNotFoundError`` for a source with no ``_meta.yaml`` (i.e. not a
    real source folder). The source name is path-jailed via :func:`safe_join`,
    mirroring :func:`write_coverage`.
    """
    base = graphs_dir or _graphs_dir()
    meta_path = safe_join(base / source, "_meta.yaml")
    if not meta_path.exists():
        raise FileNotFoundError(f"Source '{source}' not found.")
    meta = yaml.safe_load(meta_path.read_text(encoding="utf-8")) or {}
    meta["doi"] = normalize_doi(doi)
    write_yaml_atomic(meta_path, meta)
    return meta["doi"]

load_citations

load_citations(graphs_dir: Path | None = None) -> dict[str, dict]

Load all citation entities from _citations/ as a dict keyed by ID.

Citation files are YAML (not markdown notes). Each file represents a lightweight bibliographic entity that notes can link to via graph="_citations".

Source code in zettelkasten/graph_io.py

def load_citations(graphs_dir: Path | None = None) -> dict[str, dict]:
    """Load all citation entities from _citations/ as a dict keyed by ID.

    Citation files are YAML (not markdown notes). Each file represents a
    lightweight bibliographic entity that notes can link to via
    graph="_citations".
    """
    base = graphs_dir or _graphs_dir()
    citations_dir = base / "_citations"
    if not citations_dir.exists():
        return {}

    citations: dict[str, dict] = {}
    # Sort the glob so which member survives a same-key dedup is stable across
    # runs and filesystems (glob order is otherwise undefined).
    for filepath in sorted(citations_dir.glob("*.yaml")):
        # A single poison file (invalid YAML, tab indentation, non-UTF-8 bytes,
        # multi-document stream, unreadable inode) must not 500 every caller of
        # this store. Skip the offender and keep loading the rest; its id simply
        # won't resolve.
        try:
            data = yaml.safe_load(filepath.read_text(encoding="utf-8"))
        except (yaml.YAMLError, UnicodeDecodeError, OSError) as exc:
            logger.warning("Skipping unreadable citation file %s: %s", filepath, exc)
            continue
        if data and isinstance(data, dict):
            cid = data.get("id", filepath.stem)
            citations[cid] = data
    return citations

normalize_doi

normalize_doi(doi: str) -> str

Normalize a DOI to a bare lowercase 10.xxxx/... form.

Strips common URL/scheme prefixes so the same DOI written different ways (https://doi.org/10.x, doi:10.X) collapses to one key. Returns "" for empty/invalid input.

Source code in zettelkasten/graph_io.py

def normalize_doi(doi: str) -> str:
    """Normalize a DOI to a bare lowercase ``10.xxxx/...`` form.

    Strips common URL/scheme prefixes so the same DOI written different ways
    (``https://doi.org/10.x``, ``doi:10.X``) collapses to one key. Returns ``""``
    for empty/invalid input.
    """
    if not doi:
        return ""
    d = doi.strip().lower()
    for prefix in ("https://doi.org/", "http://doi.org/", "http://dx.doi.org/",
                   "https://dx.doi.org/", "doi:"):
        if d.startswith(prefix):
            d = d[len(prefix):]
            break
    return d.strip()

canonical_citation_key

canonical_citation_key(citation: dict) -> str

Compute a dedup fingerprint for a citation/source-like dict.

Priority: DOI > arXiv id > normalized (first-author-lastname + year + leading title words). Two records with the same fingerprint are considered the same work. The fingerprint is not the storage id — it is only used to detect duplicates.

Source code in zettelkasten/graph_io.py

def canonical_citation_key(citation: dict) -> str:
    """Compute a dedup fingerprint for a citation/source-like dict.

    Priority: DOI > arXiv id > normalized (first-author-lastname + year +
    leading title words). Two records with the same fingerprint are considered
    the same work. The fingerprint is *not* the storage id — it is only used to
    detect duplicates.
    """
    doi = normalize_doi(citation.get("doi", ""))
    if doi:
        # arXiv DOIs (10.48550/arxiv.xxxx) collapse to the arxiv id form.
        if "arxiv" in doi:
            tail = doi.split("arxiv.")[-1] if "arxiv." in doi else doi
            return f"arxiv:{tail}"
        return f"doi:{doi}"

    arxiv = str(citation.get("arxiv_id", "")).strip().lower()
    if arxiv:
        return f"arxiv:{arxiv}"

    authors = citation.get("authors") or []
    first_author = ""
    if authors:
        first = str(authors[0])
        first_author = re.sub(r"[^a-z]", "", first.split()[-1].lower()) if first.split() else ""
    year = str(citation.get("year", "")).strip()
    title = str(citation.get("title", "")).lower()
    title_words = re.sub(r"[^a-z0-9 ]", "", title).split()[:6]
    title_slug = "-".join(title_words)
    return f"meta:{first_author}-{year}-{title_slug}".strip("-")

note_graph_names

note_graph_names(graphs_dir: Path | None = None) -> list[str]

List note-graph folder names: every source folder plus _cross.

Excludes the non-note special folders (_citations, _projects, _reviews). Used by maintenance tools that must scan/rewrite links across all graphs.

Source code in zettelkasten/graph_io.py

def note_graph_names(graphs_dir: Path | None = None) -> list[str]:
    """List note-graph folder names: every source folder plus ``_cross``.

    Excludes the non-note special folders (``_citations``, ``_projects``,
    ``_reviews``). Used by maintenance tools that must scan/rewrite links across
    all graphs.
    """
    base = graphs_dir or _graphs_dir()
    if not base.exists():
        return []
    names: list[str] = []
    for d in sorted(base.iterdir()):
        if not d.is_dir():
            continue
        if d.name.startswith("_") and d.name != "_cross":
            continue
        names.append(d.name)
    return names

load_project

load_project(project_name: str, graphs_dir: Path | None = None) -> dict

Load a single project manifest from _projects/.yaml.

Returns {} for a missing file (no exception). A corrupt manifest (invalid YAML, non-UTF-8 bytes, unreadable inode) is NOT silently swallowed: unlike the missing case it would mask real data loss, so it is re-raised as a clear, catchable ValueError that the route layer can map to a clean status instead of a raw 500 traceback — mirroring :func:load_review.

Source code in zettelkasten/graph_io.py

def load_project(project_name: str, graphs_dir: Path | None = None) -> dict:
    """Load a single project manifest from _projects/<name>.yaml.

    Returns ``{}`` for a missing file (no exception). A *corrupt* manifest
    (invalid YAML, non-UTF-8 bytes, unreadable inode) is NOT silently swallowed:
    unlike the missing case it would mask real data loss, so it is re-raised as a
    clear, catchable ``ValueError`` that the route layer can map to a clean
    status instead of a raw 500 traceback — mirroring :func:`load_review`.
    """
    base = graphs_dir or _graphs_dir()
    validate_id(project_name, kind="project name", for_filename=True)
    filepath = safe_join(base / "_projects", f"{project_name}.yaml")
    if not filepath.exists():
        return {}
    try:
        data = yaml.safe_load(filepath.read_text(encoding="utf-8"))
    except (yaml.YAMLError, UnicodeDecodeError, OSError) as exc:
        logger.warning("Corrupt project manifest %s: %s", filepath, exc)
        raise ValueError(f"corrupt project manifest: {project_name}") from exc
    return data if isinstance(data, dict) else {}

list_projects

list_projects(graphs_dir: Path | None = None) -> list[dict]

List all project manifests from _projects/.

A single poison file (invalid YAML, non-UTF-8 bytes, unreadable inode) must not 500 the whole list, so the offender is skipped and the rest keep loading — mirroring :func:load_citations.

Source code in zettelkasten/graph_io.py

def list_projects(graphs_dir: Path | None = None) -> list[dict]:
    """List all project manifests from _projects/.

    A single poison file (invalid YAML, non-UTF-8 bytes, unreadable inode) must
    not 500 the whole list, so the offender is skipped and the rest keep
    loading — mirroring :func:`load_citations`.
    """
    base = graphs_dir or _graphs_dir()
    projects_dir = base / "_projects"
    if not projects_dir.exists():
        return []

    results: list[dict] = []
    for filepath in sorted(projects_dir.glob("*.yaml")):
        try:
            data = yaml.safe_load(filepath.read_text(encoding="utf-8"))
        except (yaml.YAMLError, UnicodeDecodeError, OSError) as exc:
            logger.warning("Skipping unreadable project file %s: %s", filepath, exc)
            continue
        if data and isinstance(data, dict):
            results.append(data)
    return results

project_cross_ids

project_cross_ids(project_data: dict) -> list[str]

Explicit _cross note ids a project claims (the manifest cross: list).

Membership of a synthesis note in a project is now EXPLICIT (curated on the manifest) rather than inferred from the link graph. Returns a de-duplicated, order-preserving list of non-empty string ids; tolerates a missing/malformed field by returning [] so a hand-edited manifest never 500s a read.

Source code in zettelkasten/graph_io.py

def project_cross_ids(project_data: dict) -> list[str]:
    """Explicit ``_cross`` note ids a project claims (the manifest ``cross:`` list).

    Membership of a synthesis note in a project is now EXPLICIT (curated on the
    manifest) rather than inferred from the link graph. Returns a de-duplicated,
    order-preserving list of non-empty string ids; tolerates a missing/malformed
    field by returning ``[]`` so a hand-edited manifest never 500s a read.
    """
    raw = project_data.get("cross", []) if isinstance(project_data, dict) else []
    if not isinstance(raw, list):
        return []
    seen: set[str] = set()
    out: list[str] = []
    for item in raw:
        cid = str(item).strip()
        if cid and cid not in seen:
            seen.add(cid)
            out.append(cid)
    return out

project_default_spine

project_default_spine(project_data: dict) -> str | None

The org id of a project's default (primary) organizing spine, or None.

The manifest default_spine field names the spine-directory entry (a project-owned org id) the project organizes by — parallel to the cross: whitelist (:func:project_cross_ids), but for organizing STRUCTURES rather than synthesis NOTES. None (unset / missing / malformed) means the implicit intrinsic lens — the organic cluster graph — which has NO org record. A blank or non-string value is treated as unset, so a hand-edited manifest never 500s a read.

Source code in zettelkasten/graph_io.py

def project_default_spine(project_data: dict) -> str | None:
    """The org id of a project's default (primary) organizing spine, or ``None``.

    The manifest ``default_spine`` field names the spine-directory entry (a
    project-owned org id) the project organizes by — parallel to the ``cross:``
    whitelist (:func:`project_cross_ids`), but for organizing STRUCTURES rather
    than synthesis NOTES. ``None`` (unset / missing / malformed) means the implicit
    **intrinsic lens** — the organic cluster graph — which has NO org record. A
    blank or non-string value is treated as unset, so a hand-edited manifest never
    500s a read.
    """
    if not isinstance(project_data, dict):
        return None
    raw = project_data.get("default_spine")
    if not isinstance(raw, str):
        return None
    val = raw.strip()
    return val or None

cross_note_connects

cross_note_connects(note, source_set: set[str], project_note_ids: set[str], referenced_cross: set[str]) -> bool

Whether a _cross note is link-connected to a project's sources.

This is the legacy inference predicate (a cross note is referenced by a project note, or it links into a project source / project note). Explicit cross: membership is now authoritative for INCLUSION; this predicate is retained only to compute the suggested bucket — link-connected cross notes the project has not yet claimed — so both composite builders score suggestions identically.

Source code in zettelkasten/graph_io.py

def cross_note_connects(
    note,
    source_set: set[str],
    project_note_ids: set[str],
    referenced_cross: set[str],
) -> bool:
    """Whether a ``_cross`` note is *link-connected* to a project's sources.

    This is the legacy inference predicate (a cross note is referenced by a
    project note, or it links into a project source / project note). Explicit
    ``cross:`` membership is now authoritative for INCLUSION; this predicate is
    retained only to compute the *suggested* bucket — link-connected cross notes
    the project has not yet claimed — so both composite builders score
    suggestions identically.
    """
    if note.id in referenced_cross:
        return True
    return any(
        (link.graph in source_set) or (link.target in project_note_ids)
        for link in note.links
    )

load_review

load_review(name: str, graphs_dir: Path | None = None) -> dict

Load a single literature-review manifest from _reviews/.yaml.

Returns {} for a missing file (no exception), mirroring :func:load_project. _reviews/ lives beside _projects/ at <repo>/.zettelkasten/.

A corrupt manifest (invalid YAML, non-UTF-8 bytes, unreadable inode) is NOT silently swallowed: unlike the missing case it would mask real data loss, so it is re-raised as a clear, catchable ValueError that the route layer can map to a clean status instead of a raw 500 traceback.

Source code in zettelkasten/graph_io.py

def load_review(name: str, graphs_dir: Path | None = None) -> dict:
    """Load a single literature-review manifest from _reviews/<name>.yaml.

    Returns ``{}`` for a missing file (no exception), mirroring
    :func:`load_project`. ``_reviews/`` lives beside ``_projects/`` at
    ``<repo>/.zettelkasten/``.

    A *corrupt* manifest (invalid YAML, non-UTF-8 bytes, unreadable inode) is
    NOT silently swallowed: unlike the missing case it would mask real data
    loss, so it is re-raised as a clear, catchable ``ValueError`` that the route
    layer can map to a clean status instead of a raw 500 traceback.
    """
    base = graphs_dir or _graphs_dir()
    validate_id(name, kind="review name", for_filename=True)
    filepath = safe_join(base / "_reviews", f"{name}.yaml")
    if not filepath.exists():
        return {}
    try:
        data = yaml.safe_load(filepath.read_text(encoding="utf-8"))
    except (yaml.YAMLError, UnicodeDecodeError, OSError) as exc:
        logger.warning("Corrupt review manifest %s: %s", filepath, exc)
        raise ValueError(f"corrupt review manifest: {name}") from exc
    return data if isinstance(data, dict) else {}

list_reviews

list_reviews(graphs_dir: Path | None = None) -> list[dict]

List all review manifests from _reviews/.

A single poison file (invalid YAML, non-UTF-8 bytes, unreadable inode) must not 500 the whole list, so the offender is skipped and the rest keep loading — mirroring :func:load_citations.

Source code in zettelkasten/graph_io.py

def list_reviews(graphs_dir: Path | None = None) -> list[dict]:
    """List all review manifests from _reviews/.

    A single poison file (invalid YAML, non-UTF-8 bytes, unreadable inode) must
    not 500 the whole list, so the offender is skipped and the rest keep
    loading — mirroring :func:`load_citations`.
    """
    base = graphs_dir or _graphs_dir()
    reviews_dir = base / "_reviews"
    if not reviews_dir.exists():
        return []

    results: list[dict] = []
    for filepath in sorted(reviews_dir.glob("*.yaml")):
        try:
            data = yaml.safe_load(filepath.read_text(encoding="utf-8"))
        except (yaml.YAMLError, UnicodeDecodeError, OSError) as exc:
            logger.warning("Skipping unreadable review file %s: %s", filepath, exc)
            continue
        if data and isinstance(data, dict):
            results.append(data)
    return results