Skip to content

memory.sharing

memory.sharing

Encryption-first firm-wide sharing: config, identity, crypto, and bundles.

This package is the foundation for sharing a filtered, per-recipient-encrypted slice of a memory tree with other machines/teams. It is a library only — the publish/sync CLI is wired separately on top of this stable API.

The three layers:

  • Policy (:mod:~memory.sharing.config) — parse the owner-authored sharing/grants config and resolve which entries are shareable to whom.
  • Identity (:mod:~memory.sharing.identity) — keypair generation, the local ~/.angelo/identity.yaml (private key, never committed), and the public-key identity registry.
  • Crypto + bundle (:mod:~memory.sharing.crypto, :mod:~memory.sharing.bundle) — age encrypt/decrypt and the on-disk bundle format (filter -> encrypt -> package, then decrypt -> materialize a .memory tree that :func:memory.federation.read_repo can read unchanged).

Typical producer flow::

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

Typical consumer flow::

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

BuildResult dataclass

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

Source code in memory/sharing/bundle.py 
267
268
269
270
271
272
273
274
275
276
277
278
279
@dataclass(frozen=True)
class BuildResult:
    """Summary of a :func:`build_bundle` run (returned in memory, never written)."""

    manifest: BundleManifest
    output_dir: Path
    entry_unit_count: int
    project_written: bool
    recipient_ids: tuple[str, ...]
    skipped_unregistered_ids: tuple[str, ...]
    document_unit_count: int = 0
    skill_unit_count: int = 0
    session_unit_count: int = 0

BundleManifest dataclass

The cleartext manifest.json at the bundle root.

Source code in memory/sharing/bundle.py 
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
@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 is one of "entry" / "document" / "skill" / "session" (opaque units/<sha256>.age files) or "project" (the project.age stub); path is the unit's location relative to the bundle root. No id or recipient info is stored here — that is the deliberate privacy-first manifest design (see module docstring).

Source code in memory/sharing/bundle.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
@dataclass(frozen=True)
class BundleUnit:
    """A pointer to one encrypted unit inside a bundle.

    ``kind`` is one of ``"entry"`` / ``"document"`` / ``"skill"`` / ``"session"``
    (opaque ``units/<sha256>.age`` files) or ``"project"`` (the ``project.age``
    stub); ``path`` is the unit's location relative to the bundle root. No id or
    recipient info is stored here — that is the deliberate privacy-first manifest
    design (see module docstring).
    """

    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 memory/sharing/bundle.py 
282
283
284
285
286
287
288
289
290
291
292
@dataclass(frozen=True)
class DecryptResult:
    """Summary of a :func:`decrypt_bundle` run."""

    output_memory_dir: Path
    entry_ids: tuple[str, ...]
    project_written: bool
    total_units: int
    document_ids: tuple[str, ...] = ()
    skill_names: tuple[str, ...] = ()
    session_ids: tuple[str, ...] = ()

Scope dataclass

A visibility scope attached to a subtree node.

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

Source code in memory/sharing/config.py 
82
83
84
85
86
87
88
89
90
91
@dataclass(frozen=True)
class Scope:
    """A visibility scope attached to a subtree node.

    ``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 node id (subproject/phase/entry) to its :class:Scope.
  • grants maps a node id to the tuple of recipient ids granted access to that node and its descendants. Grants also carry the out-of-tree kinds: grants['skill:<name>'] grants a specific skill and grants['doc:<id>'] (reserved) a specific document.
  • skills_default is the global default :class:Scope for skills (which live OUTSIDE the entry tree, so they cannot inherit a subtree scope). It defaults to private — skills are shared with nobody unless the owner sets sharing.skills or a skill:<name> grant.
Source code in memory/sharing/config.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 node id (subproject/phase/entry) to its :class:`Scope`.
    * ``grants`` maps a node id to the tuple of recipient ids granted access to
      that node and its descendants. Grants also carry the out-of-tree kinds:
      ``grants['skill:<name>']`` grants a specific skill and
      ``grants['doc:<id>']`` (reserved) a specific document.
    * ``skills_default`` is the global default :class:`Scope` for skills (which
      live OUTSIDE the entry tree, so they cannot inherit a subtree scope). It
      defaults to ``private`` — skills are shared with nobody unless the owner
      sets ``sharing.skills`` or a ``skill:<name>`` grant.
    """

    scopes: Mapping[str, Scope] = field(default_factory=dict)
    grants: Mapping[str, tuple[str, ...]] = field(default_factory=dict)
    skills_default: Scope = field(default_factory=lambda: Scope(SCOPE_PRIVATE))
    # Per-node transport tier keyed exactly like ``scopes`` (by node/entry id).
    # A node without an entry here inherits its nearest ancestor's transport and
    # ultimately defaults to :data:`TRANSPORT_OFFLINE` (deny-by-default for the
    # network tier). 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
            and self.skills_default.kind == SCOPE_PRIVATE
        )

is_empty 

is_empty() -> bool

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

Source code in memory/sharing/config.py 
118
119
120
121
122
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
        and self.skills_default.kind == SCOPE_PRIVATE
    )

ProjectionResult dataclass

Summary of a :func:project_for_caller run.

Mirrors the useful fields of :class:memory.sharing.bundle.DecryptResult (the projection is defined to equal what that caller would decrypt), plus the caller_id it was scoped to. total_units counts the entry/document/ skill/session units in this caller's view (the project stub is reported separately via project_written).

Source code in memory/sharing/projection.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@dataclass(frozen=True)
class ProjectionResult:
    """Summary of a :func:`project_for_caller` run.

    Mirrors the useful fields of :class:`memory.sharing.bundle.DecryptResult`
    (the projection is defined to equal what that caller would decrypt), plus the
    ``caller_id`` it was scoped to. ``total_units`` counts the entry/document/
    skill/session units in this caller's view (the project stub is reported
    separately via ``project_written``).
    """

    output_memory_dir: Path
    entry_ids: tuple[str, ...]
    project_written: bool
    total_units: int
    document_ids: tuple[str, ...] = ()
    skill_names: tuple[str, ...] = ()
    session_ids: tuple[str, ...] = ()
    caller_id: str = ""
    # True iff a TRANSIENT I/O fault during reconcile (dir scan) 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

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)

KeyPair dataclass

A freshly generated identity keypair (encryption + signing).

