Skip to content

Angelo

config

natashahirt/angelo

`zettelkasten.synapse.config`¶

zettelkasten.synapse.config ¶

Intersection config for synapse.

A repo has exactly one memory tree (.memory/). The intersection config names which local ZK projects are semantically relevant to that memory tree — the scope that bounds every synapse operation (unified retrieval, connection discovery, matrix axes). It is the "scope tag" a consuming agent would otherwise have to spell out on every call.

Config lives at .synapse/config.yaml — committed alongside the notes and memory, mirroring how .zettelkasten/config.yaml and .memory/config.yaml are committed. The typed link overlay is committed too (.synapse/links/); only ephemeral scratch caches live gitignored under .angelo/synapse/.

Schema::

intersects:
  - factors
  - execution

Each item is a ZK project name. A list-of-mappings form is also tolerated so per-project options can be added without breaking existing configs::

intersects:
  - project: factors
  - project: execution
    note: "only the momentum sub-tree matters"

A repo: (or repo_id:) qualifier marks the entry as a FEDERATED scope: the project lives in a peer repository declared under federated_repos in this repo's .zettelkasten/config.yaml. The peer's path and graph allowlist are resolved via :func:zettelkasten.federation.find_repo, and its source graphs are surfaced under namespaced <repo_id>:<graph> names (exactly as the dashboard federation overlay does)::

intersects:
  - project: fin-factors
    repo: fin-notes          # peer id from .zettelkasten/ federated_repos
  - execution                 # local, unchanged

The path is resolved against the workspace root at call time (not import time) so tests and dashboards that repoint the workspace see the change.

IntersectSpec `dataclass` ¶

One resolved intersects entry: a ZK project + optional federated repo.

repo is None is a LOCAL scope (project in this repo's .zettelkasten/); a non-None repo names a peer id declared under federated_repos in .zettelkasten/config.yaml.

Source code in zettelkasten/synapse/config.py

@dataclass(frozen=True)
class IntersectSpec:
    """One resolved ``intersects`` entry: a ZK project + optional federated repo.

    ``repo is None`` is a LOCAL scope (project in this repo's ``.zettelkasten/``);
    a non-None ``repo`` names a peer id declared under ``federated_repos`` in
    ``.zettelkasten/config.yaml``.
    """

    project: str
    repo: str | None = None

    def token(self) -> str:
        """Round-trippable scope token: ``project`` or ``<repo>:<project>``.

        The federated form mirrors :func:`zettelkasten.federation.namespace_id`
        so a token stored in a build manifest re-parses to the same spec.
        """
        return f"{self.repo}:{self.project}" if self.repo else self.project

token ¶

token() -> str

Round-trippable scope token: project or <repo>:<project>.

The federated form mirrors :func:zettelkasten.federation.namespace_id so a token stored in a build manifest re-parses to the same spec.

Source code in zettelkasten/synapse/config.py

def token(self) -> str:
    """Round-trippable scope token: ``project`` or ``<repo>:<project>``.

    The federated form mirrors :func:`zettelkasten.federation.namespace_id`
    so a token stored in a build manifest re-parses to the same spec.
    """
    return f"{self.repo}:{self.project}" if self.repo else self.project

synapse_dir ¶

synapse_dir() -> Path

Absolute path to the committed .synapse/ directory.

Source code in zettelkasten/synapse/config.py

def synapse_dir() -> Path:
    """Absolute path to the committed ``.synapse/`` directory."""
    return _anchor(Path(os.environ.get("SYNAPSE_DIR", ".synapse")))

cache_dir ¶

cache_dir() -> Path

Absolute path to the gitignored derived-cache dir (.angelo/synapse/).

Source code in zettelkasten/synapse/config.py

def cache_dir() -> Path:
    """Absolute path to the gitignored derived-cache dir (``.angelo/synapse/``)."""
    angelo = _anchor(Path(os.environ.get("ANGELO_DIR", ".angelo")))
    return angelo / "synapse"

config_path ¶

config_path() -> Path

Absolute path to .synapse/config.yaml.

Source code in zettelkasten/synapse/config.py

def config_path() -> Path:
    """Absolute path to ``.synapse/config.yaml``."""
    return synapse_dir() / "config.yaml"

max_sources ¶

max_sources(default: int = _DEFAULT_MAX_SOURCES) -> int

Cap on how many in-scope sources a synapse READ fans out to.

Resolution: SYNAPSE_MAX_SOURCES env override > max_sources: key in .synapse/config.yaml > default (:data:_DEFAULT_MAX_SOURCES). A value <= 0 disables pruning (unbounded fan-out, the legacy behavior). A non-integer value is ignored (falls through to the next source) so a stray config never crashes a query.

Source code in zettelkasten/synapse/config.py

def max_sources(default: int = _DEFAULT_MAX_SOURCES) -> int:
    """Cap on how many in-scope sources a synapse READ fans out to.

    Resolution: ``SYNAPSE_MAX_SOURCES`` env override > ``max_sources:`` key in
    ``.synapse/config.yaml`` > ``default`` (:data:`_DEFAULT_MAX_SOURCES`). A value
    ``<= 0`` disables pruning (unbounded fan-out, the legacy behavior). A
    non-integer value is ignored (falls through to the next source) so a stray
    config never crashes a query.
    """
    env = os.environ.get("SYNAPSE_MAX_SOURCES")
    if env is not None and env.strip():
        try:
            return int(env)
        except ValueError:
            logger.warning("Ignoring non-integer SYNAPSE_MAX_SOURCES=%r", env)
    raw = read_config().get("max_sources")
    if raw is not None:
        try:
            return int(raw)
        except (TypeError, ValueError):
            logger.warning("Ignoring non-integer synapse config max_sources=%r", raw)
    return default

read_config ¶

read_config() -> dict[str, Any]

Read the synapse config, returning {} when missing or malformed.

Source code in zettelkasten/synapse/config.py

def read_config() -> dict[str, Any]:
    """Read the synapse config, returning ``{}`` when missing or malformed."""
    path = config_path()
    if not path.exists():
        return {}
    try:
        data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError) as exc:
        logger.warning("Could not read synapse config %s: %s", path, exc)
        return {}
    return data if isinstance(data, dict) else {}

