zettelkasten.federation

zettelkasten.federation

Read-only federation helpers for projecting related Zettelkasten repos.

Mirrors memory.federation: another repo's .zettelkasten/ tree can be browsed alongside this one as a read-only overlay in the dashboard. Repos are configured via a federated_repos list in .zettelkasten/config.yaml.

Federated graphs are surfaced with namespaced names (<repo_id>:<graph>) so they never collide with local graphs. The dashboard routes resolve a namespaced name back to its repo and build a read-only :class:~zettelkasten.graph.ZettelGraph pointed at that repo's directory; because federated repos commit their .md notes but gitignore their .angelo/zettel/*.kgl embedding caches, semantic search/cluster for a federated graph is served by rebuilding its index locally from the committed notes (see ZettelGraph._embedding_cache_path).

Everything here is best-effort and read-only: a misconfigured repo yields a non-fatal validation error rather than raising.

FederatedRepo dataclass

Validated configuration for one related repository.

graphs is an optional source-graph allowlist that FAILS CLOSED:

• None means "expose ALL graphs" — the default, set only when the config OMITS graphs: or gives it an explicit YAML null. This is the fully backward-compatible behavior.
• a non-empty frozenset exposes exactly those named source graphs (plus the always-shared _cross synthesis graph).
• an EMPTY frozenset() exposes NOTHING but _cross — this is what a present-but-empty, blank-only, or malformed graphs: value coerces to, so a misconfiguration hides everything rather than silently exposing all.

Source code in zettelkasten/federation.py

@dataclass(frozen=True)
class FederatedRepo:
    """Validated configuration for one related repository.

    ``graphs`` is an optional source-graph allowlist that FAILS CLOSED:

    - ``None`` means "expose ALL graphs" — the default, set only when the config
      OMITS ``graphs:`` or gives it an explicit YAML null. This is the fully
      backward-compatible behavior.
    - a non-empty ``frozenset`` exposes exactly those named source graphs (plus
      the always-shared ``_cross`` synthesis graph).
    - an EMPTY ``frozenset()`` exposes NOTHING but ``_cross`` — this is what a
      present-but-empty, blank-only, or malformed ``graphs:`` value coerces to, so
      a misconfiguration hides everything rather than silently exposing all.
    """

    id: str
    name: str
    path: Path
    zettel_dir: Path
    graphs: frozenset[str] | None = None

federation_cache_root

federation_cache_root(root: Path | None = None) -> Path

Directory holding decrypted bundle caches (one subdir per synced peer).

Anchored at <root>/.angelo/federation-cache where root defaults to the workspace root — the same anchor :func:configured_repos uses for relative repo paths. Both the sync writer (CLI) and this reader resolve the cache to the same place; the .angelo root is shared with the memory federation cache so a peer can publish both a .memory and a .zettelkasten cache under one <peer_id>/ dir.

Source code in zettelkasten/federation.py

def federation_cache_root(root: Path | None = None) -> Path:
    """Directory holding decrypted bundle caches (one subdir per synced peer).

    Anchored at ``<root>/.angelo/federation-cache`` where ``root`` defaults to
    the workspace root — the same anchor :func:`configured_repos` uses for
    relative repo paths. Both the ``sync`` writer (CLI) and this reader resolve
    the cache to the same place; the ``.angelo`` root is shared with the memory
    federation cache so a peer can publish both a ``.memory`` and a
    ``.zettelkasten`` cache under one ``<peer_id>/`` dir.
    """
    base = _workspace_root() if root is None else root
    return base / ".angelo" / FEDERATION_CACHE_DIRNAME

graph_allowed

graph_allowed(repo: FederatedRepo, local_name: str) -> bool

Whether a federated repo exposes local_name under its allowlist.

Returns True when the repo has no allowlist (repo.graphs is None → all graphs exposed), when local_name is the shared _cross synthesis graph (ALWAYS exposed — it is not a source, and projects implicitly append it), or when local_name is explicitly listed in the allowlist.