public_key is an age1... recipient string and private_key an AGE-SECRET-KEY-... string — the age X25519 keypair used for encryption. age has no signing primitive, so an independent Ed25519 signing keypair rides alongside it (URL-safe base64 of the raw key bytes; see :mod:memory.sharing.signing). The signing fields are optional and default to None so legacy callers/keypairs without a signing key keep working. Guard the private keys like passwords.

Source code in memory/sharing/identity.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
@dataclass(frozen=True)
class KeyPair:
    """A freshly generated identity keypair (encryption + signing).

    ``public_key`` is an ``age1...`` recipient string and ``private_key`` an
    ``AGE-SECRET-KEY-...`` string — the ``age`` X25519 keypair used for
    *encryption*. ``age`` has no signing primitive, so an independent Ed25519
    *signing* keypair rides alongside it (URL-safe base64 of the raw key bytes;
    see :mod:`memory.sharing.signing`). The signing fields are optional and
    default to ``None`` so legacy callers/keypairs without a signing key keep
    working. Guard the private keys like passwords.
    """

    public_key: str
    private_key: str
    signing_public_key: str | None = None
    signing_private_key: str | None = None

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

build_bundle 

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

Filter memory entries by grants, encrypt per recipient, and write a bundle.

Pipeline (filtering happens BEFORE any encryption — ungranted/private entries are physically omitted, never encrypted-and-dropped):

  1. Read entries + project from source_memory_dir (default: the local .memory store).
  2. Resolve each entry's recipient id set from sharing (default: parsed from the source config.yaml), expanding a firm scope to the registry members. Entries resolving to no recipients are dropped.
  3. Encrypt each retained entry's on-disk markdown to its recipients' public keys and write units/<digest>.age. Before encryption, cross-link frontmatter fields (related_to/invalidates/invalidated_by/ parent_id) are scrubbed PER UNIT: a reference to id X survives only if X is itself bundled AND every recipient of this unit is also a recipient of X (plus the shared project root is always allowed). This keeps a recipient from ever learning the id/relationship of an entry it was not also granted, and is a byte-level no-op for a granted entry that has no dangling cross-links.
  4. Encrypt a MINIMAL project.md STUB to the UNION of all recipients and write project.age. The stub keeps only structural frontmatter (id/name/status/created_at) with an EMPTY body — enough for read_repo/configured_repos to render the peer, but WITHOUT the project root description/body a bilateral recipient was never granted.
  5. Write the cleartext manifest.json.

This function OWNS output_dir/units, output_dir/project.age and output_dir/manifest.json — the units directory is rebuilt from scratch on every call so stale/revoked units never linger. Other files in output_dir (e.g. a .git directory) are left untouched.

Returns a :class:BuildResult. With no sharing policy configured this writes an empty bundle (no units, no project) — the additive "nothing shared" default.

Source code in memory/sharing/bundle.py 
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
def build_bundle(
    output_dir: Path | str,
    *,
    registry: IdentityRegistry,
    source_memory_dir: Path | str | None = None,
    sharing: SharingConfig | None = None,
    producer_id: str | None = None,
    include_discarded: bool = False,
) -> BuildResult:
    """Filter memory entries by grants, encrypt per recipient, and write a bundle.

    Pipeline (filtering happens BEFORE any encryption — ungranted/private entries
    are physically omitted, never encrypted-and-dropped):

    1. Read entries + project from ``source_memory_dir`` (default: the local
       ``.memory`` store).
    2. Resolve each entry's recipient id set from ``sharing`` (default: parsed
       from the source ``config.yaml``), expanding a ``firm`` scope to the
       ``registry`` members. Entries resolving to no recipients are dropped.
    3. Encrypt each retained entry's on-disk markdown to its recipients' public
       keys and write ``units/<digest>.age``. Before encryption, cross-link
       frontmatter fields (``related_to``/``invalidates``/``invalidated_by``/
       ``parent_id``) are scrubbed PER UNIT: a reference to id X survives only if
       X is itself bundled AND every recipient of this unit is also a recipient
       of X (plus the shared project root is always allowed). This keeps a
       recipient from ever learning the id/relationship of an entry it was not
       also granted, and is a byte-level no-op for a granted entry that has no
       dangling cross-links.
    4. Encrypt a MINIMAL ``project.md`` STUB to the UNION of all recipients and
       write ``project.age``. The stub keeps only structural frontmatter
       (``id``/``name``/``status``/``created_at``) with an EMPTY body — enough for
       ``read_repo``/``configured_repos`` to render the peer, but WITHOUT the
       project root description/body a bilateral recipient was never granted.
    5. Write the cleartext ``manifest.json``.

    This function OWNS ``output_dir/units``, ``output_dir/project.age`` and
    ``output_dir/manifest.json`` — the units directory is rebuilt from scratch on
    every call so stale/revoked units never linger. Other files in ``output_dir``
    (e.g. a ``.git`` directory) are left untouched.

    Returns a :class:`BuildResult`. With no sharing policy configured this writes
    an empty bundle (no units, no project) — the additive "nothing shared"
    default.
    """
    output_dir = Path(output_dir)
    source = Path(source_memory_dir) if source_memory_dir is not None else storage.MEMORY_DIR

    if sharing is None:
        sharing = parse_sharing_config(_read_source_config(source))

    # ---- Steps 1-3 (read + recipient-resolve + per-unit scrub) live in the
    # shared collector, so build_bundle and project_for_caller filter/scrub
    # identically (one source of truth). Everything below is encrypt + write. ----
    # 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.
    collected = _collect_shared_units(
        source=source,
        sharing=sharing,
        registry=registry,
        include_discarded=include_discarded,
        allowed_transports={TRANSPORT_OFFLINE},
    )

    # ---- Prepare output (rebuild units/ from scratch; drop stale project) ----
    units_dir = output_dir / UNITS_DIRNAME
    if units_dir.exists():
        shutil.rmtree(units_dir)
    units_dir.mkdir(parents=True, exist_ok=True)
    stale_project = output_dir / PROJECT_UNIT_NAME
    if stale_project.exists():
        stale_project.unlink()

    units: list[BundleUnit] = []
    # ``all_recipient_pubkeys`` is the union of the pubkeys every ACTUALLY-WRITTEN
    # unit was encrypted to — this is what the project stub is encrypted to below,
    # so any grantee's cache can render the peer. ``all_recipient_ids`` is the
    # registered recipient ids of actually-written units (BuildResult.recipient_ids);
    # accumulated at each write site so an OMITTED unit never contributes a phantom
    # recipient (the collector already drops sessions that scrub to empty).
    all_recipient_pubkeys: set[str] = set()
    all_recipient_ids: set[str] = set()

    # Encrypt each prepared unit per recipient and write it. The collector already
    # returns units in the manifest order entries -> documents -> sessions -> skills.
    for unit in collected.units:
        ciphertext = crypto.encrypt(unit.scrubbed, list(unit.pubkeys))
        filename = _unit_filename(unit.unit_id, unit.kind)
        atomic_write(units_dir / filename, ciphertext)
        units.append(BundleUnit(kind=unit.kind, path=f"{UNITS_DIRNAME}/{filename}"))
        all_recipient_pubkeys.update(unit.pubkeys)
        all_recipient_ids.update(unit.registered_recipients)

    # ---- 4. Encrypt a MINIMAL project stub to the union of ALL recipients ----
    # (across every kind, so any grantee's cache can render). Only the structural
    # frontmatter is published (see _project_stub_bytes); the root description/
    # body is stripped so a bilateral recipient cannot read the full project root
    # it was never granted, while read_repo still has the project.md it
    # structurally requires.
    project_written = False
    if collected.project_stub is not None and all_recipient_pubkeys:
        ciphertext = crypto.encrypt(
            collected.project_stub, sorted(all_recipient_pubkeys)
        )
        atomic_write(output_dir / PROJECT_UNIT_NAME, ciphertext)
        units.append(BundleUnit(kind=KIND_PROJECT, path=PROJECT_UNIT_NAME))
        project_written = True

    # ---- 5. 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))

    return BuildResult(
        manifest=manifest,
        output_dir=output_dir,
        entry_unit_count=sum(1 for u in units if u.kind == KIND_ENTRY),
        project_written=project_written,
        recipient_ids=tuple(sorted(all_recipient_ids)),
        skipped_unregistered_ids=tuple(sorted(collected.skipped_unregistered)),
        document_unit_count=sum(1 for u in units if u.kind == KIND_DOCUMENT),
        skill_unit_count=sum(1 for u in units if u.kind == KIND_SKILL),
        session_unit_count=sum(1 for u in units if u.kind == KIND_SESSION),
    )

