Skip to content

zettelkasten.sharing

zettelkasten.sharing

Encryption-first sharing for the committed .zettelkasten store.

Mirrors :mod:memory.sharing: filter a store slice by owner-authored sharing/grants policy, encrypt each retained unit natively to only its recipients' age keys, and package a bundle a consumer can git-fetch and decrypt into a .zettelkasten-shaped federation cache.

Crypto and identity are REUSED from :mod:memory.sharing unchanged — there is a single firm identity registry (.memory/identities.yaml) and one local key (~/.angelo/identity.yaml); this package does NOT fork them or create a .zettelkasten/identities.yaml. Only the policy/scrub/bundle layers are Zettelkasten-specific.

Typical producer flow::

from memory import sharing as mem_sharing
from zettelkasten import sharing
registry = mem_sharing.load_registry()
result = sharing.build_bundle("/path/to/bundle-repo", registry=registry)

Typical consumer flow::

from zettelkasten import sharing
sharing.decrypt_bundle(
    "/path/to/fetched-bundle-repo",
    "/path/to/federation-cache/peer",   # writes <peer>/.zettelkasten/...
)  # uses ~/.angelo/identity.yaml

IdentityRegistry dataclass

An id -> RegistryEntry map of publishable (public-key) identities.

Source code in memory/sharing/identity.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
@dataclass(frozen=True)
class IdentityRegistry:
    """An ``id -> RegistryEntry`` map of publishable (public-key) identities."""

    entries: Mapping[str, RegistryEntry]

    def member_ids(self) -> set[str]:
        """All registered recipient ids (used to expand a ``firm`` scope)."""
        return set(self.entries)

    def public_key_for(self, recipient_id: str) -> str | None:
        """The public key for ``recipient_id``, or ``None`` if unregistered."""
        entry = self.entries.get(recipient_id)
        return entry.public_key if entry else None

    def signing_public_key_for(self, recipient_id: str) -> str | None:
        """The Ed25519 signing public key for ``recipient_id``.

        Returns ``None`` when the id is unregistered *or* when the registered
        entry predates signing (legacy entry with no ``signing_public_key``).
        """
        entry = self.entries.get(recipient_id)
        return entry.signing_public_key if entry else None

    def public_keys_for(self, recipient_ids: Iterable[str]) -> list[str]:
        """Resolve ids to public keys, silently dropping unregistered ids."""
        keys: list[str] = []
        for rid in recipient_ids:
            pk = self.public_key_for(rid)
            if pk:
                keys.append(pk)
        return keys

    def with_entry(self, entry: RegistryEntry) -> "IdentityRegistry":
        """A copy of this registry with ``entry`` added/overwritten (immutably)."""
        merged = dict(self.entries)
        merged[entry.id] = entry
        return IdentityRegistry(merged)

member_ids 

member_ids() -> set[str]

All registered recipient ids (used to expand a firm scope).

Source code in memory/sharing/identity.py 
150
151
152
def member_ids(self) -> set[str]:
    """All registered recipient ids (used to expand a ``firm`` scope)."""
    return set(self.entries)

public_key_for 

public_key_for(recipient_id: str) -> str | None

The public key for recipient_id, or None if unregistered.

Source code in memory/sharing/identity.py 
154
155
156
157
def public_key_for(self, recipient_id: str) -> str | None:
    """The public key for ``recipient_id``, or ``None`` if unregistered."""
    entry = self.entries.get(recipient_id)
    return entry.public_key if entry else None

signing_public_key_for 

signing_public_key_for(recipient_id: str) -> str | None

The Ed25519 signing public key for recipient_id.

Returns None when the id is unregistered or when the registered entry predates signing (legacy entry with no signing_public_key).

Source code in memory/sharing/identity.py 
159
160
161
162
163
164
165
166
def signing_public_key_for(self, recipient_id: str) -> str | None:
    """The Ed25519 signing public key for ``recipient_id``.

    Returns ``None`` when the id is unregistered *or* when the registered
    entry predates signing (legacy entry with no ``signing_public_key``).
    """
    entry = self.entries.get(recipient_id)
    return entry.signing_public_key if entry else None

public_keys_for 

public_keys_for(recipient_ids: Iterable[str]) -> list[str]

Resolve ids to public keys, silently dropping unregistered ids.

Source code in memory/sharing/identity.py 
168
169
170
171
172
173
174
175
def public_keys_for(self, recipient_ids: Iterable[str]) -> list[str]:
    """Resolve ids to public keys, silently dropping unregistered ids."""
    keys: list[str] = []
    for rid in recipient_ids:
        pk = self.public_key_for(rid)
        if pk:
            keys.append(pk)
    return keys

with_entry 

with_entry(entry: RegistryEntry) -> 'IdentityRegistry'

A copy of this registry with entry added/overwritten (immutably).

Source code in memory/sharing/identity.py 
177
178
179
180
181
def with_entry(self, entry: RegistryEntry) -> "IdentityRegistry":
    """A copy of this registry with ``entry`` added/overwritten (immutably)."""
    merged = dict(self.entries)
    merged[entry.id] = entry
    return IdentityRegistry(merged)

LocalIdentity dataclass

This machine's identity: a logical id plus its keypair(s).

Persisted (with the private keys) to ~/.angelo/identity.yaml. Carries both the age X25519 encryption keypair and the Ed25519 signing keypair; the signing fields are optional (default None) so a legacy identity file with no signing key still loads.

Source code in memory/sharing/identity.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
@dataclass(frozen=True)
class LocalIdentity:
    """This machine's identity: a logical id plus its keypair(s).

    Persisted (with the private keys) to ``~/.angelo/identity.yaml``. Carries both
    the ``age`` X25519 encryption keypair and the Ed25519 signing keypair; the
    signing fields are optional (default ``None``) so a legacy identity file with
    no signing key still loads.
    """

    id: str
    private_key: str
    public_key: str
    display_name: str | None = None
    github: str | None = None
    signing_public_key: str | None = None
    signing_private_key: str | None = None

    def to_pyrage_identity(self) -> x25519.Identity:
        """Parse the private key into a ``pyrage`` identity for decryption."""
        return x25519.Identity.from_str(self.private_key)

    def to_registry_entry(self) -> RegistryEntry:
        """The public-only :class:`RegistryEntry` for publishing to the registry."""
        return RegistryEntry(
            id=self.id,
            public_key=self.public_key,
            display_name=self.display_name,
            github=self.github,
            signing_public_key=self.signing_public_key,
        )

to_pyrage_identity 

to_pyrage_identity() -> Identity

Parse the private key into a pyrage identity for decryption.

Source code in memory/sharing/identity.py 
129
130
131
def to_pyrage_identity(self) -> x25519.Identity:
    """Parse the private key into a ``pyrage`` identity for decryption."""
    return x25519.Identity.from_str(self.private_key)

to_registry_entry 

to_registry_entry() -> RegistryEntry

The public-only :class:RegistryEntry for publishing to the registry.

Source code in memory/sharing/identity.py 
133
134
135
136
137
138
139
140
141
def to_registry_entry(self) -> RegistryEntry:
    """The public-only :class:`RegistryEntry` for publishing to the registry."""
    return RegistryEntry(
        id=self.id,
        public_key=self.public_key,
        display_name=self.display_name,
        github=self.github,
        signing_public_key=self.signing_public_key,
    )

RegistryEntry dataclass

One published identity: a logical id bound to a public key.

display_name and github are optional and used only for attribution / discovery — never for access control (the public key is the anchor). signing_public_key is the identity's Ed25519 signing public key (base64, public part only); optional so legacy registries without it still parse.

Source code in memory/sharing/identity.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@dataclass(frozen=True)
class RegistryEntry:
    """One published identity: a logical ``id`` bound to a public key.

    ``display_name`` and ``github`` are optional and used only for attribution /
    discovery — never for access control (the public key is the anchor).
    ``signing_public_key`` is the identity's Ed25519 *signing* public key (base64,
    public part only); optional so legacy registries without it still parse.
    """

    id: str
    public_key: str
    display_name: str | None = None
    github: str | None = None
    signing_public_key: str | None = None

BuildResult dataclass

Summary of a :func:build_bundle run (returned in memory, never written).

Source code in zettelkasten/sharing/bundle.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
@dataclass(frozen=True)
class BuildResult:
    """Summary of a :func:`build_bundle` run (returned in memory, never written)."""

    manifest: BundleManifest
    output_dir: Path
    note_unit_count: int
    source_unit_count: int
    project_unit_count: int
    citation_unit_count: int
    review_unit_count: int
    org_unit_count: int
    table_unit_count: int
    outline_unit_count: int
    sidecar_unit_count: int
    recipient_ids: tuple[str, ...]
    skipped_unregistered_ids: tuple[str, ...]
    # Number of notes covered by the sibling recipient-index export (0 when no
    # note ships or the export could not be written). See
    # :mod:`zettelkasten.synapse.sharing.recipient_index`.
    recipient_index_note_count: int = 0

BundleManifest dataclass

The cleartext manifest.json at the bundle root.

Source code in zettelkasten/sharing/bundle.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
@dataclass(frozen=True)
class BundleManifest:
    """The cleartext ``manifest.json`` at the bundle root."""

    format: str
    format_version: str
    created_at: str
    units: tuple[BundleUnit, ...]
    producer_id: str | None = None

    def to_dict(self) -> dict[str, Any]:
        data: dict[str, Any] = {
            "format": self.format,
            "format_version": self.format_version,
            "created_at": self.created_at,
            "units": [u.to_dict() for u in self.units],
        }
        if self.producer_id:
            data["producer_id"] = self.producer_id
        return data

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "BundleManifest":
        raw_units = data.get("units") or []
        units = tuple(BundleUnit.from_dict(u) for u in raw_units if isinstance(u, dict))
        producer = data.get("producer_id")
        return cls(
            format=str(data.get("format") or ""),
            format_version=str(data.get("format_version") or ""),
            created_at=str(data.get("created_at") or ""),
            units=units,
            producer_id=str(producer) if producer else None,
        )