Source code in zettelkasten/federation.py

def graph_allowed(repo: FederatedRepo, local_name: str) -> bool:
    """Whether a federated repo exposes ``local_name`` under its allowlist.

    Returns True when the repo has no allowlist (``repo.graphs is None`` → all
    graphs exposed), when ``local_name`` is the shared ``_cross`` synthesis graph
    (ALWAYS exposed — it is not a source, and projects implicitly append it), or
    when ``local_name`` is explicitly listed in the allowlist.
    """
    if repo.graphs is None:
        return True
    if local_name == "_cross":
        return True
    return local_name in repo.graphs

federated_review_allowed

federated_review_allowed(repo: FederatedRepo, manifest: dict[str, Any], *, load_project: Callable[[str], dict[str, Any]] | None = None) -> bool

Whether a federated repo exposes a review under its allowlist (subset rule).

Single source of truth for review visibility so the three seams (:func:federated_reviews list overlay, _ensure_federated_review_allowed, and _load_review_or_error in the dashboard routes) cannot diverge.

• No allowlist (repo.graphs is None) → always True (unchanged behavior).
• Non-empty manifest['graph'] → gate on that graph exactly as before (graph_allowed(repo, graph)).
• Empty/missing graph (a PROJECT-SCOPED review) → resolve the review's project manifest and require ALL of its sources allowed, mirroring :func:federated_projects. A review with neither a graph nor a resolvable project, or a project with no sources, is an edge case: hide it under an allowlist (conservative but consistent).

load_project maps a project name to its manifest dict; it defaults to reading the peer's .zettelkasten/_projects/ via zettelkasten.graph.load_project. Any resolution failure is treated as "not resolvable" → hidden.

Source code in zettelkasten/federation.py

def federated_review_allowed(
    repo: FederatedRepo,
    manifest: dict[str, Any],
    *,
    load_project: Callable[[str], dict[str, Any]] | None = None,
) -> bool:
    """Whether a federated repo exposes a review under its allowlist (subset rule).

    Single source of truth for review visibility so the three seams
    (:func:`federated_reviews` list overlay, ``_ensure_federated_review_allowed``,
    and ``_load_review_or_error`` in the dashboard routes) cannot diverge.

    - No allowlist (``repo.graphs is None``) → always ``True`` (unchanged behavior).
    - Non-empty ``manifest['graph']`` → gate on that graph exactly as before
      (``graph_allowed(repo, graph)``).
    - Empty/missing ``graph`` (a PROJECT-SCOPED review) → resolve the review's
      ``project`` manifest and require ALL of its ``sources`` allowed, mirroring
      :func:`federated_projects`. A review with neither a graph nor a resolvable
      project, or a project with no sources, is an edge case: hide it under an
      allowlist (conservative but consistent).

    ``load_project`` maps a project name to its manifest dict; it defaults to
    reading the peer's ``.zettelkasten/_projects/`` via
    ``zettelkasten.graph.load_project``. Any resolution failure is treated as
    "not resolvable" → hidden.
    """
    if repo.graphs is None:
        return True
    graph = str(manifest.get("graph") or "")
    if graph:
        return graph_allowed(repo, graph)

    project_name = str(manifest.get("project") or "")
    if not project_name:
        return False

    if load_project is None:
        from zettelkasten.graph import load_project as _load_project

        def load_project(name: str) -> dict[str, Any]:
            return _load_project(name, graphs_dir=repo.zettel_dir)

    try:
        project = load_project(project_name)
    except Exception:
        return False
    if not isinstance(project, dict):
        return False
    raw_sources = project.get("sources", [])
    sources = raw_sources if isinstance(raw_sources, list) else []
    if not sources:
        return False
    return all(graph_allowed(repo, str(src)) for src in sources)

federated_project_allowed

federated_project_allowed(repo: FederatedRepo, project: dict[str, Any]) -> bool

Whether a federated repo exposes a project under its allowlist (subset rule).