decrypt_bundle 

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

Decrypt every unit this key can open and materialize a .memory tree.

For each manifest unit the private key is tried; units that fail to decrypt (i.e. were not encrypted to this key) are silently skipped. Successfully decrypted units are written into a .memory-shaped output_memory_dir (project.md plus entries/<id>.md, documents/<id>.md, sessions/<id>.md and skills/<name>.md) so that :func:memory.federation.read_repo can read the result unchanged.

Each unit's id/name is read from the decrypted frontmatter (it is never stored in the clear) and validated / path-jailed before being used as a filename, so a hostile bundle cannot escape output_memory_dir.

Each cleartext manifest.json unit path is validated BEFORE it is read (rejecting absolute paths, .. traversal, and anything but the opaque units/<sha256>.age / project.age layout — enforced regardless of the manifest's self-declared format) so a malicious publisher cannot use this function to read arbitrary local files. A missing or corrupt manifest is treated as an empty bundle (the cache is still reconciled), never a crash.

The output .memory tree is RECONCILED to the current bundle: stale files (units that dropped out of a newer, re-scoped bundle or are no longer openable by this key) are removed from EACH owned dir (entries/, documents/, skills/, sessions/), and project.md is refreshed or removed to match. Only the .md files in those four dirs and project.md this function owns are touched — unrelated files (e.g. .git) are left alone. A transient I/O fault during the pass SKIPS all eviction that run (never destroys valid cached plaintext); revocation still works on any clean pass.

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

Source code in memory/sharing/bundle.py 
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
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
def decrypt_bundle(
    bundle_dir: Path | str,
    output_memory_dir: Path | str,
    *,
    identity: LocalIdentity | Any | None = None,
    private_key: str | None = None,
) -> DecryptResult:
    """Decrypt every unit this key can open and materialize a ``.memory`` tree.

    For each manifest unit the private key is tried; units that fail to decrypt
    (i.e. were not encrypted to this key) are silently skipped. Successfully
    decrypted units are written into a ``.memory``-shaped ``output_memory_dir``
    (``project.md`` plus ``entries/<id>.md``, ``documents/<id>.md``,
    ``sessions/<id>.md`` and ``skills/<name>.md``) so that
    :func:`memory.federation.read_repo` can read the result unchanged.

    Each unit's id/name is read from the decrypted frontmatter (it is never
    stored in the clear) and validated / path-jailed before being used as a
    filename, so a hostile bundle cannot escape ``output_memory_dir``.

    Each cleartext ``manifest.json`` unit path is validated BEFORE it is read
    (rejecting absolute paths, ``..`` traversal, and anything but the opaque
    ``units/<sha256>.age`` / ``project.age`` layout — enforced regardless of the
    manifest's self-declared format) so a malicious publisher cannot use this
    function to read arbitrary local files. A missing or corrupt manifest is
    treated as an empty bundle (the cache is still reconciled), never a crash.

    The output ``.memory`` tree is RECONCILED to the current bundle: stale files
    (units that dropped out of a newer, re-scoped bundle or are no longer openable
    by this key) are removed from EACH owned dir (``entries/``, ``documents/``,
    ``skills/``, ``sessions/``), and ``project.md`` is refreshed or removed to
    match. Only the ``.md`` files in those four dirs and ``project.md`` this
    function owns are touched — unrelated files (e.g. ``.git``) are left alone. A
    transient I/O fault during the pass SKIPS all eviction that run (never
    destroys valid cached plaintext); revocation still works on any clean pass.

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

    # A genuinely-ABSENT manifest (FileNotFoundError) or a CORRUPT one
    # (json.JSONDecodeError / ValueError) means the bundle really is empty/gone:
    # treat it as an EMPTY bundle so the reconcile phase still evicts revoked
    # ``entries/*.md`` + ``project.md`` (revocation). Only log a warning.
    #
    # Any OTHER OSError is a TRANSIENT I/O fault (EACCES/EIO/NFS, or a OneDrive
    # placeholder not yet hydrated). Reconciling as empty on a transient fault
    # would destroy the entire valid peer cache, so ABORT WITHOUT reconciling —
    # leave the cache untouched and let the next clean sync reconcile normally.
    # (FileNotFoundError is a subclass of OSError, so it is matched first.)
    try:
        manifest = read_manifest(bundle_dir)
    except FileNotFoundError as exc:
        logger.warning(
            "bundle manifest absent (%s); reconciling cache as empty bundle", 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 bundle", 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_memory_dir=output,
            entry_ids=(),
            project_written=False,
            total_units=0,
        )

    # Each unit kind materializes into its own owned ``.memory`` subdir.
    kind_dirs: dict[str, Path] = {
        kind: output / dirname for kind, dirname in _KIND_DIRNAME.items()
    }

    # Phase 1: decrypt everything this key can open, buffering results so the
    # output tree can be reconciled (stale units removed) atomically at the end.
    # (dest, plaintext, identifier) per kind.
    writes: dict[str, list[tuple[Path, bytes, str]]] = {k: [] for k in kind_dirs}
    seen_names: dict[str, set[str]] = {k: set() for k in kind_dirs}
    project_plaintext: bytes | None = None
    # Set if ANY transient I/O error occurs during this pass. When set, the
    # stale-eviction/reconcile step is skipped so a transient fault can never
    # delete valid cached plaintext (see Phase 2).
    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:
            # The unit IS present on disk (``exists()`` above) but could not be
            # read — a transient I/O fault (EACCES/EIO, or a OneDrive placeholder
            # not yet hydrated), NOT a legit "missing unit". Flag it so this
            # unit's absence from ``seen_names`` does not evict its cached entry;
            # eviction is skipped this run and waits for the next clean sync.
            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:
            # Not encrypted to this key — nothing to see here.
            continue
        except Exception as exc:  # pragma: no cover - defensive
            logger.warning("failed to decrypt unit %s: %s", unit.path, exc)
            continue

        if unit.kind == KIND_PROJECT:
            project_plaintext = plaintext
            continue
        if unit.kind not in kind_dirs:
            logger.warning("decrypted unit %s has unknown kind %r; skipping",
                           unit.path, unit.kind)
            continue

        try:
            meta, _ = storage._parse_frontmatter(plaintext.decode("utf-8"))
        except UnicodeDecodeError:
            logger.warning("decrypted unit %s is not valid UTF-8; skipping", unit.path)
            continue
        dest, ident = _unit_dest(unit.kind, meta, kind_dirs[unit.kind])
        if dest is None:
            logger.warning("decrypted %s unit %s has no usable id; skipping",
                           unit.kind, unit.path)
            continue
        writes[unit.kind].append((dest, plaintext, ident))
        seen_names[unit.kind].add(dest.name)

    # Phase 2: reconcile each owned dir so it reflects ONLY what this bundle now
    # opens. Remove stale ``*.md`` (surgically, per dir), then write the current
    # ones. Only reconcile (evict) when the pass completed with NO transient I/O
    # error — a transient fault means we may not have seen every still-granted
    # unit, so evicting now could delete valid cached plaintext; skip eviction and
    # let the next clean sync reconcile. Revocation still works on any clean pass.
    # Freshly decrypted units are written either way (additive, never destructive).
    written: dict[str, list[str]] = {k: [] for k in kind_dirs}
    for kind, out_dir in kind_dirs.items():
        # ``entries/`` is always materialized (preserves prior behavior). The
        # other kinds are only touched when this producer actually shares them
        # (units present) OR a prior sync already created the dir — so a peer that
        # never receives documents/skills/sessions is not littered with empty
        # dirs, while full revocation of a kind still reconciles its pre-existing
        # dir to empty (same as entries).
        if kind == KIND_ENTRY or writes[kind]:
            out_dir.mkdir(parents=True, exist_ok=True)
        elif not out_dir.exists():
            continue
        if not transient_io_error:
            for existing in out_dir.glob("*.md"):
                if existing.name in seen_names[kind]:
                    continue
                # ``entries/`` is EXCLUSIVELY owned -> evict by filename. The other
                # kinds' dirs may legitimately hold files this function does not
                # own (another slice's drop-ins), so only evict a stale file that
                # is a well-formed unit of this kind (see _is_owned_unit): a
                # revoked unit we materialized always parses as owned and is
                # evicted, while an unrelated file is left untouched.
                if kind != KIND_ENTRY and not _is_owned_unit(existing, kind):
                    continue
                try:
                    existing.unlink()
                except FileNotFoundError:
                    pass
        for dest, plaintext, ident in writes[kind]:
            atomic_write(dest, plaintext)
            written[kind].append(ident)

    # Refresh or drop project.md to match the current bundle. The drop is part of
    # reconcile, so it too is gated on a clean pass — a transient fault must not
    # evict a valid cached project.md.
    project_path = output / "project.md"
    project_written = False
    if project_plaintext is not None:
        atomic_write(project_path, project_plaintext)
        project_written = True
    elif project_path.exists() and not transient_io_error:
        try:
            project_path.unlink()
        except FileNotFoundError:
            pass

    return DecryptResult(
        output_memory_dir=output,
        entry_ids=tuple(written[KIND_ENTRY]),
        project_written=project_written,
        total_units=len(manifest.units),
        document_ids=tuple(written[KIND_DOCUMENT]),
        skill_names=tuple(written[KIND_SKILL]),
        session_ids=tuple(written[KIND_SESSION]),
    )

open_bundle 

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

Alias for :func:decrypt_bundle (decrypt + materialize a .memory tree).

Source code in memory/sharing/bundle.py 
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
def open_bundle(
    bundle_dir: Path | str,
    output_memory_dir: Path | str,
    *,
    identity: LocalIdentity | Any | None = None,
    private_key: str | None = None,
) -> DecryptResult:
    """Alias for :func:`decrypt_bundle` (decrypt + materialize a ``.memory`` tree)."""
    return decrypt_bundle(
        bundle_dir,
        output_memory_dir,
        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 memory/sharing/bundle.py 
355
356
357
358
359
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)

document_recipients 

document_recipients(sharing: SharingConfig, documents: Iterable[Mapping[str, Any]], entry_recipients_map: Mapping[str, set[str]]) -> dict[str, set[str]]

Map each shareable document id to its recipient set.

A document rides along with the entry it documents: it is shareable to a recipient R iff EVERY entry named in its links_to is bundled (present in entry_recipients_map) AND R can see all of them. The recipient set is therefore the INTERSECTION of the links_to targets' recipient sets — this guarantees that any retained links_to target is visible to every recipient of the document (so the id never leaks; see the per-unit scrub in :mod:~memory.sharing.bundle).

A document is OMITTED (returns no entry) when it has no links_to (orphan) or any target is itself omitted/private. Only status == "active" documents are considered. Per-document grants['doc:<id>'] overrides are intentionally NOT honoured here (see the module notes): widening a document's audience beyond its link target's would force the links_to to be scrubbed for the extra recipients, defeating the document's purpose. Deferred.

Source code in memory/sharing/config.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
def document_recipients(
    sharing: SharingConfig,
    documents: Iterable[Mapping[str, Any]],
    entry_recipients_map: Mapping[str, set[str]],
) -> dict[str, set[str]]:
    """Map each shareable document id to its recipient set.

    A document rides along with the entry it documents: it is shareable to a
    recipient R iff EVERY entry named in its ``links_to`` is bundled (present in
    ``entry_recipients_map``) AND R can see all of them. The recipient set is
    therefore the INTERSECTION of the ``links_to`` targets' recipient sets — this
    guarantees that any retained ``links_to`` target is visible to every
    recipient of the document (so the id never leaks; see the per-unit scrub in
    :mod:`~memory.sharing.bundle`).

    A document is OMITTED (returns no entry) when it has no ``links_to`` (orphan)
    or any target is itself omitted/private. Only ``status == "active"``
    documents are considered. Per-document ``grants['doc:<id>']`` overrides are
    intentionally NOT honoured here (see the module notes): widening a document's
    audience beyond its link target's would force the ``links_to`` to be scrubbed
    for the extra recipients, defeating the document's purpose. Deferred.
    """
    result: dict[str, set[str]] = {}
    for doc in documents:
        if str(doc.get("status") or "active") != "active":
            continue
        doc_id = str(doc.get("id") or "").strip()
        if not doc_id:
            continue
        targets = _link_targets(doc.get("links_to"))
        if not targets:
            continue
        recips: set[str] | None = None
        for target in targets:
            target_recips = entry_recipients_map.get(target)
            if not target_recips:
                recips = set()
                break
            recips = set(target_recips) if recips is None else (recips & set(target_recips))
        if recips:
            result[doc_id] = recips
    return result

effective_grants 

effective_grants(entry: Mapping[str, Any], entries_by_id: Mapping[str, Mapping[str, Any]], sharing: SharingConfig) -> set[str]

Recipient ids explicitly granted to entry via any ancestor grant.

Source code in memory/sharing/config.py 
302
303
304
305
306
307
308
309
310
311
def effective_grants(
    entry: Mapping[str, Any],
    entries_by_id: Mapping[str, Mapping[str, Any]],
    sharing: SharingConfig,
) -> set[str]:
    """Recipient ids explicitly granted to ``entry`` via any ancestor grant."""
    entry_id = str(entry.get("id") or "")
    if not entry_id:
        return set()
    return _grants_for_chain(_ancestor_ids(entry_id, entries_by_id), sharing)

effective_scope 

effective_scope(entry: Mapping[str, Any], entries_by_id: Mapping[str, Mapping[str, Any]], sharing: SharingConfig) -> Scope

Resolve the scope that applies to entry after subtree inheritance.

Source code in memory/sharing/config.py 
290
291
292
293
294
295
296
297
298
299
def effective_scope(
    entry: Mapping[str, Any],
    entries_by_id: Mapping[str, Mapping[str, Any]],
    sharing: SharingConfig,
) -> Scope:
    """Resolve the scope that applies to ``entry`` after subtree inheritance."""
    entry_id = str(entry.get("id") or "")
    if not entry_id:
        return Scope(SCOPE_PRIVATE)
    return _scope_for_chain(_ancestor_ids(entry_id, entries_by_id), sharing)

entry_recipients 

entry_recipients(sharing: SharingConfig, entries: Iterable[Mapping[str, Any]], firm_members: Iterable[str] = (), *, index_entries: Iterable[Mapping[str, Any]] | None = None) -> dict[str, set[str]]

Map each shareable entry id to the concrete set of recipient ids.

Unlike :func:is_shareable_to, this EXPANDS a firm scope to the actual firm_members (the registered identities) so the bundle builder knows the exact public keys to encrypt to. Entries that resolve to no recipients (private and ungranted) are omitted from the result — the caller uses this to physically drop them from the bundle before any encryption.

entries are the entries actually considered for bundling (typically the ACTIVE entries). index_entries is the pool used to build the ancestor-resolution index — pass ALL entries (including discarded/any status) here so scope/grant inheritance can traverse a discarded intermediate plan node. When index_entries is None the index is built from entries (backwards-compatible behaviour). Only ids in entries are resolved and returned, so a discarded ancestor never gets bundled itself — it only keeps the chain from its active descendants intact.

Source code in memory/sharing/config.py 
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
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
def entry_recipients(
    sharing: SharingConfig,
    entries: Iterable[Mapping[str, Any]],
    firm_members: Iterable[str] = (),
    *,
    index_entries: Iterable[Mapping[str, Any]] | None = None,
) -> dict[str, set[str]]:
    """Map each shareable entry id to the concrete set of recipient ids.

    Unlike :func:`is_shareable_to`, this EXPANDS a ``firm`` scope to the actual
    ``firm_members`` (the registered identities) so the bundle builder knows the
    exact public keys to encrypt to. Entries that resolve to no recipients
    (``private`` and ungranted) are omitted from the result — the caller uses
    this to physically drop them from the bundle before any encryption.

    ``entries`` are the entries actually considered for bundling (typically the
    ACTIVE entries). ``index_entries`` is the pool used to build the
    ancestor-resolution index — pass ALL entries (including discarded/any
    status) here so scope/grant inheritance can traverse a discarded
    intermediate plan node. When ``index_entries`` is ``None`` the index is
    built from ``entries`` (backwards-compatible behaviour). Only ids in
    ``entries`` are resolved and returned, so a discarded ancestor never gets
    bundled itself — it only keeps the chain from its active descendants intact.
    """
    entries = list(entries)
    index_source = list(index_entries) if index_entries is not None else entries
    by_id = {str(e.get("id")): e for e in index_source if e.get("id")}
    firm = set(firm_members)
    result: dict[str, set[str]] = {}
    for e in entries:
        entry_id = str(e.get("id") or "")
        if not entry_id:
            continue
        chain = _ancestor_ids(entry_id, by_id)
        recips = set(_grants_for_chain(chain, sharing))
        scope = _scope_for_chain(chain, sharing)
        if scope.kind == SCOPE_FIRM:
            recips.update(firm)
        elif scope.kind == SCOPE_BILATERAL:
            recips.update(scope.audience)
        if recips:
            result[entry_id] = recips
    return result

is_shareable_to 

is_shareable_to(entry: Mapping[str, Any], entries_by_id: Mapping[str, Mapping[str, Any]], sharing: SharingConfig, recipient_id: str) -> bool

Whether entry is shareable to recipient_id.

Note: a firm-scoped entry is shareable to ANY recipient id here — firm membership is enforced by the registry at bundle-build time (see :func:entry_recipients), not by this logical query.

Source code in memory/sharing/config.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def is_shareable_to(
    entry: Mapping[str, Any],
    entries_by_id: Mapping[str, Mapping[str, Any]],
    sharing: SharingConfig,
    recipient_id: str,
) -> bool:
    """Whether ``entry`` is shareable to ``recipient_id``.

    Note: a ``firm``-scoped entry is shareable to ANY recipient id here — firm
    membership is enforced by the registry at bundle-build time (see
    :func:`entry_recipients`), not by this logical query.
    """
    entry_id = str(entry.get("id") or "")
    if not entry_id:
        return False
    chain = _ancestor_ids(entry_id, entries_by_id)
    if recipient_id in _grants_for_chain(chain, sharing):
        return True
    scope = _scope_for_chain(chain, sharing)
    if scope.kind == SCOPE_FIRM:
        return True
    if scope.kind == SCOPE_BILATERAL and recipient_id in scope.audience:
        return True
    return False

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 returned by :func:memory.storage.read_config. Missing/empty/malformed blocks yield an empty :class:SharingConfig (nothing shared) — this function never raises on a partially-formed config; unusable items are skipped.

Source code in memory/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
206
207
208
209
210
211
212
213
214
215
216
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 returned by
    :func:`memory.storage.read_config`. Missing/empty/malformed blocks yield an
    empty :class:`SharingConfig` (nothing shared) — this function never raises on
    a partially-formed config; unusable items are skipped.
    """
    if not isinstance(config, Mapping):
        return SharingConfig()

    scopes: dict[str, Scope] = {}
    transports: dict[str, str] = {}
    skills_default = Scope(SCOPE_PRIVATE)
    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-node transport tier (``sharing.transports``), keyed exactly like
        # ``sharing.scopes``. Unlike scopes (which fail-closed to ``private`` on a
        # typo), an unrecognised transport is REJECTED with a clear error: the
        # network tier is deny-by-default, and a silent fallback could either hide
        # a hosting typo or (worse) be read as widening — so we surface it loudly.
        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)
        # Skills live OUTSIDE the entry tree, so they take a single global
        # default scope from ``sharing.skills`` (string shorthand or a
        # ``{scope: bilateral, audience: [...]}`` mapping). Absent -> private.
        if "skills" in sharing_block:
            skills_default = _parse_scope(sharing_block.get("skills"))

    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,
        skills_default=skills_default,
        transports=transports,
    )

