memory.federation

memory.federation

Read-only federation helpers for projecting related memory trees.

FederatedRepo dataclass

Validated configuration for one related repository.

Source code in memory/federation.py

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

    id: str
    name: str
    path: Path
    memory_dir: Path

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. Both the sync writer (CLI) and this reader resolve the cache to the same place.

Source code in memory/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. Both the
    ``sync`` writer (CLI) and this reader resolve the cache to the same place.
    """

    base = storage._workspace_root() if root is None else root
    return base / ".angelo" / FEDERATION_CACHE_DIRNAME

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 memory/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."""

    config = storage.read_config() if config is None else config
    # Resolve relative repo paths against the same workspace root that anchors
    # .memory/config.yaml, so federation behaves consistently whether the server
    # runs from the project root or an installed console script elsewhere.
    root = storage._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

        memory_dir = repo_path / ".memory"
        project_path = memory_dir / "project.md"
        if not project_path.exists():
            errors.append(_repo_error(repo_id, "repo does not contain .memory/project.md", repo_path))
            continue

        name = str(item.get("name") or repo_path.name or repo_id)
        repos.append(FederatedRepo(id=repo_id, name=name, path=repo_path, memory_dir=memory_dir))
        seen_ids.add(repo_id)

    # Additively surface decrypted bundle caches under
    # .angelo/federation-cache/<peer_id>/ as 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).
    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
            memory_dir = cache_path / ".memory"
            if not (memory_dir / "project.md").exists():
                continue
            repos.append(FederatedRepo(id=repo_id, name=peer_dir.name or repo_id, path=cache_path, memory_dir=memory_dir))
            seen_ids.add(repo_id)

    return repos, errors

read_repo

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

Read a related repo's .memory directory without mutating it.

Source code in memory/federation.py

def read_repo(repo: FederatedRepo) -> dict[str, Any]:
    """Read a related repo's `.memory` directory without mutating it."""

    project = storage.read_file(repo.memory_dir / "project.md")
    project["description"] = project.pop("body", "")
    return {
        "repo": repo,
        "project": project,
        "entries": _read_dir(repo.memory_dir / "entries"),
        "documents": _read_dir(repo.memory_dir / "documents", body_key="description"),
        "skills": _read_dir(repo.memory_dir / "skills", body_key="content"),
        "sessions": _read_dir(repo.memory_dir / "sessions", body_key="search_text"),
    }

latest_mtime

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

Return the newest source-file mtime across configured federated repos.

Source code in memory/federation.py

def latest_mtime(config: dict[str, Any] | None = None, root: Path | None = None) -> float:
    """Return the newest source-file mtime across configured federated repos."""

    repos, _errors = configured_repos(config=config, root=root)
    latest = 0.0
    for repo in repos:
        for path in repo.memory_dir.rglob("*.md"):
            try:
                latest = max(latest, path.stat().st_mtime)
            except OSError:
                continue
    return latest

repo_embeddings_path

repo_embeddings_path(repo: FederatedRepo) -> Path

Path to a federated repo's embedding sidecar (.angelo/memory.embeddings.json).

Source code in memory/federation.py

def repo_embeddings_path(repo: FederatedRepo) -> Path:
    """Path to a federated repo's embedding sidecar (``.angelo/memory.embeddings.json``)."""

    return repo.path / ".angelo" / "memory.embeddings.json"

coordinator_state_path

coordinator_state_path(repo: FederatedRepo) -> Path

Path to a federated repo's coordinator run-state sidecar.

Mirrors the coordinator server's default layout: .angelo/coordinator-state.json as a sibling of the repo's .memory directory — the directory we already resolved for federation. Federation can't observe a peer's COORDINATOR_STATE_FILE / ANGELO_DIR env overrides, so we assume the default location (callers also probe the legacy fallbacks).

Source code in memory/federation.py