Single source of truth for project visibility so the two seams (:func:federated_projects list overlay and _load_project_or_404 in the dashboard routes) cannot diverge. Mirrors :func:federated_review_allowed:

• No allowlist (repo.graphs is None) → always True (unchanged behavior, INCLUDING a source-less project which stays visible/resolvable).
• Under an allowlist, a project with EMPTY or MISSING sources is hidden: all([]) is vacuously True, so without this guard a source-less peer project would leak even under a graphs: [] lockdown.
• Otherwise, require EVERY source graph allowed (_cross is always allowed), so a visible project never dangles on a hidden source.

Source code in zettelkasten/federation.py

def federated_project_allowed(repo: FederatedRepo, project: dict[str, Any]) -> bool:
    """Whether a federated repo exposes a project under its allowlist (subset rule).

    Single source of truth for project visibility so the two seams
    (:func:`federated_projects` list overlay and ``_load_project_or_404`` in the
    dashboard routes) cannot diverge. Mirrors :func:`federated_review_allowed`:

    - No allowlist (``repo.graphs is None``) → always ``True`` (unchanged
      behavior, INCLUDING a source-less project which stays visible/resolvable).
    - Under an allowlist, a project with EMPTY or MISSING ``sources`` is hidden:
      ``all([])`` is vacuously ``True``, so without this guard a source-less peer
      project would leak even under a ``graphs: []`` lockdown.
    - Otherwise, require EVERY source graph allowed (``_cross`` is always allowed),
      so a visible project never dangles on a hidden source.
    """
    if repo.graphs is None:
        return True
    raw_sources = project.get("sources", [])
    sources = raw_sources if isinstance(raw_sources, list) else []
    if not sources:
        return False
    return all(graph_allowed(repo, str(src)) for src in sources)

split_namespace

split_namespace(node_id: str) -> tuple[str, str] | None

Split <repo_id>:<local_id> into its parts, or None if not namespaced.

Source code in zettelkasten/federation.py

def split_namespace(node_id: str) -> tuple[str, str] | None:
    """Split ``<repo_id>:<local_id>`` into its parts, or ``None`` if not namespaced."""
    if NS_SEP not in node_id:
        return None
    repo_id, local_id = node_id.split(NS_SEP, 1)
    if not repo_id or not local_id:
        return None
    return repo_id, local_id

configured_repos

configured_repos(config: dict[str, Any] | None = None, root: Path | None = None) -> tuple[list[FederatedRepo], list[dict[str, str]]]

Return enabled, valid federated repos plus non-fatal validation errors.

Source code in zettelkasten/federation.py