session_recipients 

session_recipients(sharing: SharingConfig, sessions: Iterable[Mapping[str, Any]], entry_recipients_map: Mapping[str, set[str]]) -> dict[str, set[str]]

Map each shareable session id to its recipient set.

A session is shareable to a recipient R iff at least ONE of its referenced entries (the union of entries_created/entries_accessed/ entries_modified) is visible to R. The session unit's recipient set is therefore the UNION over its referenced entries of their recipient sets.

This is the same per-unit model as entries: the unit is encrypted to that union U, and the per-unit scrub then keeps in the three ref arrays only the entry ids whose recipient set is a superset of U (so every recipient of the session can open every id it retains). A session whose arrays scrub to empty is omitted entirely by the builder. Sessions have no status field, so all present sessions are considered.

Source code in memory/sharing/config.py 
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
def session_recipients(
    sharing: SharingConfig,
    sessions: Iterable[Mapping[str, Any]],
    entry_recipients_map: Mapping[str, set[str]],
) -> dict[str, set[str]]:
    """Map each shareable session id to its recipient set.

    A session is shareable to a recipient R iff at least ONE of its referenced
    entries (the union of ``entries_created``/``entries_accessed``/
    ``entries_modified``) is visible to R. The session unit's recipient set is
    therefore the UNION over its referenced entries of their recipient sets.

    This is the same per-unit model as entries: the unit is encrypted to that
    union U, and the per-unit scrub then keeps in the three ref arrays only the
    entry ids whose recipient set is a superset of U (so every recipient of the
    session can open every id it retains). A session whose arrays scrub to empty
    is omitted entirely by the builder. Sessions have no ``status`` field, so all
    present sessions are considered.
    """
    result: dict[str, set[str]] = {}
    for session in sessions:
        sid = str(session.get("id") or "").strip()
        if not sid:
            continue
        recips: set[str] = set()
        for eid in _session_referenced_ids(session):
            recips |= entry_recipients_map.get(eid, set())
        if recips:
            result[sid] = recips
    return result