BundleUnit dataclass

A pointer to one encrypted unit inside a bundle (kind + relative path).

Source code in zettelkasten/sharing/bundle.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
@dataclass(frozen=True)
class BundleUnit:
    """A pointer to one encrypted unit inside a bundle (``kind`` + relative path)."""

    kind: str
    path: str

    def to_dict(self) -> dict[str, str]:
        return {"kind": self.kind, "path": self.path}

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "BundleUnit":
        return cls(kind=str(data.get("kind") or ""), path=str(data.get("path") or ""))

DecryptResult dataclass

Summary of a :func:decrypt_bundle run.

Source code in zettelkasten/sharing/bundle.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
@dataclass(frozen=True)
class DecryptResult:
    """Summary of a :func:`decrypt_bundle` run."""

    output_dir: Path
    note_keys: tuple[str, ...]
    source_boxes: tuple[str, ...]
    project_names: tuple[str, ...]
    citation_ids: tuple[str, ...]
    review_names: tuple[str, ...]
    org_ids: tuple[str, ...]
    table_reviews: tuple[str, ...]
    outline_reviews: tuple[str, ...]
    sidecar_paths: tuple[str, ...]
    total_units: int

ProjectionResult dataclass

Summary of a :func:project_for_caller run.

Mirrors :class:zettelkasten.sharing.bundle.BuildResult's per-kind counts and :class:~zettelkasten.sharing.bundle.DecryptResult's per-kind identifiers, so a projection can be compared directly against a decrypt for parity. ids use the same scheme :func:~zettelkasten.sharing.bundle.decrypt_bundle reports (note "<box>/<id>", source box name, project name, citation id, review name, org id, table/outline review name, sidecar "<box>/<rel>").

Source code in zettelkasten/sharing/projection.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@dataclass(frozen=True)
class ProjectionResult:
    """Summary of a :func:`project_for_caller` run.

    Mirrors :class:`zettelkasten.sharing.bundle.BuildResult`'s per-kind counts and
    :class:`~zettelkasten.sharing.bundle.DecryptResult`'s per-kind identifiers, so
    a projection can be compared directly against a decrypt for parity. ``ids``
    use the same scheme :func:`~zettelkasten.sharing.bundle.decrypt_bundle`
    reports (note ``"<box>/<id>"``, source box name, project name, citation id,
    review name, org id, table/outline review name, sidecar ``"<box>/<rel>"``).
    """

    caller_id: str
    output_dir: Path  # ``<out_root>/.zettelkasten``
    note_keys: tuple[str, ...]
    source_boxes: tuple[str, ...]
    project_names: tuple[str, ...]
    citation_ids: tuple[str, ...]
    review_names: tuple[str, ...]
    org_ids: tuple[str, ...]
    table_reviews: tuple[str, ...]
    outline_reviews: tuple[str, ...]
    sidecar_paths: tuple[str, ...]
    note_unit_count: int
    source_unit_count: int
    project_unit_count: int
    citation_unit_count: int
    review_unit_count: int
    org_unit_count: int
    table_unit_count: int
    outline_unit_count: int
    sidecar_unit_count: int
    total_units: int
    # True iff a TRANSIENT I/O fault during reconcile (owned-sidecar record read
    # or dir listing) forced this run to SKIP eviction — the projection was served
    # additively but may still hold just-revoked plaintext on disk, so a caller
    # (e.g. the broker cache) must treat this build as serve-but-don't-cache.
    # Default False (clean pass).
    transient_io_error: bool = False

Scope dataclass

A visibility scope attached to a project or source box.

audience is only meaningful when kind == 'bilateral' and holds the recipient ids that may access the subtree.

Source code in zettelkasten/sharing/config.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
@dataclass(frozen=True)
class Scope:
    """A visibility scope attached to a project or source box.

    ``audience`` is only meaningful when ``kind == 'bilateral'`` and holds the
    recipient ids that may access the subtree.
    """

    kind: str = SCOPE_PRIVATE
    audience: tuple[str, ...] = ()

SharingConfig dataclass

Parsed, typed view of the sharing + grants config blocks.

  • scopes maps a project name (or a source-box name) to its :class:Scope.
  • grants maps a key (project name, box name, box/note_id, cross/note_id alias, or citation:<id>) to the tuple of recipient ids granted access to that unit (and, for a project/box key, its closure).
Source code in zettelkasten/sharing/config.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@dataclass(frozen=True)
class SharingConfig:
    """Parsed, typed view of the ``sharing`` + ``grants`` config blocks.

    * ``scopes`` maps a project name (or a source-box name) to its :class:`Scope`.
    * ``grants`` maps a key (project name, box name, ``box/note_id``,
      ``cross/note_id`` alias, or ``citation:<id>``) to the tuple of recipient
      ids granted access to that unit (and, for a project/box key, its closure).
    """

    scopes: Mapping[str, Scope] = field(default_factory=dict)
    grants: Mapping[str, tuple[str, ...]] = field(default_factory=dict)
    # Per-unit transport tier keyed exactly like ``scopes``/``grants`` (a project
    # name, box name, or per-note key). A note inherits (per-note > box >
    # project-closure) and ultimately defaults to :data:`TRANSPORT_OFFLINE`
    # (deny-by-default). Values are validated to ``offline``/``hosted`` at parse.
    transports: Mapping[str, str] = field(default_factory=dict)

    def is_empty(self) -> bool:
        """True when no sharing policy is configured (today's behaviour)."""
        return not self.scopes and not self.grants and not self.transports

is_empty 

is_empty() -> bool