def configured_repos(
    config: dict[str, Any] | None = None, root: Path | None = None
) -> tuple[list[FederatedRepo], list[dict[str, str]]]:
    """Return enabled, valid federated repos plus non-fatal validation errors."""
    if config is None:
        from zettelkasten import config as zk_config

        config = zk_config.read_config()
    root = _workspace_root() if root is None else root

    raw_repos = config.get(CONFIG_KEY, [])
    if raw_repos in (None, ""):
        return [], []
    if not isinstance(raw_repos, list):
        return [], [_repo_error("", f"{CONFIG_KEY} must be a list")]

    repos: list[FederatedRepo] = []
    errors: list[dict[str, str]] = []
    seen_ids: set[str] = set()

    for index, item in enumerate(raw_repos):
        if not isinstance(item, dict):
            errors.append(_repo_error(str(index), "repo config must be an object"))
            continue
        if item.get("enabled", True) is False:
            continue

        raw_path = str(item.get("path") or "").strip()
        fallback_id = f"repo-{index + 1}"
        repo_id = _coerce_repo_id(item.get("id"), fallback_id)
        if not ID_RE.match(repo_id):
            errors.append(_repo_error(repo_id, "repo id must contain only letters, numbers, dots, underscores, or hyphens"))
            continue
        if repo_id in seen_ids:
            errors.append(_repo_error(repo_id, "duplicate repo id"))
            continue
        if not raw_path:
            errors.append(_repo_error(repo_id, "repo path is required"))
            continue

        repo_path = Path(raw_path).expanduser()
        if not repo_path.is_absolute():
            repo_path = root / repo_path
        try:
            repo_path = repo_path.resolve()
        except OSError:
            errors.append(_repo_error(repo_id, "repo path cannot be resolved", repo_path))
            continue

        zettel_dir = repo_path / ".zettelkasten"
        if not zettel_dir.is_dir():
            errors.append(_repo_error(repo_id, "repo does not contain a .zettelkasten/ directory", repo_path))
            continue

        name = str(item.get("name") or repo_path.name or repo_id)
        graphs, graphs_error = _coerce_graphs(item.get("graphs"))
        if graphs_error is not None:
            errors.append(_repo_error(repo_id, graphs_error))
        repos.append(
            FederatedRepo(
                id=repo_id,
                name=name,
                path=repo_path,
                zettel_dir=zettel_dir,
                graphs=graphs,
            )
        )
        seen_ids.add(repo_id)

    # Additively surface decrypted bundle caches under
    # .angelo/federation-cache/<peer_id>/ as read-only federated repos. Absent
    # cache dir => behaviour is exactly as before. A cache whose id collides with
    # a configured repo id is skipped (the explicit config item wins). A cache is
    # already per-recipient filtered, so it exposes ALL its graphs (graphs=None).
    cache_root = federation_cache_root(root)
    if cache_root.is_dir():
        for peer_dir in sorted(cache_root.iterdir()):
            if not peer_dir.is_dir():
                continue
            fallback_id = f"cache-{peer_dir.name}"
            repo_id = _coerce_repo_id(peer_dir.name, fallback_id)
            if not ID_RE.match(repo_id) or repo_id in seen_ids:
                continue
            try:
                cache_path = peer_dir.resolve()
            except OSError:
                errors.append(_repo_error(repo_id, "cache path cannot be resolved", peer_dir))
                continue
            zettel_dir = cache_path / ".zettelkasten"
            if not zettel_dir.is_dir():
                continue
            repos.append(
                FederatedRepo(
                    id=repo_id,
                    name=peer_dir.name or repo_id,
                    path=cache_path,
                    zettel_dir=zettel_dir,
                    graphs=None,
                )
            )
            seen_ids.add(repo_id)

    return repos, errors

find_repo

find_repo(repo_id: str, config: dict[str, Any] | None = None, root: Path | None = None) -> FederatedRepo | None

Resolve a configured repo by id (validated), or None.

Source code in zettelkasten/federation.py

def find_repo(
    repo_id: str, config: dict[str, Any] | None = None, root: Path | None = None
) -> FederatedRepo | None:
    """Resolve a configured repo by id (validated), or ``None``."""
    repos, _errors = configured_repos(config=config, root=root)
    for repo in repos:
        if repo.id == repo_id:
            return repo
    return None

federated_graph_list

federated_graph_list(repo: FederatedRepo) -> list[dict[str, Any]]

List a federated repo's note-graphs as namespaced, read-only graph metas.

Shape matches the local /graphs rows, plus federation fields so the UI can group and badge them.

Source code in zettelkasten/federation.py

def federated_graph_list(repo: FederatedRepo) -> list[dict[str, Any]]:
    """List a federated repo's note-graphs as namespaced, read-only graph metas.

    Shape matches the local ``/graphs`` rows, plus federation fields so the UI
    can group and badge them.
    """
    from zettelkasten.graph import load_source_meta, note_graph_names

    rows: list[dict[str, Any]] = []
    for name in note_graph_names(graphs_dir=repo.zettel_dir):
        if not graph_allowed(repo, name):
            continue
        meta = load_source_meta(name, graphs_dir=repo.zettel_dir)
        rows.append({
            "name": namespace_id(repo.id, name),
            "local_name": name,
            "description": meta.get("description", ""),
            "source": meta.get("source", ""),
            "note_count": _note_count(repo.zettel_dir / name),
            "repo_id": repo.id,
            "repo_name": repo.name,
            "read_only": True,
        })
    return rows