shareable_entries 

shareable_entries(sharing: SharingConfig, entries: Iterable[Mapping[str, Any]], recipient_id: str) -> list[Mapping[str, Any]]

Return the subset of entries shareable to recipient_id.

entries is the list of entry dicts (as produced by :func:memory.storage.read_all; each must carry id and parent_id). With an empty :class:SharingConfig this returns [] — nothing shared.

Source code in memory/sharing/config.py 
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
def shareable_entries(
    sharing: SharingConfig,
    entries: Iterable[Mapping[str, Any]],
    recipient_id: str,
) -> list[Mapping[str, Any]]:
    """Return the subset of ``entries`` shareable to ``recipient_id``.

    ``entries`` is the list of entry dicts (as produced by
    :func:`memory.storage.read_all`; each must carry ``id`` and ``parent_id``).
    With an empty :class:`SharingConfig` this returns ``[]`` — nothing shared.
    """
    entries = list(entries)
    by_id = {str(e.get("id")): e for e in entries if e.get("id")}
    return [
        e
        for e in entries
        if e.get("id") and is_shareable_to(e, by_id, sharing, recipient_id)
    ]

skill_recipients 

skill_recipients(sharing: SharingConfig, skills: Iterable[Mapping[str, Any]], firm_members: Iterable[str] = ()) -> dict[str, set[str]]

