memory.sharing.bundle

memory.sharing.bundle

On-disk bundle format: filter -> encrypt (publish) and decrypt -> materialize.

This module defines the contract between a producer (who publishes a filtered, per-recipient-encrypted bundle) and a consumer (who git-fetches the bundle and decrypts only what its key opens). The downstream publish/sync CLI is built on top of these two functions; it never needs to read this module's internals.

On-disk bundle layout

A bundle is a plain directory (typically the working tree of a producer-owned "bundle repo") laid out as::

<bundle_root>/
    manifest.json          # cleartext structural index (see below)
    project.age            # the project.md node, age-encrypted
    units/<digest>.age     # one age ciphertext per shared entry

Each units/*.age file is a self-contained age ciphertext of one entry's rendered markdown (the exact bytes memory/storage.py wrote, frontmatter included), encrypted natively to every recipient allowed to see that entry. A recipient decrypts a unit iff their private key is among its recipients.

Manifest — what is in the clear vs encrypted

The manifest is deliberately MINIMAL and privacy-first. In the clear it carries only: the format id/version, a creation timestamp, an optional producer id (for attribution / peer registration), and a list of {kind, path} unit pointers.

It intentionally OMITS entry ids, titles, parent links, and recipient identities. Everything about an entry — including its id — lives INSIDE the encrypted unit (in the frontmatter) and is recovered only after a successful decrypt. Consumers therefore learn a unit's id only if they can open it.

Metadata-leak tradeoff (accepted, per the encryption-first design decision): a ciphertext store still unavoidably leaks structural metadata — the number of units, their kind (entry vs project), and each unit's ciphertext size. It does NOT leak entry ids, contents, or who-can-see-what (recipient sets are encoded only inside the age header stanzas, not in the manifest). Unit filenames are opaque SHA-256 digests of the entry id: stable across republishes (nice git deltas) and meaningless to anyone who doesn't already know the id. Fully hiding unit count/sizes would require padding or an all-in-one encrypted archive; that is deferred.

Revocation-in-history (accepted limitation)

Revocation is NOT retroactive. A bundle is typically the working tree of a git "bundle repo", so every previously published ciphertext persists in that repo's git history, and the per-unit age recipient keys are never rotated. Once a recipient has fetched (or could re-fetch) an old commit, narrowing a scope or dropping a grant only affects future bundles — it cannot claw back bytes the recipient already holds or can read from history. In short: you can un-share the future, not the past. Truly retroactive revocation would require key rotation plus re-encryption of the whole history, which is out of scope for this design.

Scrub scope (what is redacted vs what still leaks)

The per-unit scrubs redact references to entries a recipient cannot open:

• ENTRIES — :func:_scrub_cross_links redacts ungranted entry ids in the STRUCTURED frontmatter link fields (related_to/invalidates/invalidated_by/parent_id/session_id).
• DOCUMENTS — :func:_scrub_document_links drops links_to targets the recipient cannot open.
• SESSIONS — :func:_scrub_session_refs scrubs BOTH the structured ref arrays (entries_created/accessed/modified) AND the auto-derived FREE-TEXT (the search_text body plus a title/summary built from entry titles), so a session shared to R never names — by id OR title, in any AUTO-DERIVED field — an entry R cannot open. Note this covers only auto-derived fields: user-authored prose (a session's notes, or a handoff title/ summary the user typed) is NOT auto-derived and ships verbatim, the same accepted body-leak class as an entry body (see residual leaks below).

Accepted residual leaks (out of scope for this slice):

• Free-text prose that is NOT auto-derived from entry titles still ships verbatim — an ungranted id/title in an entry's BODY, in files/node_label, in a session's notes, or in a document's description/path (the latter two ride a narrower intersection audience, the same class as an entry body leak).
• The cleartext manifest still reveals STRUCTURAL metadata (unit count, kind, and each unit's ciphertext size) but no ids/contents/recipients; and revocation is forward-only (see below).
• The project.md stub (id + name only, empty body) is INTENTIONALLY encrypted to every grantee — including skill-only recipients — so their cache can render the peer.

Decrypted-cache layout

:func:decrypt_bundle writes a .memory-shaped directory so the existing :func:memory.federation.read_repo can read it UNCHANGED::

<output_memory_dir>/
    project.md
    entries/<id>.md

The caller passes the .memory directory of a federation cache (e.g. .angelo/federation-cache/<peer_id>/.memory); registering that cache root as a federated repo then renders it like any other peer.

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

@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 ""))

BundleManifest dataclass

The cleartext manifest.json at the bundle root.

Source code in memory/sharing/bundle.py

@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,
        )

BuildResult dataclass

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

Source code in memory/sharing/bundle.py

@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

DecryptResult dataclass

Summary of a :func:decrypt_bundle run.

Source code in memory/sharing/bundle.py

@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, ...] = ()

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

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)

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

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

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

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,
    )