True when no sharing policy is configured (today's behaviour).

Source code in zettelkasten/sharing/config.py 
123
124
125
def is_empty(self) -> bool:
    """True when no sharing policy is configured (today's behaviour)."""
    return not self.scopes and not self.grants and not self.transports

decrypt 

decrypt(ciphertext: bytes, private_key: PrivateKey) -> bytes

Decrypt ciphertext with private_key.

private_key may be an AGE-SECRET-KEY-... string or a pre-parsed :class:pyrage.x25519.Identity. Raises :class:DecryptError if the ciphertext was not encrypted to this key.

Source code in memory/sharing/crypto.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
def decrypt(ciphertext: bytes, private_key: PrivateKey) -> bytes:
    """Decrypt ``ciphertext`` with ``private_key``.

    ``private_key`` may be an ``AGE-SECRET-KEY-...`` string or a pre-parsed
    :class:`pyrage.x25519.Identity`. Raises :class:`DecryptError` if the
    ciphertext was not encrypted to this key.
    """
    identity = (
        private_key
        if isinstance(private_key, x25519.Identity)
        else x25519.Identity.from_str(str(private_key))
    )
    return pyrage.decrypt(bytes(ciphertext), [identity])

encrypt 

encrypt(plaintext: bytes, recipient_pubkeys: list[str]) -> bytes

Encrypt plaintext to every recipient in recipient_pubkeys.

Each key is an age X25519 recipient string (age1...). Any recipient holding the matching private key can :func:decrypt the result. Returns the binary age ciphertext.

Raises :class:ValueError if no recipients are given (encrypting to nobody would produce an unopenable blob) or if a recipient string is malformed (surfaced as :class:pyrage.RecipientError).

Source code in memory/sharing/crypto.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def encrypt(plaintext: bytes, recipient_pubkeys: list[str]) -> bytes:
    """Encrypt ``plaintext`` to every recipient in ``recipient_pubkeys``.

    Each key is an ``age`` X25519 recipient string (``age1...``). Any recipient
    holding the matching private key can :func:`decrypt` the result. Returns the
    binary ``age`` ciphertext.

    Raises :class:`ValueError` if no recipients are given (encrypting to nobody
    would produce an unopenable blob) or if a recipient string is malformed
    (surfaced as :class:`pyrage.RecipientError`).
    """
    if not recipient_pubkeys:
        raise ValueError("at least one recipient public key is required to encrypt")
    recipients = [x25519.Recipient.from_str(pk) for pk in _dedupe(list(recipient_pubkeys))]
    return pyrage.encrypt(bytes(plaintext), recipients)

create_identity 

create_identity(identity_id: str, *, display_name: str | None = None, github: str | None = None, path: Path | str | None = None, overwrite: bool = False) -> LocalIdentity

Generate a keypair, build a :class:LocalIdentity, and persist it.

Refuses to clobber an existing identity file unless overwrite=True — a lost private key cannot be recovered, so overwriting is opt-in.

Source code in memory/sharing/identity.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
def create_identity(
    identity_id: str,
    *,
    display_name: str | None = None,
    github: str | None = None,
    path: Path | str | None = None,
    overwrite: bool = False,
) -> LocalIdentity:
    """Generate a keypair, build a :class:`LocalIdentity`, and persist it.

    Refuses to clobber an existing identity file unless ``overwrite=True`` — a
    lost private key cannot be recovered, so overwriting is opt-in.
    """
    dest = Path(path) if path is not None else default_identity_path()
    if dest.exists() and not overwrite:
        raise FileExistsError(
            f"identity file already exists at {dest}; pass overwrite=True to replace it"
        )
    keypair = generate_keypair()
    identity = LocalIdentity(
        id=identity_id,
        private_key=keypair.private_key,
        public_key=keypair.public_key,
        display_name=display_name,
        github=github,
        signing_public_key=keypair.signing_public_key,
        signing_private_key=keypair.signing_private_key,
    )
    save_local_identity(identity, dest)
    return identity

default_identity_path 

default_identity_path() -> Path

Path of this machine's private identity file: ~/.angelo/identity.yaml.

Under HOME (never the workspace) so the private key cannot be committed.

Source code in memory/sharing/identity.py 
55
56
57
58
59
60
def default_identity_path() -> Path:
    """Path of this machine's private identity file: ``~/.angelo/identity.yaml``.

    Under HOME (never the workspace) so the private key cannot be committed.
    """
    return Path.home() / ".angelo" / "identity.yaml"

default_registry_path 

default_registry_path() -> Path

Default path of the shared identity registry (.memory/identities.yaml).

Anchored to the memory store so it is versioned alongside the tree. Contains public keys only, so committing it is safe. Callers may override the path.

Source code in memory/sharing/identity.py 
63
64
65
66
67
68
69
70
71
72
def default_registry_path() -> Path:
    """Default path of the shared identity registry (``.memory/identities.yaml``).

    Anchored to the memory store so it is versioned alongside the tree. Contains
    public keys only, so committing it is safe. Callers may override the path.
    """
    # Imported lazily to avoid a hard import-time dependency on storage globals.
    from memory import storage

    return storage.MEMORY_DIR / "identities.yaml"

load_local_identity 

load_local_identity(path: Path | str | None = None) -> LocalIdentity | None

Load this machine's identity, or None if the file is absent/invalid.

The public key is derived from the private key when the file omits it, so an identity file carrying only id + private_key still loads.

Source code in memory/sharing/identity.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def load_local_identity(path: Path | str | None = None) -> LocalIdentity | None:
    """Load this machine's identity, or ``None`` if the file is absent/invalid.

    The public key is derived from the private key when the file omits it, so an
    identity file carrying only ``id`` + ``private_key`` still loads.
    """
    src = Path(path) if path is not None else default_identity_path()
    if not src.exists():
        return None
    try:
        data = yaml.safe_load(src.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError) as exc:
        logger.warning("could not read identity file %s: %s", src, exc)
        return None
    if not isinstance(data, Mapping):
        return None
    identity_id = str(data.get("id") or "").strip()
    private_key = str(data.get("private_key") or "").strip()
    if not identity_id or not private_key:
        return None
    public_key = str(data.get("public_key") or "").strip()
    if not public_key:
        try:
            public_key = str(x25519.Identity.from_str(private_key).to_public())
        except Exception as exc:  # pragma: no cover - malformed key
            logger.warning("identity file %s has an unusable private key: %s", src, exc)
            return None
    display_name = data.get("display_name")
    github = data.get("github")
    signing_public_key = data.get("signing_public_key")
    signing_private_key = data.get("signing_private_key")
    return LocalIdentity(
        id=identity_id,
        private_key=private_key,
        public_key=public_key,
        display_name=str(display_name) if display_name else None,
        github=str(github) if github else None,
        signing_public_key=str(signing_public_key) if signing_public_key else None,
        signing_private_key=str(signing_private_key) if signing_private_key else None,
    )

load_registry 

load_registry(path: Path | str | None = None) -> IdentityRegistry

Load the shared identity registry, or an empty one if absent/invalid.

Expected YAML shape::

identities:
  alice:
    public_key: age1...
    display_name: Alice
    github: alice-gh
    signing_public_key: ...   # optional Ed25519 signing public key (base64)
  bob: age1...          # shorthand: id -> public key (no signing key)

Malformed rows are skipped rather than raising, so a partially-authored registry still loads the good rows.

Source code in memory/sharing/identity.py 
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
def load_registry(path: Path | str | None = None) -> IdentityRegistry:
    """Load the shared identity registry, or an empty one if absent/invalid.

    Expected YAML shape::

        identities:
          alice:
            public_key: age1...
            display_name: Alice
            github: alice-gh
            signing_public_key: ...   # optional Ed25519 signing public key (base64)
          bob: age1...          # shorthand: id -> public key (no signing key)

    Malformed rows are skipped rather than raising, so a partially-authored
    registry still loads the good rows.
    """
    src = Path(path) if path is not None else default_registry_path()
    if not src.exists():
        return IdentityRegistry({})
    try:
        data = yaml.safe_load(src.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError) as exc:
        logger.warning("could not read identity registry %s: %s", src, exc)
        return IdentityRegistry({})
    raw = data.get("identities") if isinstance(data, Mapping) else None
    if not isinstance(raw, Mapping):
        return IdentityRegistry({})
    entries: dict[str, RegistryEntry] = {}
    for entry_id, info in raw.items():
        eid = str(entry_id).strip()
        if not eid:
            continue
        parsed = _parse_registry_entry(eid, info)
        if parsed is not None:
            entries[eid] = parsed
    return IdentityRegistry(entries)

save_registry 

save_registry(registry: IdentityRegistry, path: Path | str | None = None) -> Path

Write the identity registry (public keys only) to disk.

Source code in memory/sharing/identity.py 
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
def save_registry(registry: IdentityRegistry, path: Path | str | None = None) -> Path:
    """Write the identity registry (public keys only) to disk."""
    dest = Path(path) if path is not None else default_registry_path()
    identities: dict[str, Any] = {}
    for entry_id, entry in registry.entries.items():
        row: dict[str, Any] = {"public_key": entry.public_key}
        if entry.display_name:
            row["display_name"] = entry.display_name
        if entry.github:
            row["github"] = entry.github
        if entry.signing_public_key:
            row["signing_public_key"] = entry.signing_public_key
        identities[entry_id] = row
    atomic_write(
        dest,
        yaml.dump({"identities": identities}, default_flow_style=False, sort_keys=True),
    )
    return dest

build_bundle 

build_bundle(output_dir: Path | str, *, registry: IdentityRegistry, store_root: Path | str | None = None, sharing: SharingConfig | None = None, producer_id: str | None = None) -> BuildResult

Filter store units by grants, encrypt per recipient, and write a bundle.

Pipeline (filtering happens BEFORE any encryption):

  1. Enumerate note / source_meta / project / citation units from store_root (default: the local .zettelkasten store).
  2. Resolve each unit's recipient id set from sharing (default: parsed from store_root/config.yaml), expanding a firm scope to the registry members. Units resolving to no registered recipient are dropped (physically omitted).
  3. For each retained unit compute its ALLOWED set — the unit-keys whose recipient set is a superset of this unit's recipients — scrub cross-unit references down to that set (see :mod:zettelkasten.sharing.scrub), wrap the scrubbed bytes in a placement envelope, encrypt to the recipients' public keys and write units/<digest>.age.
  4. Write the cleartext manifest.json.

This function OWNS output_dir/units and output_dir/manifest.json — the units directory is rebuilt from scratch on every call so stale/revoked units never linger. Other files (e.g. .git) are left untouched. With no sharing policy configured it writes an empty bundle.

Source code in zettelkasten/sharing/bundle.py 
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
def build_bundle(
    output_dir: Path | str,
    *,
    registry: IdentityRegistry,
    store_root: Path | str | None = None,
    sharing: SharingConfig | None = None,
    producer_id: str | None = None,
) -> BuildResult:
    """Filter store units by grants, encrypt per recipient, and write a bundle.

    Pipeline (filtering happens BEFORE any encryption):

    1. Enumerate ``note`` / ``source_meta`` / ``project`` / ``citation`` units
       from ``store_root`` (default: the local ``.zettelkasten`` store).
    2. Resolve each unit's recipient id set from ``sharing`` (default: parsed
       from ``store_root/config.yaml``), expanding a ``firm`` scope to the
       ``registry`` members. Units resolving to no registered recipient are
       dropped (physically omitted).
    3. For each retained unit compute its ALLOWED set — the unit-keys whose
       recipient set is a superset of this unit's recipients — scrub cross-unit
       references down to that set (see :mod:`zettelkasten.sharing.scrub`),
       wrap the scrubbed bytes in a placement envelope, encrypt to the
       recipients' public keys and write ``units/<digest>.age``.
    4. Write the cleartext ``manifest.json``.

    This function OWNS ``output_dir/units`` and ``output_dir/manifest.json`` —
    the units directory is rebuilt from scratch on every call so stale/revoked
    units never linger. Other files (e.g. ``.git``) are left untouched. With no
    sharing policy configured it writes an empty bundle.
    """
    output_dir = Path(output_dir)

    # Enumerate + recipient-resolve + filter (shared with project_for_caller):
    # store_root / sharing defaulting happens inside _collect_pending. This is the
    # OFFLINE P2P path: it collects ONLY ``offline`` units so a ``hosted`` unit is
    # never packaged into an offline bundle. With no transport declared everything
    # defaults ``offline``, so this filter is a no-op and the bundle is byte-for-
    # byte identical to before the transport tier existed.
    pending, bundled_recip, skipped_unregistered = _collect_pending(
        store_root, sharing, registry, {TRANSPORT_OFFLINE}
    )

    # ---- Prepare output (rebuild units/ from scratch) ---------------------
    units_dir = output_dir / UNITS_DIRNAME
    if units_dir.exists():
        shutil.rmtree(units_dir)
    units_dir.mkdir(parents=True, exist_ok=True)

    # ---- Per-unit scrub (superset-allowed) + encrypt ----------------------
    units: list[BundleUnit] = []
    all_recipient_ids: set[str] = set()
    counts = {
        KIND_NOTE: 0, KIND_SOURCE_META: 0, KIND_PROJECT: 0, KIND_CITATION: 0,
        KIND_REVIEW: 0, KIND_ORG: 0, KIND_TABLE: 0, KIND_OUTLINE: 0,
        KIND_SIDECAR: 0,
    }

    for kind, unit_key, header, payload, recips, pubkeys in _scrub_pending(
        pending, bundled_recip
    ):
        ciphertext = crypto.encrypt(_encode_envelope(header, payload), pubkeys)
        filename = _unit_filename(unit_key, kind)
        atomic_write(units_dir / filename, ciphertext)
        units.append(BundleUnit(kind=kind, path=f"{UNITS_DIRNAME}/{filename}"))
        counts[kind] += 1
        all_recipient_ids.update(
            rid for rid in recips if registry.public_key_for(rid)
        )

    # ---- 3. Write the cleartext manifest ----------------------------------
    manifest = BundleManifest(
        format=BUNDLE_FORMAT,
        format_version=BUNDLE_FORMAT_VERSION,
        created_at=datetime.now(timezone.utc).isoformat(),
        units=tuple(units),
        producer_id=producer_id,
    )
    atomic_write(output_dir / MANIFEST_NAME, json.dumps(manifest.to_dict(), indent=2))

    # ---- 4. Emit the firm-scoped recipient-index export -------------------
    # A sibling artifact naming, for EXACTLY the notes shipped above, which firm
    # recipients may open each — the enabler for a PEER's cross-producer synapse
    # edges (see :mod:`zettelkasten.synapse.sharing.recipient_index`). It reuses
    # the per-note recipients already resolved here (``bundled_recip`` KIND_NOTE),
    # is encrypted firm-wide, and is rebuilt from scratch (revoked notes drop
    # out). Kept best-effort: an export failure never aborts the bundle publish.
    recipient_index_note_count = 0
    try:
        from zettelkasten.synapse.sharing import recipient_index as _recipient_index

        shipped_note_recip = {
            uk: r for (k, uk), r in bundled_recip.items() if k == KIND_NOTE
        }
        # Sign the export with this producer's Ed25519 signing key so peers can
        # verify tamper-evident, non-repudiable binding to the producer. A missing
        # local identity / signing key -> publish UNSIGNED (build_recipient_index
        # warns); load_local_identity raising must never abort the bundle.
        signing_private_key: str | None = None
        try:
            _local_identity = load_local_identity()
            if _local_identity is not None:
                signing_private_key = _local_identity.signing_private_key
        except Exception as exc:  # pragma: no cover - defensive
            logger.warning(
                "could not load local identity for recipient-index signing: %s", exc
            )
        if not signing_private_key:
            logger.warning(
                "recipient-index export: no local signing key; publishing UNSIGNED"
            )
        ri_result = _recipient_index.build_recipient_index(
            output_dir, shipped_note_recip, registry, producer_id=producer_id,
            signing_private_key=signing_private_key,
        )
        recipient_index_note_count = ri_result.note_count
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("failed to write recipient-index export: %s", exc)

    return BuildResult(
        manifest=manifest,
        output_dir=output_dir,
        note_unit_count=counts[KIND_NOTE],
        source_unit_count=counts[KIND_SOURCE_META],
        project_unit_count=counts[KIND_PROJECT],
        citation_unit_count=counts[KIND_CITATION],
        review_unit_count=counts[KIND_REVIEW],
        org_unit_count=counts[KIND_ORG],
        table_unit_count=counts[KIND_TABLE],
        outline_unit_count=counts[KIND_OUTLINE],
        sidecar_unit_count=counts[KIND_SIDECAR],
        recipient_ids=tuple(sorted(all_recipient_ids)),
        skipped_unregistered_ids=tuple(sorted(skipped_unregistered)),
        recipient_index_note_count=recipient_index_note_count,
    )

decrypt_bundle 

decrypt_bundle(bundle_dir: Path | str, out_root: Path | str, *, identity: LocalIdentity | Any | None = None, private_key: str | None = None) -> DecryptResult

Decrypt every unit this key can open and materialize a .zettelkasten cache.

Units are written into <out_root>/.zettelkasten/ as <box>/<id>.md (notes), <box>/_meta.yaml (source meta), _projects/<name>.yaml (projects) and _citations/<id>.yaml (citations), so :func:zettelkasten.federation.configured_repos renders the result like any other peer. Each unit's placement is read from its decrypted envelope header and validated / path-jailed before use.

The cache is RECONCILED to the current bundle: stale units (dropped from a newer, re-scoped bundle or no longer openable by this key) are evicted from each owned location. A transient I/O fault during the pass SKIPS eviction that run (never destroys valid cached plaintext); revocation still works on any clean pass. A missing/corrupt manifest is treated as an empty bundle (still reconciled); an unreadable manifest (transient I/O) aborts without reconciling.

Provide the key via identity (a :class:LocalIdentity / pyrage identity), private_key (an AGE-SECRET-KEY-... string), or neither (to load ~/.angelo/identity.yaml).

Source code in zettelkasten/sharing/bundle.py 
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
def decrypt_bundle(
    bundle_dir: Path | str,
    out_root: Path | str,
    *,
    identity: LocalIdentity | Any | None = None,
    private_key: str | None = None,
) -> DecryptResult:
    """Decrypt every unit this key can open and materialize a ``.zettelkasten`` cache.

    Units are written into ``<out_root>/.zettelkasten/`` as
    ``<box>/<id>.md`` (notes), ``<box>/_meta.yaml`` (source meta),
    ``_projects/<name>.yaml`` (projects) and ``_citations/<id>.yaml``
    (citations), so :func:`zettelkasten.federation.configured_repos` renders the
    result like any other peer. Each unit's placement is read from its decrypted
    envelope header and validated / path-jailed before use.

    The cache is RECONCILED to the current bundle: stale units (dropped from a
    newer, re-scoped bundle or no longer openable by this key) are evicted from
    each owned location. A transient I/O fault during the pass SKIPS eviction
    that run (never destroys valid cached plaintext); revocation still works on
    any clean pass. A missing/corrupt manifest is treated as an empty bundle
    (still reconciled); an unreadable manifest (transient I/O) aborts without
    reconciling.

    Provide the key via ``identity`` (a :class:`LocalIdentity` / pyrage
    identity), ``private_key`` (an ``AGE-SECRET-KEY-...`` string), or neither (to
    load ``~/.angelo/identity.yaml``).
    """
    bundle_dir = Path(bundle_dir)
    zettel_dir = Path(out_root) / ".zettelkasten"
    key = _resolve_private_key(identity, private_key)

    try:
        manifest = read_manifest(bundle_dir)
    except FileNotFoundError as exc:
        logger.warning("bundle manifest absent (%s); reconciling cache as empty", exc)
        manifest = BundleManifest(format="", format_version="", created_at="", units=())
    except (json.JSONDecodeError, ValueError) as exc:
        logger.warning("bundle manifest corrupt (%s); reconciling cache as empty", exc)
        manifest = BundleManifest(format="", format_version="", created_at="", units=())
    except OSError as exc:
        logger.warning(
            "bundle manifest unreadable due to transient I/O (%s); aborting decrypt "
            "without reconciling to preserve valid cached plaintext", exc
        )
        return DecryptResult(
            output_dir=zettel_dir,
            note_keys=(),
            source_boxes=(),
            project_names=(),
            citation_ids=(),
            review_names=(),
            org_ids=(),
            table_reviews=(),
            outline_reviews=(),
            sidecar_paths=(),
            total_units=0,
        )

    # ---- Phase 1: decrypt everything this key can open, buffering writes ----
    # (dest, payload, identifier) per kind.
    note_writes: list[tuple[str, Path, bytes, str]] = []  # (box, dest, payload, key)
    meta_writes: list[tuple[str, Path, bytes]] = []       # (box, dest, payload)
    project_writes: list[tuple[Path, bytes, str]] = []
    citation_writes: list[tuple[Path, bytes, str]] = []
    review_writes: list[tuple[Path, bytes, str]] = []
    org_writes: list[tuple[Path, bytes, str]] = []
    table_writes: list[tuple[Path, bytes, str]] = []
    outline_writes: list[tuple[Path, bytes, str]] = []
    sidecar_writes: list[tuple[Path, bytes, str]] = []
    seen_notes: dict[str, set[str]] = {}   # box -> {filename}
    seen_meta_boxes: set[str] = set()
    seen_projects: set[str] = set()
    seen_citations: set[str] = set()
    # v2 editorial kinds are reconciled by their destination filename within the
    # dir they own (``_reviews`` holds three shapes; ``_organizations`` is nested).
    seen_reviews: set[str] = set()      # <name>.yaml
    seen_tables: set[str] = set()       # <name>.tables.json
    seen_outlines: set[str] = set()     # <name>.outlines.json
    seen_org_paths: set[Path] = set()   # resolved <id>.json dest paths
    transient_io_error = False

    for unit in manifest.units:
        unit_path = _resolve_unit_path(bundle_dir, unit)
        if unit_path is None:
            continue
        if not unit_path.exists():
            logger.warning("bundle unit missing on disk: %s", unit_path)
            continue
        try:
            ciphertext = unit_path.read_bytes()
        except OSError as exc:
            logger.warning(
                "could not read bundle unit %s (transient I/O): %s; "
                "skipping stale-eviction this run", unit_path, exc
            )
            transient_io_error = True
            continue
        try:
            plaintext = crypto.decrypt(ciphertext, key)
        except crypto.DecryptError:
            continue  # not encrypted to this key
        except Exception as exc:  # pragma: no cover - defensive
            logger.warning("failed to decrypt unit %s: %s", unit.path, exc)
            continue

        decoded = _decode_envelope(plaintext)
        if decoded is None:
            logger.warning("decrypted unit %s has a malformed envelope; skipping", unit.path)
            continue
        header, payload = decoded
        dest, kind, ident = _unit_dest(header, zettel_dir)
        if dest is None:
            logger.warning("decrypted unit %s has no usable placement; skipping", unit.path)
            continue

        if kind == KIND_NOTE:
            box = str(header.get("box"))
            note_writes.append((box, dest, payload, ident))
            seen_notes.setdefault(box, set()).add(dest.name)
        elif kind == KIND_SOURCE_META:
            box = str(header.get("box"))
            meta_writes.append((box, dest, payload))
            seen_meta_boxes.add(box)
        elif kind == KIND_PROJECT:
            project_writes.append((dest, payload, ident))
            seen_projects.add(dest.name)
        elif kind == KIND_CITATION:
            citation_writes.append((dest, payload, ident))
            seen_citations.add(dest.name)
        elif kind == KIND_REVIEW:
            review_writes.append((dest, payload, ident))
            seen_reviews.add(dest.name)
        elif kind == KIND_TABLE:
            table_writes.append((dest, payload, ident))
            seen_tables.add(dest.name)
        elif kind == KIND_OUTLINE:
            outline_writes.append((dest, payload, ident))
            seen_outlines.add(dest.name)
        elif kind == KIND_ORG:
            org_writes.append((dest, payload, ident))
            seen_org_paths.add(dest.resolve())
        elif kind == KIND_SIDECAR:
            sidecar_writes.append((dest, payload, ident))

    # ---- Phase 2: reconcile (evict stale) then write (additive) -----------
    # The set of binary/dataset sidecar idents (``"<box>/<rel>"``) materialized
    # THIS run — the provenance the reconcile evicts against (see below).
    current_shipped = {ident for _d, _p, ident in sidecar_writes}

    # Load the prior owned-sidecars provenance record BEFORE reconciling. A
    # transient read fault aborts ALL eviction this run (never deletes valid
    # cached plaintext), mirroring the per-unit ``transient_io_error`` guard; a
    # MISSING/corrupt record (first sync / legacy cache) yields an empty prior
    # set — start recording, evict nothing this run.
    prior_owned: set[str] = set()
    if not transient_io_error:
        prior_owned, record_ok = _load_owned_sidecars(zettel_dir)
        if not record_ok:
            transient_io_error = True

    # Eviction is gated on a clean pass so a transient fault never deletes valid
    # cached plaintext. The cache is EXCLUSIVELY produced by this function, so a
    # stale unit can be evicted by filename (like memory's ``entries/``).
    if not transient_io_error:
        # Cache-listing during reconcile — ``_note_graph_dirs`` plus the per-box
        # ``*.md`` glob and the ``_projects``/``_citations``/``_reviews``/
        # ``_organizations`` enumerations that compute what's on disk to evict —
        # can hit a transient ``OSError``. Mirror the per-unit/record transient
        # guard: on such a fault skip ALL eviction AND the record persist this
        # run, leaving the old ``.owned-sidecars.json`` and cached files intact
        # (revocation still works on any later clean pass). ``_unlink_quietly``
        # already swallows unlink faults, so only a listing fault trips this.
        try:
            boxes_to_reconcile = (
                set(seen_notes) | seen_meta_boxes | set(_note_graph_dirs(zettel_dir))
            )
            for box in boxes_to_reconcile:
                try:
                    box_dir = safe_join(zettel_dir, box)
                except ValueError:
                    continue
                if not box_dir.is_dir():
                    continue
                keep = seen_notes.get(box, set())
                for existing in box_dir.glob("*.md"):
                    if existing.name.startswith("_") or existing.name in keep:
                        continue
                    _unlink_quietly(existing)
                # A box's _meta.yaml is stale when this box has no shared meta now.
                if box not in seen_meta_boxes:
                    _unlink_quietly(box_dir / "_meta.yaml")
            _reconcile_dir(zettel_dir / "_projects", seen_projects, KIND_PROJECT)
            _reconcile_dir(zettel_dir / "_citations", seen_citations, KIND_CITATION)
            # ``_reviews`` holds three OWNED shapes (manifest .yaml, .tables.json,
            # .outlines.json); each is reconciled against its own seen-filename set
            # with a kind-specific owned-check, so a stale re-scoped review/table/
            # outline is evicted while a never-shipped ``.md`` (foreign) is preserved.
            reviews_dir = zettel_dir / "_reviews"
            if reviews_dir.is_dir():
                for existing in reviews_dir.glob("*.yaml"):
                    if existing.name not in seen_reviews and _is_owned_unit(existing, KIND_REVIEW):
                        _unlink_quietly(existing)
                for existing in reviews_dir.glob("*.tables.json"):
                    if existing.name not in seen_tables and _is_owned_unit(existing, KIND_TABLE):
                        _unlink_quietly(existing)
                for existing in reviews_dir.glob("*.outlines.json"):
                    if existing.name not in seen_outlines and _is_owned_unit(existing, KIND_OUTLINE):
                        _unlink_quietly(existing)
            _reconcile_orgs(zettel_dir / "_organizations", seen_org_paths)
        except OSError as exc:
            logger.warning(
                "cache listing failed during sidecar reconcile (transient I/O): "
                "%s; skipping eviction and record persist this run to preserve "
                "valid cached plaintext (revocation retries on a later clean pass)",
                exc,
            )
            transient_io_error = True

    # Evict binary/dataset sidecars LAST — only after EVERY directory
    # enumeration above (per-box ``*.md``, ``_projects``/``_citations``/
    # ``_reviews``/``_organizations``) has succeeded. Any transient listing
    # ``OSError`` there sets ``transient_io_error`` and this block is skipped, so
    # a fault evicts ZERO sidecar blobs (all-or-nothing) — leaving cache + record
    # intact for a later clean retry — instead of leaving the record old while the
    # bytes are already gone. Eviction is by PROVENANCE, not shape: EXACTLY the
    # paths we materialized before (``prior_owned``) that the new bundle no longer
    # ships (``prior_owned - current_shipped``). A path we never materialized — a
    # DVC-pulled data file, a foreign drop-in, a user's own file — is NEVER in
    # ``prior_owned`` and so is always preserved, regardless of its suffix. Each
    # stale ident is re-jailed before unlink.
    if not transient_io_error:
        for stale_ident in prior_owned - current_shipped:
            try:
                stale_path = safe_join(zettel_dir, *stale_ident.split("/"))
            except ValueError:
                continue
            _unlink_quietly(stale_path)

    for box, dest, payload, _ident in note_writes:
        atomic_write(dest, payload)
    for box, dest, payload in meta_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in project_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in citation_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in review_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in table_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in outline_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in org_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in sidecar_writes:
        atomic_write(dest, payload)

    # Persist this run's owned-sidecar provenance record (atomic) so the NEXT
    # decrypt evicts exactly what this run stops shipping. Only on a clean pass —
    # a transient fault leaves the prior record intact so revocation still works
    # on a later clean pass. An empty set removes the record rather than writing
    # ``[]`` (avoids materializing a spurious empty ``.zettelkasten`` dir on a
    # sidecar-free sync).
    if not transient_io_error:
        _persist_owned_sidecars(zettel_dir, current_shipped)

    return DecryptResult(
        output_dir=zettel_dir,
        note_keys=tuple(k for _b, _d, _p, k in note_writes),
        source_boxes=tuple(b for b, _d, _p in meta_writes),
        project_names=tuple(n for _d, _p, n in project_writes),
        citation_ids=tuple(c for _d, _p, c in citation_writes),
        review_names=tuple(n for _d, _p, n in review_writes),
        org_ids=tuple(n for _d, _p, n in org_writes),
        table_reviews=tuple(n for _d, _p, n in table_writes),
        outline_reviews=tuple(n for _d, _p, n in outline_writes),
        sidecar_paths=tuple(p for _d, _pl, p in sidecar_writes),
        total_units=len(manifest.units),
    )

open_bundle 

open_bundle(bundle_dir: Path | str, out_root: Path | str, *, identity: LocalIdentity | Any | None = None, private_key: str | None = None) -> DecryptResult

Alias for :func:decrypt_bundle.

Source code in zettelkasten/sharing/bundle.py 
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
def open_bundle(
    bundle_dir: Path | str,
    out_root: Path | str,
    *,
    identity: LocalIdentity | Any | None = None,
    private_key: str | None = None,
) -> DecryptResult:
    """Alias for :func:`decrypt_bundle`."""
    return decrypt_bundle(
        bundle_dir, out_root, identity=identity, private_key=private_key
    )

read_manifest 

read_manifest(bundle_dir: Path | str) -> BundleManifest

Read and parse a bundle's cleartext manifest.json.

Source code in zettelkasten/sharing/bundle.py 
319
320
321
322
323
def read_manifest(bundle_dir: Path | str) -> BundleManifest:
    """Read and parse a bundle's cleartext ``manifest.json``."""
    path = Path(bundle_dir) / MANIFEST_NAME
    data = json.loads(path.read_text(encoding="utf-8"))
    return BundleManifest.from_dict(data)

project_for_caller 

project_for_caller(caller_id: str, *, registry: IdentityRegistry, out_root: Path | str, store_root: Path | str | None = None, sharing: SharingConfig | None = None, allowed_transports: set[str] | None = None) -> ProjectionResult

Materialize the PLAINTEXT units caller_id may open under out_root.

Writes a .zettelkasten-shaped plaintext store into <out_root>/.zettelkasten/ — the SAME layout :func:~zettelkasten.sharing.bundle.decrypt_bundle produces — containing only the units caller_id is entitled to, with references scrubbed exactly as :func:~zettelkasten.sharing.bundle.build_bundle scrubs them.

store_root defaults to the local .zettelkasten store and sharing to the policy parsed from <store_root>/config.yaml (both resolved inside the shared :func:~zettelkasten.sharing.bundle._collect_pending).

allowed_transports restricts the projection to units of a given transport tier (design decision D3). The hosted-federation broker passes {"hosted"} so it serves ONLY hosted units — structurally, a unit whose effective transport is offline can never be projected to a broker caller, so an offline-only store is never served over the network. The default None applies no transport filter (every entitled unit is projected), preserving the pre-transport-tier behaviour and keeping the projection byte-identical to an offline build_bundle + decrypt_bundle when the store has no transport annotations.

This function OWNS <out_root>/.zettelkasten and RECONCILES it on every call, exactly like :func:~zettelkasten.sharing.bundle.decrypt_bundle: a re-projection (e.g. after a grant change) surgically evicts only the units this projection previously produced that the caller is no longer entitled to, while PRESERVING foreign/unowned files and honoring the same transient-I/O guard (a read glitch skips eviction rather than destroying a valid cache). The result matches a fresh decrypt of the caller's CURRENT grants even when projecting into a previously-populated store.

Source code in zettelkasten/sharing/projection.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
def project_for_caller(
    caller_id: str,
    *,
    registry: IdentityRegistry,
    out_root: Path | str,
    store_root: Path | str | None = None,
    sharing: SharingConfig | None = None,
    allowed_transports: set[str] | None = None,
) -> ProjectionResult:
    """Materialize the PLAINTEXT units ``caller_id`` may open under ``out_root``.

    Writes a ``.zettelkasten``-shaped plaintext store into
    ``<out_root>/.zettelkasten/`` — the SAME layout
    :func:`~zettelkasten.sharing.bundle.decrypt_bundle` produces — containing only
    the units ``caller_id`` is entitled to, with references scrubbed exactly as
    :func:`~zettelkasten.sharing.bundle.build_bundle` scrubs them.

    ``store_root`` defaults to the local ``.zettelkasten`` store and ``sharing``
    to the policy parsed from ``<store_root>/config.yaml`` (both resolved inside
    the shared :func:`~zettelkasten.sharing.bundle._collect_pending`).

    ``allowed_transports`` restricts the projection to units of a given transport
    tier (design decision D3). The hosted-federation broker passes ``{"hosted"}``
    so it serves ONLY hosted units — structurally, a unit whose effective
    transport is ``offline`` can never be projected to a broker caller, so an
    offline-only store is never served over the network. The default ``None``
    applies no transport filter (every entitled unit is projected), preserving
    the pre-transport-tier behaviour and keeping the projection byte-identical to
    an offline ``build_bundle`` + ``decrypt_bundle`` when the store has no
    transport annotations.

    This function OWNS ``<out_root>/.zettelkasten`` and RECONCILES it on every
    call, exactly like :func:`~zettelkasten.sharing.bundle.decrypt_bundle`: a
    re-projection (e.g. after a grant change) surgically evicts only the units this
    projection previously produced that the caller is no longer entitled to, while
    PRESERVING foreign/unowned files and honoring the same transient-I/O guard
    (a read glitch skips eviction rather than destroying a valid cache). The result
    matches a fresh decrypt of the caller's CURRENT grants even when projecting
    into a previously-populated store.
    """
    out_root = Path(out_root)
    zettel_dir = out_root / ".zettelkasten"

    # Enumerate + recipient-resolve + filter — IDENTICAL to the bundle path
    # (save the transport tier: the broker passes ``{"hosted"}`` to serve only
    # hosted units; ``None`` projects every entitled unit, unchanged behaviour).
    pending, bundled_recip, _skipped = _collect_pending(
        store_root, sharing, registry, allowed_transports
    )

    # A caller opens exactly the units its key can DECRYPT. build_bundle encrypts
    # each unit to a set of age PUBLIC KEYS, so the caller opens a unit iff its own
    # resolvable public key is among that unit's ``pubkeys`` — NOT merely iff its
    # registry id appears in the recipient set. Selecting by public key (rather
    # than id) keeps projection == decrypt even when two distinct registry ids
    # resolve to the SAME age key: an id-based check could withhold a unit the
    # caller's key CAN decrypt (or project one it cannot), diverging from the
    # decrypt. An unregistered caller (no resolvable key) opens nothing — parity
    # with a decrypt that opens no unit.
    caller_pubkey = registry.public_key_for(caller_id)

    # ---- Phase 1: select this caller's units, BUFFERING writes -------------
    # Mirror decrypt_bundle: buffer ``(dest, payload, ident)`` per kind and record
    # the destination names/paths we will own (``seen_*``), so Phase 2 can evict
    # ONLY the stale units this projection previously produced — never foreign
    # drop-ins. Writing happens after reconcile so eviction+write is one pass.
    note_writes: list[tuple[str, Path, bytes, str]] = []  # (box, dest, payload, ident)
    meta_writes: list[tuple[str, Path, bytes, str]] = []  # (box, dest, payload, ident)
    project_writes: list[tuple[Path, bytes, str]] = []
    citation_writes: list[tuple[Path, bytes, str]] = []
    review_writes: list[tuple[Path, bytes, str]] = []
    org_writes: list[tuple[Path, bytes, str]] = []
    table_writes: list[tuple[Path, bytes, str]] = []
    outline_writes: list[tuple[Path, bytes, str]] = []
    sidecar_writes: list[tuple[Path, bytes, str]] = []
    seen_notes: dict[str, set[str]] = {}   # box -> {filename}
    seen_meta_boxes: set[str] = set()
    seen_projects: set[str] = set()
    seen_citations: set[str] = set()
    seen_reviews: set[str] = set()      # <name>.yaml
    seen_tables: set[str] = set()       # <name>.tables.json
    seen_outlines: set[str] = set()     # <name>.outlines.json
    seen_org_paths: set[Path] = set()   # resolved <id>.json dest paths

    if caller_pubkey is not None:
        for _kind, _unit_key, header, payload, _recips, pubkeys in _scrub_pending(
            pending, bundled_recip
        ):
            # A unit is decryptable by this caller iff their resolved public key is
            # among the unit's recipient keys (the SAME ``pubkeys`` build_bundle
            # encrypts to) — key-based, matching decrypt_bundle exactly. Using the
            # SUPERSET-scrubbed ``payload`` (not a caller-relative re-scrub) is what
            # makes the plaintext byte-identical to what the caller would decrypt
            # from the bundle.
            if caller_pubkey not in pubkeys:
                continue
            dest, kind, ident = _unit_dest(header, zettel_dir)
            if dest is None:
                # A malformed/unsafe placement is skipped exactly as decrypt does.
                logger.warning("skipping unit with no usable placement: %r", header)
                continue
            if kind == KIND_NOTE:
                box = str(header.get("box"))
                note_writes.append((box, dest, payload, ident))
                seen_notes.setdefault(box, set()).add(dest.name)
            elif kind == KIND_SOURCE_META:
                box = str(header.get("box"))
                meta_writes.append((box, dest, payload, ident))
                seen_meta_boxes.add(box)
            elif kind == KIND_PROJECT:
                project_writes.append((dest, payload, ident))
                seen_projects.add(dest.name)
            elif kind == KIND_CITATION:
                citation_writes.append((dest, payload, ident))
                seen_citations.add(dest.name)
            elif kind == KIND_REVIEW:
                review_writes.append((dest, payload, ident))
                seen_reviews.add(dest.name)
            elif kind == KIND_TABLE:
                table_writes.append((dest, payload, ident))
                seen_tables.add(dest.name)
            elif kind == KIND_OUTLINE:
                outline_writes.append((dest, payload, ident))
                seen_outlines.add(dest.name)
            elif kind == KIND_ORG:
                org_writes.append((dest, payload, ident))
                seen_org_paths.add(dest.resolve())
            elif kind == KIND_SIDECAR:
                sidecar_writes.append((dest, payload, ident))

    # ---- Phase 2: reconcile (evict stale) then write (additive) -----------
    # This is decrypt_bundle's Phase 2 verbatim (minus the manifest/ciphertext
    # bits it does not apply here), so a projection and a decrypt reconcile a
    # populated store IDENTICALLY. The scoped store is EXCLUSIVELY produced by this
    # projection, so notes are evicted by filename (like memory's ``entries/``),
    # while the shared ``_projects``/``_citations``/``_reviews``/``_organizations``
    # dirs evict only well-formed owned units (foreign drop-ins preserved). Binary/
    # dataset sidecars carry no self-identifying frontmatter, so their ownership is
    # tracked by the ``.owned-sidecars.json`` provenance record.
    current_shipped = {ident for _d, _p, ident in sidecar_writes}

    # Load the prior owned-sidecars provenance record BEFORE reconciling. A
    # transient read fault aborts ALL eviction this run (never deletes valid cached
    # plaintext); a MISSING/corrupt record yields an empty prior set (evict nothing
    # this run, just start recording).
    transient_io_error = False
    prior_owned, record_ok = _load_owned_sidecars(zettel_dir)
    if not record_ok:
        transient_io_error = True

    # Eviction is gated on a clean pass so a transient listing fault never deletes
    # valid cached plaintext (revocation still works on any later clean pass).
    if not transient_io_error:
        try:
            boxes_to_reconcile = (
                set(seen_notes) | seen_meta_boxes | set(_note_graph_dirs(zettel_dir))
            )
            for box in boxes_to_reconcile:
                try:
                    box_dir = safe_join(zettel_dir, box)
                except ValueError:
                    continue
                if not box_dir.is_dir():
                    continue
                keep = seen_notes.get(box, set())
                for existing in box_dir.glob("*.md"):
                    if existing.name.startswith("_") or existing.name in keep:
                        continue
                    _unlink_quietly(existing)
                # A box's _meta.yaml is stale when this box has no shared meta now.
                if box not in seen_meta_boxes:
                    _unlink_quietly(box_dir / "_meta.yaml")
            _reconcile_dir(zettel_dir / "_projects", seen_projects, KIND_PROJECT)
            _reconcile_dir(zettel_dir / "_citations", seen_citations, KIND_CITATION)
            reviews_dir = zettel_dir / "_reviews"
            if reviews_dir.is_dir():
                for existing in reviews_dir.glob("*.yaml"):
                    if existing.name not in seen_reviews and _is_owned_unit(existing, KIND_REVIEW):
                        _unlink_quietly(existing)
                for existing in reviews_dir.glob("*.tables.json"):
                    if existing.name not in seen_tables and _is_owned_unit(existing, KIND_TABLE):
                        _unlink_quietly(existing)
                for existing in reviews_dir.glob("*.outlines.json"):
                    if existing.name not in seen_outlines and _is_owned_unit(existing, KIND_OUTLINE):
                        _unlink_quietly(existing)
            _reconcile_orgs(zettel_dir / "_organizations", seen_org_paths)
        except OSError as exc:
            logger.warning(
                "cache listing failed during projection reconcile (transient I/O): "
                "%s; skipping eviction and record persist this run to preserve valid "
                "cached plaintext (revocation retries on a later clean pass)", exc
            )
            transient_io_error = True

    # Evict binary/dataset sidecars by PROVENANCE (not shape): exactly the paths we
    # materialized before that the caller no longer receives. A path we never
    # produced (a DVC-pulled data file, a foreign drop-in) is never in
    # ``prior_owned`` and so is always preserved. All-or-nothing under a fault.
    if not transient_io_error:
        for stale_ident in prior_owned - current_shipped:
            try:
                stale_path = safe_join(zettel_dir, *stale_ident.split("/"))
            except ValueError:
                continue
            _unlink_quietly(stale_path)

    # Materialize the caller's current units (additive). ``atomic_write`` creates
    # parent dirs, so an empty projection writes nothing (no dir churn).
    for _box, dest, payload, _ident in note_writes:
        atomic_write(dest, payload)
    for _box, dest, payload, _ident in meta_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in project_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in citation_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in review_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in table_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in outline_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in org_writes:
        atomic_write(dest, payload)
    for dest, payload, _ident in sidecar_writes:
        atomic_write(dest, payload)

    # Persist this run's owned-sidecar provenance so the NEXT projection evicts
    # exactly what this run stops shipping. Only on a clean pass; an empty set
    # removes the record (no spurious empty cache dir).
    if not transient_io_error:
        _persist_owned_sidecars(zettel_dir, current_shipped)

    note_keys = [ident for _b, _d, _p, ident in note_writes]
    source_boxes = [ident for _b, _d, _p, ident in meta_writes]
    project_names = [ident for _d, _p, ident in project_writes]
    citation_ids = [ident for _d, _p, ident in citation_writes]
    review_names = [ident for _d, _p, ident in review_writes]
    org_ids = [ident for _d, _p, ident in org_writes]
    table_reviews = [ident for _d, _p, ident in table_writes]
    outline_reviews = [ident for _d, _p, ident in outline_writes]
    sidecar_paths = [ident for _d, _p, ident in sidecar_writes]

    total = (
        len(note_keys) + len(source_boxes) + len(project_names) + len(citation_ids)
        + len(review_names) + len(org_ids) + len(table_reviews) + len(outline_reviews)
        + len(sidecar_paths)
    )

    return ProjectionResult(
        caller_id=caller_id,
        output_dir=zettel_dir,
        note_keys=tuple(note_keys),
        source_boxes=tuple(source_boxes),
        project_names=tuple(project_names),
        citation_ids=tuple(citation_ids),
        review_names=tuple(review_names),
        org_ids=tuple(org_ids),
        table_reviews=tuple(table_reviews),
        outline_reviews=tuple(outline_reviews),
        sidecar_paths=tuple(sidecar_paths),
        note_unit_count=len(note_keys),
        source_unit_count=len(source_boxes),
        project_unit_count=len(project_names),
        citation_unit_count=len(citation_ids),
        review_unit_count=len(review_names),
        org_unit_count=len(org_ids),
        table_unit_count=len(table_reviews),
        outline_unit_count=len(outline_reviews),
        sidecar_unit_count=len(sidecar_paths),
        total_units=total,
        transient_io_error=transient_io_error,
    )

citation_recipients 

citation_recipients(sharing: SharingConfig, note_recipients_map: Mapping[str, set[str]], note_citation_refs: Mapping[str, Iterable[str]]) -> dict[str, set[str]]

Map each shareable citation (citation:<id>) to its recipient set.

A citation is a leaf bibliographic entity referenced BY notes (via a links[].graph == "_citations" edge). Its recipients are the UNION over every SHARED note that references it of that note's recipient set — so every recipient of a note can also open the citation the note points at (the superset property that keeps the citation link from ever dangling). Any direct grants['citation:<id>'] recipients are added on top. Citations resolving to no recipients are omitted.

Source code in zettelkasten/sharing/config.py 
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
def citation_recipients(
    sharing: SharingConfig,
    note_recipients_map: Mapping[str, set[str]],
    note_citation_refs: Mapping[str, Iterable[str]],
) -> dict[str, set[str]]:
    """Map each shareable citation (``citation:<id>``) to its recipient set.

    A citation is a leaf bibliographic entity referenced BY notes (via a
    ``links[].graph == "_citations"`` edge). Its recipients are the UNION over
    every SHARED note that references it of that note's recipient set — so every
    recipient of a note can also open the citation the note points at (the
    superset property that keeps the citation link from ever dangling). Any
    direct ``grants['citation:<id>']`` recipients are added on top. Citations
    resolving to no recipients are omitted.
    """
    out: dict[str, set[str]] = {}
    for note_key, cids in note_citation_refs.items():
        note_recips = note_recipients_map.get(note_key)
        if not note_recips:
            continue
        for cid in cids:
            out.setdefault(f"citation:{cid}", set()).update(note_recips)

    for key, recips in sharing.grants.items():
        if key.startswith("citation:"):
            out.setdefault(key, set()).update(recips)

    return out

note_recipients 

note_recipients(sharing: SharingConfig, projects: Iterable[Mapping[str, Any]], notes_by_box: Mapping[str, Iterable[str]], firm_members: Iterable[str] = ()) -> dict[str, set[str]]

Map each shareable note (<box>/<note_id>) to its recipient set.

The recipients of a note are the UNION of:

  • every project closure that includes it — a note is in project P's closure when it lives in one of P's sources[] boxes, or when its id is listed in P's cross[] (a _cross synthesis note); such notes get P's recipients;
  • a direct source-box scope/grant on the note's box; and
  • a direct per-note grant keyed on <box>/<note_id> (cross/<id> alias accepted).

Notes resolving to no recipients are omitted.

Source code in zettelkasten/sharing/config.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def note_recipients(
    sharing: SharingConfig,
    projects: Iterable[Mapping[str, Any]],
    notes_by_box: Mapping[str, Iterable[str]],
    firm_members: Iterable[str] = (),
) -> dict[str, set[str]]:
    """Map each shareable note (``<box>/<note_id>``) to its recipient set.

    The recipients of a note are the UNION of:

    * every project closure that includes it — a note is in project P's closure
      when it lives in one of P's ``sources[]`` boxes, or when its id is listed
      in P's ``cross[]`` (a ``_cross`` synthesis note); such notes get P's
      recipients;
    * a direct source-box scope/grant on the note's box; and
    * a direct per-note grant keyed on ``<box>/<note_id>`` (``cross/<id>`` alias
      accepted).

    Notes resolving to no recipients are omitted.
    """
    firm = set(firm_members)
    out: dict[str, set[str]] = {}

    def add(key: str, recips: set[str]) -> None:
        if recips:
            out.setdefault(key, set()).update(recips)

    proj_recip = project_recipients(sharing, [n for n, _, _ in _iter_projects(projects)], firm)

    # Project closures: every note in each source box + each claimed cross note.
    for name, sources, cross in _iter_projects(projects):
        recips = proj_recip.get(name)
        if not recips:
            continue
        for box in sources:
            for nid in notes_by_box.get(box, ()):  # type: ignore[union-attr]
                add(f"{box}/{nid}", recips)
        for cid in cross:
            add(f"{CROSS_BOX}/{cid}", recips)

    # Direct source-box scope/grant applies to every note in the box.
    for box, nids in notes_by_box.items():
        box_recips = set(sharing.grants.get(box, ()))
        box_recips |= _scope_recipients(sharing.scopes.get(box), firm)
        if box_recips:
            for nid in nids:
                add(f"{box}/{nid}", box_recips)

    # Direct per-note grants (keys containing '/', not a citation key).
    for key, recips in sharing.grants.items():
        if "/" not in key or key.startswith("citation:"):
            continue
        add(normalize_note_key(key), set(recips))

    return out

org_recipients 

org_recipients(sharing: SharingConfig, organizations: Iterable[Mapping[str, Any]], note_recipients_map: Mapping[str, set[str]], source_recipients_map: Mapping[str, set[str]], project_recipients_map: Mapping[str, set[str]]) -> dict[str, set[str]]

Map each organization id to its recipient set.

An organization is owned by exactly one project OR one graph (see :data:zettelkasten.organizations.OWNER_TYPES). Its recipients are the owner's recipients — the project's recipients, or (for a graph owner) everyone who can see that box — plus any direct grants['org:<id>']. Organizations resolving to no recipients are omitted.

Source code in zettelkasten/sharing/config.py 
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
def org_recipients(
    sharing: SharingConfig,
    organizations: Iterable[Mapping[str, Any]],
    note_recipients_map: Mapping[str, set[str]],
    source_recipients_map: Mapping[str, set[str]],
    project_recipients_map: Mapping[str, set[str]],
) -> dict[str, set[str]]:
    """Map each organization ``id`` to its recipient set.

    An organization is owned by exactly one project OR one graph (see
    :data:`zettelkasten.organizations.OWNER_TYPES`). Its recipients are the
    owner's recipients — the project's recipients, or (for a ``graph`` owner)
    everyone who can see that box — plus any direct ``grants['org:<id>']``.
    Organizations resolving to no recipients are omitted.
    """
    out: dict[str, set[str]] = {}
    for org in organizations:
        if not isinstance(org, Mapping):
            continue
        oid = str(org.get("id") or "").strip()
        if not oid:
            continue
        owner = org.get("owner") if isinstance(org.get("owner"), Mapping) else {}
        owner_type = str(owner.get("type") or "").strip()
        owner_name = str(owner.get("name") or "").strip()
        if owner_type == "project" and owner_name:
            recips = set(project_recipients_map.get(owner_name, set()))
        elif owner_type == "graph" and owner_name:
            recips = _box_recipients(note_recipients_map, source_recipients_map, owner_name)
        else:
            recips = set()
        recips |= set(sharing.grants.get(f"org:{oid}", ()))
        if recips:
            out[oid] = recips
    return out

parse_sharing_config 

parse_sharing_config(config: Mapping[str, Any] | None) -> SharingConfig

Parse the sharing and grants blocks of a loaded config dict.

config is the plain dict read from .zettelkasten/config.yaml. Missing/empty/malformed blocks yield an empty :class:SharingConfig (nothing shared) — this never raises on a partially-formed config; unusable items are skipped. The existing rerank/federated_repos blocks are ignored here (left untouched for their own readers).

Source code in zettelkasten/sharing/config.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def parse_sharing_config(config: Mapping[str, Any] | None) -> SharingConfig:
    """Parse the ``sharing`` and ``grants`` blocks of a loaded config dict.

    ``config`` is the plain dict read from ``.zettelkasten/config.yaml``.
    Missing/empty/malformed blocks yield an empty :class:`SharingConfig`
    (nothing shared) — this never raises on a partially-formed config; unusable
    items are skipped. The existing ``rerank``/``federated_repos`` blocks are
    ignored here (left untouched for their own readers).
    """
    if not isinstance(config, Mapping):
        return SharingConfig()

    scopes: dict[str, Scope] = {}
    transports: dict[str, str] = {}
    sharing_block = config.get("sharing")
    if isinstance(sharing_block, Mapping):
        raw_scopes = sharing_block.get("scopes")
        if isinstance(raw_scopes, Mapping):
            for node_id, raw in raw_scopes.items():
                nid = str(node_id).strip()
                if nid:
                    scopes[nid] = _parse_scope(raw)
        # Per-unit transport tier (``sharing.transports``), keyed like scopes.
        # Unlike scopes (fail-closed to ``private`` on a typo), an unrecognised
        # transport is REJECTED with a clear error — the network tier is
        # deny-by-default, so a mistyped value must never be silently swallowed.
        raw_transports = sharing_block.get("transports")
        if isinstance(raw_transports, Mapping):
            for node_id, raw in raw_transports.items():
                nid = str(node_id).strip()
                if not nid:
                    continue
                transports[nid] = _parse_transport(raw, nid)

    grants: dict[str, tuple[str, ...]] = {}
    raw_grants = config.get("grants")
    if isinstance(raw_grants, Mapping):
        for node_id, raw in raw_grants.items():
            nid = str(node_id).strip()
            recips = _parse_recipient_list(raw)
            if nid and recips:
                grants[nid] = recips

    return SharingConfig(scopes=scopes, grants=grants, transports=transports)

project_recipients 

project_recipients(sharing: SharingConfig, project_names: Iterable[str], firm_members: Iterable[str] = ()) -> dict[str, set[str]]

Map each project name to the concrete set of recipient ids.

A project's recipients are its scopes[name] (firm expanded to firm_members) UNIONED with any grants[name]. Projects resolving to no recipients are omitted (the caller uses this to drop them from the bundle).

Source code in zettelkasten/sharing/config.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def project_recipients(
    sharing: SharingConfig,
    project_names: Iterable[str],
    firm_members: Iterable[str] = (),
) -> dict[str, set[str]]:
    """Map each project name to the concrete set of recipient ids.

    A project's recipients are its ``scopes[name]`` (firm expanded to
    ``firm_members``) UNIONED with any ``grants[name]``. Projects resolving to no
    recipients are omitted (the caller uses this to drop them from the bundle).
    """
    firm = set(firm_members)
    out: dict[str, set[str]] = {}
    for name in project_names:
        recips = set(sharing.grants.get(name, ()))
        recips |= _scope_recipients(sharing.scopes.get(name), firm)
        if recips:
            out[name] = recips
    return out

review_recipients 

review_recipients(sharing: SharingConfig, reviews: Iterable[Mapping[str, Any]], note_recipients_map: Mapping[str, set[str]], source_recipients_map: Mapping[str, set[str]], project_recipients_map: Mapping[str, set[str]]) -> dict[str, set[str]]

Map each review name to its recipient set (also used for its tables/outlines).

A review is shared to the recipients who can see its SCOPE, plus any direct grants['review:<name>']:

  • graph set -> box-scoped: everyone who can see box G (see :func:_box_recipients);
  • else project set -> project-scoped: the project's recipients;
  • empty/empty -> full-corpus: everyone who can see any shared unit.

A direct grants['review:<name>'] widens the audience. Reviews resolving to no recipients are omitted. The returned map is reused verbatim for a review's KIND_TABLE grid file and KIND_OUTLINE sidecar (both keyed by the review name), so a table/outline can never reach a recipient the review itself could not.

Source code in zettelkasten/sharing/config.py 
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
def review_recipients(
    sharing: SharingConfig,
    reviews: Iterable[Mapping[str, Any]],
    note_recipients_map: Mapping[str, set[str]],
    source_recipients_map: Mapping[str, set[str]],
    project_recipients_map: Mapping[str, set[str]],
) -> dict[str, set[str]]:
    """Map each review ``name`` to its recipient set (also used for its tables/outlines).

    A review is shared to the recipients who can see its SCOPE, plus any direct
    ``grants['review:<name>']``:

    * ``graph`` set  -> box-scoped: everyone who can see box ``G`` (see
      :func:`_box_recipients`);
    * else ``project`` set -> project-scoped: the project's recipients;
    * empty/empty    -> full-corpus: everyone who can see any shared unit.

    A direct ``grants['review:<name>']`` widens the audience. Reviews resolving
    to no recipients are omitted. The returned map is reused verbatim for a
    review's ``KIND_TABLE`` grid file and ``KIND_OUTLINE`` sidecar (both keyed by
    the review name), so a table/outline can never reach a recipient the review
    itself could not.
    """
    out: dict[str, set[str]] = {}
    for manifest in reviews:
        if not isinstance(manifest, Mapping):
            continue
        name = str(manifest.get("name") or "").strip()
        if not name:
            continue
        graph = str(manifest.get("graph") or "").strip()
        project = str(manifest.get("project") or "").strip()
        if graph:
            recips = _box_recipients(note_recipients_map, source_recipients_map, graph)
        elif project:
            recips = set(project_recipients_map.get(project, set()))
        else:
            recips = _full_corpus_recipients(
                note_recipients_map, source_recipients_map, project_recipients_map
            )
        recips = set(recips)
        recips |= set(sharing.grants.get(f"review:{name}", ()))
        if recips:
            out[name] = recips
    return out

source_recipients 

source_recipients(sharing: SharingConfig, projects: Iterable[Mapping[str, Any]], boxes: Iterable[str], firm_members: Iterable[str] = ()) -> dict[str, set[str]]

Map each shareable source box to its _meta.yaml recipient set.

A box's meta is shared to the UNION of every project that lists the box in its sources[] (getting that project's recipients) plus any direct box-level scope/grant. Boxes resolving to no recipients are omitted.

Source code in zettelkasten/sharing/config.py 
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
def source_recipients(
    sharing: SharingConfig,
    projects: Iterable[Mapping[str, Any]],
    boxes: Iterable[str],
    firm_members: Iterable[str] = (),
) -> dict[str, set[str]]:
    """Map each shareable source box to its ``_meta.yaml`` recipient set.

    A box's meta is shared to the UNION of every project that lists the box in
    its ``sources[]`` (getting that project's recipients) plus any direct
    box-level scope/grant. Boxes resolving to no recipients are omitted.
    """
    firm = set(firm_members)
    box_set = set(boxes)
    out: dict[str, set[str]] = {}

    def add(box: str, recips: set[str]) -> None:
        if recips and box in box_set:
            out.setdefault(box, set()).update(recips)

    proj_recip = project_recipients(sharing, [n for n, _, _ in _iter_projects(projects)], firm)
    for name, sources, _cross in _iter_projects(projects):
        recips = proj_recip.get(name)
        if not recips:
            continue
        for box in sources:
            add(box, recips)

    for box in box_set:
        box_recips = set(sharing.grants.get(box, ()))
        box_recips |= _scope_recipients(sharing.scopes.get(box), firm)
        add(box, box_recips)

    return out