Map each shareable skill (by name) to its recipient set.

Skills live OUTSIDE the entry tree, so they cannot inherit a subtree scope. Their audience is the global sharing.skills default (:attr:SharingConfig.skills_default) UNIONED with any per-skill grants['skill:<name>'] override:

  • private (default) — nobody, unless a skill:<name> grant lists them.
  • firm — every registered firm_members id.
  • bilateral — the scope's audience ids.

Only status == "active" skills are considered; a skill resolving to no recipients is omitted from the result (physically dropped from the bundle).

Source code in memory/sharing/config.py 
545
546
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
def skill_recipients(
    sharing: SharingConfig,
    skills: Iterable[Mapping[str, Any]],
    firm_members: Iterable[str] = (),
) -> dict[str, set[str]]:
    """Map each shareable skill (by name) to its recipient set.

    Skills live OUTSIDE the entry tree, so they cannot inherit a subtree scope.
    Their audience is the global ``sharing.skills`` default
    (:attr:`SharingConfig.skills_default`) UNIONED with any per-skill
    ``grants['skill:<name>']`` override:

    * ``private`` (default) — nobody, unless a ``skill:<name>`` grant lists them.
    * ``firm``   — every registered ``firm_members`` id.
    * ``bilateral`` — the scope's ``audience`` ids.

    Only ``status == "active"`` skills are considered; a skill resolving to no
    recipients is omitted from the result (physically dropped from the bundle).
    """
    firm = set(firm_members)
    scope = sharing.skills_default
    result: dict[str, set[str]] = {}
    for skill in skills:
        if str(skill.get("status") or "active") != "active":
            continue
        name = str(skill.get("name") or skill.get("id") or "").strip()
        if not name:
            continue
        recips: set[str] = set(sharing.grants.get(f"skill:{name}", ()))
        if scope.kind == SCOPE_FIRM:
            recips |= firm
        elif scope.kind == SCOPE_BILATERAL:
            recips |= set(scope.audience)
        if recips:
            result[name] = recips
    return result

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)