federated_projects

federated_projects(repo: FederatedRepo) -> list[dict[str, Any]]

List a federated repo's projects as namespaced, read-only manifests.

Source code in zettelkasten/federation.py

def federated_projects(repo: FederatedRepo) -> list[dict[str, Any]]:
    """List a federated repo's projects as namespaced, read-only manifests."""
    from zettelkasten.graph import list_projects

    rows: list[dict[str, Any]] = []
    for project in list_projects(graphs_dir=repo.zettel_dir):
        if not isinstance(project, dict):
            continue
        local = str(project.get("name") or "")
        if not local:
            continue
        # Subset rule: hide a project unless EVERY one of its source graphs is
        # exposed, so a visible project never dangles on a hidden source. Under an
        # allowlist a source-less project is also hidden (see
        # :func:`federated_project_allowed`); with no allowlist every project
        # passes (unchanged behavior).
        if not federated_project_allowed(repo, project):
            continue
        rows.append({
            **project,
            "name": namespace_id(repo.id, local),
            "local_name": local,
            "repo_id": repo.id,
            "repo_name": repo.name,
            "read_only": True,
        })
    return rows

federated_reviews

federated_reviews(repo: FederatedRepo) -> list[dict[str, Any]]

List a federated repo's reviews as namespaced, read-only summaries.

Mirrors :func:federated_projects, but a review manifest carries a heavy editorial overlay block (tiering/exclusions/ordering/renames/scaffolding) that the picker does not need, so this returns only the manifest's scalar summary (name/title/project/graph/question) plus the federation fields. The full reconciled view is served on demand by GET /reviews/{name}. Federated reviews are strictly READ-ONLY — there is no federated write path.

Source code in zettelkasten/federation.py

def federated_reviews(repo: FederatedRepo) -> list[dict[str, Any]]:
    """List a federated repo's reviews as namespaced, read-only summaries.

    Mirrors :func:`federated_projects`, but a review manifest carries a heavy
    editorial ``overlay`` block (tiering/exclusions/ordering/renames/scaffolding)
    that the picker does not need, so this returns only the manifest's scalar
    summary (name/title/project/graph/question) plus the federation fields. The
    full reconciled view is served on demand by ``GET /reviews/{name}``.
    Federated reviews are strictly READ-ONLY — there is no federated write path.
    """
    from zettelkasten.graph import list_reviews

    rows: list[dict[str, Any]] = []
    for review in list_reviews(graphs_dir=repo.zettel_dir):
        if not isinstance(review, dict):
            continue
        local = str(review.get("name") or "")
        if not local:
            continue
        # Subset rule: a graph-scoped review gates on its graph; a
        # project-scoped review (graph == '') gates on ALL its project's
        # sources. See :func:`federated_review_allowed`.
        if not federated_review_allowed(repo, review):
            continue
        rows.append({
            "name": namespace_id(repo.id, local),
            "local_name": local,
            "title": review.get("title", ""),
            "project": review.get("project", ""),
            "graph": review.get("graph", ""),
            "question": review.get("question", ""),
            "repo_id": repo.id,
            "repo_name": repo.name,
            "read_only": True,
        })
    return rows

latest_mtime

latest_mtime(config: dict[str, Any] | None = None, root: Path | None = None) -> float

Newest *.md mtime across configured federated repos (for cache keys).

Source code in zettelkasten/federation.py

def latest_mtime(config: dict[str, Any] | None = None, root: Path | None = None) -> float:
    """Newest ``*.md`` mtime across configured federated repos (for cache keys)."""
    repos, _errors = configured_repos(config=config, root=root)
    latest = 0.0
    for repo in repos:
        for path in repo.zettel_dir.rglob("*.md"):
            try:
                latest = max(latest, path.stat().st_mtime)
            except OSError:
                continue
    return latest