def coordinator_state_path(repo: FederatedRepo) -> Path:
    """Path to a federated repo's coordinator run-state sidecar.

    Mirrors the coordinator server's default layout: ``.angelo/coordinator-state.json``
    as a sibling of the repo's ``.memory`` directory — the directory we already
    resolved for federation. Federation can't observe a peer's
    ``COORDINATOR_STATE_FILE`` / ``ANGELO_DIR`` env overrides, so we assume the
    default location (callers also probe the legacy fallbacks).
    """

    return repo.path / ".angelo" / "coordinator-state.json"

coordinator_state_candidates

coordinator_state_candidates(repo: FederatedRepo) -> list[Path]

Ordered coordinator-state paths to probe for a federated repo.

Canonical .angelo location first, then the legacy .cassius and repo-root fallbacks — the same precedence the local dashboard uses. Shared by the read path (:func:read_coordinator_runs) and by cleanup writes so both agree on which peer files hold run state.

Source code in memory/federation.py

def coordinator_state_candidates(repo: FederatedRepo) -> list[Path]:
    """Ordered coordinator-state paths to probe for a federated repo.

    Canonical ``.angelo`` location first, then the legacy ``.cassius`` and
    repo-root fallbacks — the same precedence the local dashboard uses. Shared by
    the read path (:func:`read_coordinator_runs`) and by cleanup writes so both
    agree on which peer files hold run state.
    """

    return [
        coordinator_state_path(repo),
        repo.path / ".cassius" / "coordinator-state.json",
        repo.path / ".coordinator-state.json",
    ]

read_coordinator_runs

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

Read coordinator run dicts from a federated repo, best-effort and read-only.

Probes the canonical .angelo location first, then the legacy .cassius and repo-root fallbacks (same precedence the local dashboard uses). Returns an empty list when no readable state file exists or none carries a modern runs list — federation never raises on a missing/malformed peer.

Note: only the modern {"runs": [...]} format is understood here; the legacy flat task-list format is intentionally not reconstructed for peers.

Source code in memory/federation.py

def read_coordinator_runs(repo: FederatedRepo) -> list[dict[str, Any]]:
    """Read coordinator run dicts from a federated repo, best-effort and read-only.

    Probes the canonical ``.angelo`` location first, then the legacy ``.cassius``
    and repo-root fallbacks (same precedence the local dashboard uses). Returns an
    empty list when no readable state file exists or none carries a modern
    ``runs`` list — federation never raises on a missing/malformed peer.

    Note: only the modern ``{"runs": [...]}`` format is understood here; the
    legacy flat task-list format is intentionally not reconstructed for peers.
    """

    import json

    for path in coordinator_state_candidates(repo):
        if not path.exists():
            continue
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError):
            continue
        if not isinstance(data, dict):
            continue
        runs = data.get("runs")
        if isinstance(runs, list):
            return [run for run in runs if isinstance(run, dict)]
    return []

load_embeddings

load_embeddings(repo: FederatedRepo) -> dict[str, list[float]]

Read a federated repo's embedding sidecar, namespacing keys to <repo_id>:<entry_id>.

Returns an empty dict when the sidecar is missing or malformed — federation is always best-effort and read-only.

Source code in memory/federation.py

def load_embeddings(repo: FederatedRepo) -> dict[str, list[float]]:
    """Read a federated repo's embedding sidecar, namespacing keys to ``<repo_id>:<entry_id>``.

    Returns an empty dict when the sidecar is missing or malformed — federation is
    always best-effort and read-only.
    """

    import json

    path = repo_embeddings_path(repo)
    if not path.exists():
        return {}
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return {}
    if not isinstance(raw, dict):
        return {}
    return {
        namespace_id(repo.id, str(local_id)): vector
        for local_id, vector in raw.items()
        if isinstance(vector, list)
    }

latest_embeddings_mtime

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

Return the newest embedding-sidecar mtime across configured federated repos.

Source code in memory/federation.py

def latest_embeddings_mtime(config: dict[str, Any] | None = None, root: Path | None = None) -> float:
    """Return the newest embedding-sidecar mtime across configured federated repos."""

    repos, _errors = configured_repos(config=config, root=root)
    latest = 0.0
    for repo in repos:
        try:
            latest = max(latest, repo_embeddings_path(repo).stat().st_mtime)
        except OSError:
            continue
    return latest