project_for_caller 

project_for_caller(caller_id: str, *, registry: IdentityRegistry, output_memory_dir: Path | str, source_memory_dir: Path | str | None = None, sharing: SharingConfig | None = None, include_discarded: bool = False, allowed_transports: set[str] | None = None) -> ProjectionResult

Write a .memory-shaped PLAINTEXT store of ONLY what caller_id may see.

Reuses the shared filter+scrub collector (the same code :func:memory.sharing.bundle.build_bundle runs), then materializes — as PLAINTEXT, with no encryption round-trip — exactly the units the caller is a registered recipient of, into output_memory_dir (project.md plus entries/<id>.md, documents/<id>.md, sessions/<id>.md and skills/<name>.md), matching :func:~memory.sharing.bundle.decrypt_bundle's output shape so :func:memory.federation.read_repo reads it unchanged.

Parity: the produced set of units and their scrubbed content are byte-for-byte what caller_id would obtain via build_bundle + decrypt_bundle with that caller's key (see the module docstring). sharing defaults to the policy parsed from <source_memory_dir>/config.yaml; source_memory_dir defaults to the local .memory store.

A caller that is unregistered or has nothing shared to it yields an empty projection (only an empty entries/ dir, no project.md) — matching a decrypt that opens nothing.

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 output_memory_dir and RECONCILES it on every call, exactly like :func:~memory.sharing.bundle.decrypt_bundle: re-projecting a caller into a previously-populated dir evicts the *.md units the caller is no longer entitled to (from entries/, documents/, skills/, sessions/) and drops a stale project.md when the caller ends up with zero units, so a revoked grant genuinely disappears on the next read. Foreign (unowned) files are preserved, and a transient I/O fault while scanning the output skips eviction that run (never destroying a valid cache). The result is byte-identical to a fresh build_bundle + decrypt_bundle for the caller's CURRENT grants even when projecting into a reused output dir.