write_config ¶

write_config(config: dict[str, Any]) -> Path

Write the synapse config atomically (temp file + rename).

Source code in zettelkasten/synapse/config.py

def write_config(config: dict[str, Any]) -> Path:
    """Write the synapse config atomically (temp file + rename)."""
    path = config_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".yaml.tmp")
    tmp.write_text(
        yaml.dump(config, default_flow_style=False, allow_unicode=True, sort_keys=False),
        encoding="utf-8",
    )
    tmp.replace(path)
    return path

intersecting_scopes ¶

intersecting_scopes(config: dict[str, Any] | None = None) -> list[IntersectSpec]

Return the ordered, de-duplicated list of intersecting scopes.

Reads .synapse/config.yaml when config is not supplied. Malformed or blank entries are dropped (best-effort, never raises) so a partial config still yields the valid subset. De-dup is keyed on (repo, project) so the same project can intersect from both a local and a federated source.

Source code in zettelkasten/synapse/config.py

def intersecting_scopes(config: dict[str, Any] | None = None) -> list[IntersectSpec]:
    """Return the ordered, de-duplicated list of intersecting scopes.

    Reads ``.synapse/config.yaml`` when ``config`` is not supplied. Malformed or
    blank entries are dropped (best-effort, never raises) so a partial config
    still yields the valid subset. De-dup is keyed on ``(repo, project)`` so the
    same project can intersect from both a local and a federated source.
    """
    if config is None:
        config = read_config()

    raw = config.get(CONFIG_KEY, [])
    if raw in (None, ""):
        return []
    if not isinstance(raw, list):
        logger.warning("synapse config %r must be a list, got %s", CONFIG_KEY, type(raw).__name__)
        return []

    scopes: list[IntersectSpec] = []
    seen: set[tuple[str | None, str]] = set()
    for item in raw:
        spec = _coerce_spec(item)
        if spec is None:
            continue
        key = (spec.repo, spec.project)
        if key in seen:
            continue
        seen.add(key)
        scopes.append(spec)
    return scopes

intersecting_projects ¶

intersecting_projects(config: dict[str, Any] | None = None) -> list[str]

Return the ordered, de-duplicated list of intersecting LOCAL project names.

Backward-compatible view over :func:intersecting_scopes that keeps only local (non-federated) entries — used where a plain local project name is expected (e.g. legacy callers). Federated scopes are surfaced via :func:intersecting_scopes.

Source code in zettelkasten/synapse/config.py

def intersecting_projects(config: dict[str, Any] | None = None) -> list[str]:
    """Return the ordered, de-duplicated list of intersecting LOCAL project names.

    Backward-compatible view over :func:`intersecting_scopes` that keeps only
    local (non-federated) entries — used where a plain local project name is
    expected (e.g. legacy callers). Federated scopes are surfaced via
    :func:`intersecting_scopes`.
    """
    return [s.project for s in intersecting_scopes(config) if s.repo is None]

parse_scope_token ¶

parse_scope_token(token: Any) -> IntersectSpec | None

Parse one override token (from synapse(projects=[...])) into a spec.

A <repo_id>:<project> token (the namespace separator : can never appear in a local project name) resolves to a FEDERATED spec; any other non-blank string is a local project. Returns None for blank/invalid input.

Source code in zettelkasten/synapse/config.py

def parse_scope_token(token: Any) -> IntersectSpec | None:
    """Parse one override token (from ``synapse(projects=[...])``) into a spec.

    A ``<repo_id>:<project>`` token (the namespace separator ``:`` can never
    appear in a local project name) resolves to a FEDERATED spec; any other
    non-blank string is a local project. Returns None for blank/invalid input.
    """
    if not isinstance(token, str):
        return None
    token = token.strip()
    if not token:
        return None
    from zettelkasten import federation

    split = federation.split_namespace(token)
    if split is not None:
        repo_id, project = split
        return IntersectSpec(project=project, repo=repo_id)
    return IntersectSpec(project=token)