Source code in memory/sharing/projection.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
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
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
def project_for_caller(
    caller_id: str,
    *,
    registry: IdentityRegistry,
    output_memory_dir: Path | str,
    source_memory_dir: Path | str | None = None,
    sharing: SharingConfig | None = None,
    include_discarded: bool = False,
    allowed_transports: set[str] | None = None,
) -> ProjectionResult:
    """Write a ``.memory``-shaped PLAINTEXT store of ONLY what ``caller_id`` may see.

    Reuses the shared filter+scrub collector (the same code
    :func:`memory.sharing.bundle.build_bundle` runs), then materializes — as
    PLAINTEXT, with no encryption round-trip — exactly the units the caller is a
    registered recipient of, into ``output_memory_dir`` (``project.md`` plus
    ``entries/<id>.md``, ``documents/<id>.md``, ``sessions/<id>.md`` and
    ``skills/<name>.md``), matching :func:`~memory.sharing.bundle.decrypt_bundle`'s
    output shape so :func:`memory.federation.read_repo` reads it unchanged.

    Parity: the produced set of units and their scrubbed content are byte-for-byte
    what ``caller_id`` would obtain via ``build_bundle`` + ``decrypt_bundle`` with
    that caller's key (see the module docstring). ``sharing`` defaults to the
    policy parsed from ``<source_memory_dir>/config.yaml``; ``source_memory_dir``
    defaults to the local ``.memory`` store.

    A caller that is unregistered or has nothing shared to it yields an empty
    projection (only an empty ``entries/`` dir, no ``project.md``) — matching a
    decrypt that opens nothing.

    ``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 ``output_memory_dir`` and RECONCILES it on every call,
    exactly like :func:`~memory.sharing.bundle.decrypt_bundle`: re-projecting a
    caller into a previously-populated dir evicts the ``*.md`` units the caller is
    no longer entitled to (from ``entries/``, ``documents/``, ``skills/``,
    ``sessions/``) and drops a stale ``project.md`` when the caller ends up with
    zero units, so a revoked grant genuinely disappears on the next read. Foreign
    (unowned) files are preserved, and a transient I/O fault while scanning the
    output skips eviction that run (never destroying a valid cache). The result is
    byte-identical to a fresh ``build_bundle`` + ``decrypt_bundle`` for the
    caller's CURRENT grants even when projecting into a reused output dir.
    """
    source = (
        Path(source_memory_dir) if source_memory_dir is not None else storage.MEMORY_DIR
    )
    output = Path(output_memory_dir)

    if sharing is None:
        sharing = parse_sharing_config(_read_source_config(source))

    collected = _collect_shared_units(
        source=source,
        sharing=sharing,
        registry=registry,
        include_discarded=include_discarded,
        allowed_transports=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 ``registered_recipients``. 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 project a unit the
    # caller's key cannot decrypt (or vice versa), 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)
    caller_units = (
        [u for u in collected.units if caller_pubkey in u.pubkeys]
        if caller_pubkey is not None
        else []
    )

    kind_dirs = {kind: output / dirname for kind, dirname in _KIND_DIRNAME.items()}

    # Phase 1: resolve each selected unit's path-jailed destination exactly as
    # decrypt_bundle does (so filenames match) and BUFFER the writes, recording the
    # basenames a live unit occupies this run in ``seen_names`` — the provenance the
    # reconcile evicts against.
    writes: dict[str, list[tuple[Path, bytes, str]]] = {k: [] for k in kind_dirs}
    seen_names: dict[str, set[str]] = {k: set() for k in kind_dirs}
    for unit in caller_units:
        try:
            meta, _ = storage._parse_frontmatter(unit.scrubbed.decode("utf-8"))
        except UnicodeDecodeError:
            continue
        dest, ident = _unit_dest(unit.kind, meta, kind_dirs[unit.kind])
        if dest is None:
            continue
        writes[unit.kind].append((dest, unit.scrubbed, ident))
        seen_names[unit.kind].add(dest.name)

    # Phase 2: reconcile each owned dir so it reflects ONLY this caller's CURRENT
    # units — identical to decrypt_bundle. Evict stale ``*.md`` (surgically, per
    # dir), then write the current ones. ``entries/`` is EXCLUSIVELY owned so a
    # stale file is evicted by filename; the other kinds' dirs may hold foreign
    # drop-ins, so only a well-formed unit of that kind (``_is_owned_unit``) is
    # evicted (a revoked unit we materialized always parses as owned). A transient
    # I/O fault while scanning a dir SKIPS all eviction this run so a read glitch
    # never destroys a valid cache (revocation still takes effect on any clean
    # pass); writes are additive either way.
    transient_io_error = False
    written: dict[str, list[str]] = {k: [] for k in kind_dirs}
    for kind, out_dir in kind_dirs.items():
        # ``entries/`` is always materialized (mirrors decrypt_bundle). The other
        # kinds are only touched when this caller receives such a unit OR the dir
        # already exists — so a caller that never gets documents/skills/sessions is
        # not littered with empty dirs, while full revocation of a kind still
        # reconciles its pre-existing dir to empty.
        if kind == KIND_ENTRY or writes[kind]:
            out_dir.mkdir(parents=True, exist_ok=True)
        elif not out_dir.exists():
            continue
        try:
            existing_md = list(out_dir.glob("*.md"))
        except OSError as exc:
            logger.warning(
                "could not scan %s for reconcile (transient I/O): %s; "
                "skipping stale-eviction this run to preserve valid cached "
                "plaintext", out_dir, exc
            )
            transient_io_error = True
            existing_md = []
        for existing in existing_md:
            if existing.name in seen_names[kind]:
                continue
            if kind != KIND_ENTRY and not _is_owned_unit(existing, kind):
                continue
            try:
                existing.unlink()
            except FileNotFoundError:
                pass
        for dest, scrubbed, ident in writes[kind]:
            atomic_write(dest, scrubbed)
            written[kind].append(ident)

    # project.md is written iff the caller can open the project stub. build_bundle
    # encrypts the stub to the UNION of every written unit's recipients, so the
    # caller can decrypt it iff it is a recipient of at least one written unit —
    # exactly "this caller has >= 1 projected unit". When the caller now has zero
    # units a stale project.md is dropped (part of reconcile, gated on a clean pass
    # so a transient fault never evicts a valid cached project.md).
    project_path = output / "project.md"
    project_written = False
    if collected.project_stub is not None and caller_units:
        output.mkdir(parents=True, exist_ok=True)
        atomic_write(project_path, collected.project_stub)
        project_written = True
    elif project_path.exists() and not transient_io_error:
        try:
            project_path.unlink()
        except FileNotFoundError:
            pass

    return ProjectionResult(
        output_memory_dir=output,
        entry_ids=tuple(written[KIND_ENTRY]),
        project_written=project_written,
        total_units=len(caller_units),
        document_ids=tuple(written[KIND_DOCUMENT]),
        skill_names=tuple(written[KIND_SKILL]),
        session_ids=tuple(written[KIND_SESSION]),
        caller_id=caller_id,
        transient_io_error=transient_io_error,
    )

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"

generate_keypair 

generate_keypair() -> KeyPair

Generate a fresh identity keypair: an age X25519 pair + Ed25519 signing pair.

Source code in memory/sharing/identity.py 
184
185
186
187
188
189
190
191
192
193
def generate_keypair() -> KeyPair:
    """Generate a fresh identity keypair: an ``age`` X25519 pair + Ed25519 signing pair."""
    identity = x25519.Identity.generate()
    signing_public, signing_private = signing.generate_signing_keypair()
    return KeyPair(
        public_key=str(identity.to_public()),
        private_key=str(identity),
        signing_public_key=signing_public,
        signing_private_key=signing_private,
    )

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_local_identity 

save_local_identity(identity: LocalIdentity, path: Path | str | None = None) -> Path

Write identity (including the private key) to disk, chmod 0600.

Defaults to ~/.angelo/identity.yaml. The chmod is best-effort (a no-op on filesystems that don't support POSIX modes, e.g. some Windows setups).

Source code in memory/sharing/identity.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def save_local_identity(identity: LocalIdentity, path: Path | str | None = None) -> Path:
    """Write ``identity`` (including the private key) to disk, chmod ``0600``.

    Defaults to ``~/.angelo/identity.yaml``. The chmod is best-effort (a no-op on
    filesystems that don't support POSIX modes, e.g. some Windows setups).
    """
    dest = Path(path) if path is not None else default_identity_path()
    atomic_write(
        dest,
        yaml.dump(_identity_to_dict(identity), default_flow_style=False, sort_keys=False),
    )
    try:
        os.chmod(dest, 0o600)
    except OSError as exc:  # pragma: no cover - platform dependent
        logger.warning("could not chmod identity file %s to 0600: %s", dest, exc)
    return dest

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