zettelkasten.synapse.sharing.bundle

zettelkasten.synapse.sharing.bundle

On-disk bundle format for synapse-edge sharing: filter -> encrypt (publish) and decrypt -> materialize a .synapse overlay cache.

Mirrors :mod:memory.sharing.bundle and :mod:zettelkasten.sharing.bundle, but for the cross-store LINK OVERLAY (.synapse/links/links.json). A producer publishes a filtered, per-recipient-encrypted slice of its local synapse edges; a consumer git-fetches the bundle and decrypts only the edges its key can open, REBUILDING a valid overlay into a federation cache (.angelo/federation-cache/<peer>/.synapse) that :func:zettelkasten.synapse.overlay.load_overlay / get_connections read like any other overlay.

Id-secrecy is the whole point

A synapse edge is bipartite: it names BOTH a memory entry id and a ZK note id. So it must ship ONLY to recipients who can already open BOTH endpoints — the INTERSECTION of the two stores' per-unit recipient sets (see :func:zettelkasten.synapse.sharing.config.edge_recipients). Filtering happens BEFORE any encryption: an edge with no both-ends recipient (or an unresolvable endpoint) is physically OMITTED, never encrypted-then-dropped.

Cross-producer (federated) edges

An edge whose zk_source contains ":" (e.g. "remote:alpha") names a ZK note in a DIFFERENT producer's repo. Its ZK-end recipients cannot be computed locally, so :func:build_bundle loads the peer's DECRYPTED recipient-index policy from the federation cache (published by that peer alongside its ZK bundle; see :mod:zettelkasten.synapse.sharing.recipient_index) and intersects it with the local memory-entry recipients. This is strictly FAIL-CLOSED: if the peer policy is missing, unresolvable, or does not contain the note, the edge is OMITTED. Same-producer (non-":") edges are unaffected — their behaviour is identical to before.

On-disk bundle layout

::

<bundle_root>/
    manifest.json          # cleartext structural index: [{kind, path}]
    units/<digest>.age     # one age ciphertext per shared edge

Each units/*.age is an age ciphertext of a small envelope: a single-line JSON placement header ({kind, memory_id, zk_source, zk_id}) plus the newline-separated edge-record JSON payload. The placement lives INSIDE the ciphertext; the cleartext manifest.json carries only {kind, path} (no ids, no recipients) — the same privacy-first design as the memory/zk bundles. Unit filenames are sha256("<kind>:<unit_key>") digests (unit_key = "<memory_id>:<zk_source>:<zk_id>"): stable across republishes and opaque to anyone who does not already know both ids.

Deny-by-default payload allowlist

The edge JSON payload is NOT the raw edge dict — it is projected onto an EXPLICIT field allowlist (:data:_SHARED_EDGE_FIELDS, built by :func:_shareable_edge_payload), mirroring the filter-before-encrypt / scrub discipline of the memory and zk bundles. Only known-safe fields ship; any field NOT on the allowlist — including any FUTURE field an edge might grow in :mod:zettelkasten.synapse.candidates or the typer — is silently omitted, so a third-party id can never leak into a unit that reaches a one-sided-narrower recipient. Specifically:

• memory_id/zk_source/zk_id ship for self-consistency but are advisory — :func:decrypt_bundle overwrites them from the validated header.
• relation/confidence/created_at ship (overlay needs relation+confidence; created_at is display-only). confidence is float-coerced (and dropped if non-numeric) so the allowlist is not merely name-deep — a hostile value cannot smuggle an id under a numeric key.
• rationale ships: it is endpoint-derived free text, the same accepted body-leak class as an entry body in the parent memory/zk bundles. The typer that generates it is fed only a shared-provenance COUNT, never the dedicated raw shared_provenance token LIST (see :func:zettelkasten.synapse.typing_llm._build_prompt) — so the rationale can never echo a token that appears ONLY in that list (e.g. a memory entry.files path or a _citations target absent from either endpoint's text). It is NOT, however, "provably free of provenance tokens": a PATH-type token (e.g. foo/bar.py) is extracted from a ZK note BODY by :func:zettelkasten.synapse.candidates._zk_provenance, and the typer is fed that endpoint body verbatim, so such a token can appear in the rationale via the endpoint text. That is not a leak — the rationale stays bounded by the two endpoints' visible content, which a both-ends recipient can already read.
• signals is reduced to its non-identifying numeric part — {"semantic_sim": <float>} plus a shared_provenance_count (an int). The raw shared_provenance token LIST is NEVER shipped; the count preserves the "N shared artifacts" signal while leaking nothing, and a both-ends recipient can recompute the tokens from the two endpoints it already holds. semantic_sim is likewise float-coerced (dropped if non-numeric).

NOTE on overlay-shape divergence: a rebuilt federation-cache overlay's edge signals are intentionally {semantic_sim, shared_provenance_count} and DIVERGE from a natively-built overlay's {semantic_sim, shared_provenance} (see :func:zettelkasten.synapse.overlay.build_connections). This is safe today: no overlay consumer reads persisted signals (get_connections / high_conf_neighbors only touch relation/confidence/ids). A FUTURE consumer that reads signals must tolerate shared_provenance being absent (replaced by shared_provenance_count) in a bundle-materialized overlay.

Decrypt is a FULL REBUILD

The federation cache is EXCLUSIVELY bundle-owned, so each sync rebuilds <cache_root>/.synapse/links/links.json from exactly the edges this key can open — a revoked edge simply doesn't reappear next sync (forward-only revocation, same accepted limitation as the parents: a bundle repo's git history retains old ciphertexts). The rebuild is atomic and fail-safe: a transient read fault aborts the rebuild WITHOUT touching the existing cache, so a blip never corrupts or erases a good overlay.

Publish reflects each peer policy AS OF the last federation sync

build_bundle (synapse publish) resolves a FEDERATED edge's ZK-end recipients from the peer's DECRYPTED recipient-index policy already sitting in the federation cache — it never re-fetches the peer's export itself. So a cross-producer edge reflects each peer's recipient policy exactly AS OF the consumer's LAST angelo federation sync. Revocation is therefore FORWARD-ONLY and doubly gated: a producer's revocation propagates only after the consumer (1) re-syncs (which re-decrypts / removes the peer policy) AND (2) re-publishes off the refreshed cache. Publish should follow sync: run angelo federation sync immediately before synapse publish so the edges you ship reflect current grants. As an observability aid (NOT a security control), publish emits a NON-FATAL stderr WARNING — via :func:load_recipient_index_stamp — when a federated edge relies on a peer policy whose freshness stamp is missing or older than :data:_STALE_POLICY_WARN_AFTER; it never drops an edge and never enforces a hard TTL. The warning tells the operator which peer policy is stale and to re-sync.

BundleUnit dataclass

A pointer to one encrypted edge unit (kind + relative path).

Source code in zettelkasten/synapse/sharing/bundle.py

@dataclass(frozen=True)
class BundleUnit:
    """A pointer to one encrypted edge unit (``kind`` + relative path)."""

    kind: str
    path: str

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

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

BundleManifest dataclass

The cleartext manifest.json at the bundle root.

Source code in zettelkasten/synapse/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 zettelkasten/synapse/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
    edge_unit_count: int
    recipient_ids: tuple[str, ...]
    skipped_unregistered_ids: tuple[str, ...]

DecryptResult dataclass

Summary of a :func:decrypt_bundle run.

Source code in zettelkasten/synapse/sharing/bundle.py

@dataclass(frozen=True)
class DecryptResult:
    """Summary of a :func:`decrypt_bundle` run."""

    output_dir: Path
    edge_count: int
    total_units: int

StalePeerPolicyError

Bases: RuntimeError

Raised by :func:build_bundle under the OPT-IN fail-closed staleness TTL.

When max_policy_age_hours is configured, a synapse publish REFUSES to ship cross-producer edges whose peer recipient-index policy is missing or older than the TTL — a "publish-requires-recent-sync" gate — rather than emitting a possibly over-permissive (stale) edge set. It is raised BEFORE any bundle output is written, so an existing bundle is left intact on failure.

stale_peers is a list of (peer_id, decrypted_at_iso_or_None) for programmatic handling; None means the peer's freshness stamp was missing/unparseable.

Source code in zettelkasten/synapse/sharing/bundle.py

class StalePeerPolicyError(RuntimeError):
    """Raised by :func:`build_bundle` under the OPT-IN fail-closed staleness TTL.

    When ``max_policy_age_hours`` is configured, a synapse publish REFUSES to ship
    cross-producer edges whose peer recipient-index policy is missing or older than
    the TTL — a "publish-requires-recent-sync" gate — rather than emitting a
    possibly over-permissive (stale) edge set. It is raised BEFORE any bundle
    output is written, so an existing bundle is left intact on failure.

    ``stale_peers`` is a list of ``(peer_id, decrypted_at_iso_or_None)`` for
    programmatic handling; ``None`` means the peer's freshness stamp was
    missing/unparseable.
    """

    def __init__(self, message: str, stale_peers: list[tuple[str, str | None]]):
        super().__init__(message)
        self.stale_peers = stale_peers

read_manifest

read_manifest(bundle_dir: Path | str) -> BundleManifest

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

Source code in zettelkasten/synapse/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, memory_dir: Path | str | None = None, store_root: Path | str | None = None, overlay: Mapping[str, Any] | None = None, producer_id: str | None = None, federation_cache_root: Path | str | None = None, max_policy_age_hours: float | None = None) -> BuildResult

Filter synapse edges by both-ends grants, encrypt per recipient, write a bundle.

Pipeline (filtering happens BEFORE any encryption):

1. Load the local .synapse overlay edges (overlay arg, else :func:zettelkasten.synapse.overlay.load_overlay).
2. Build the memory entry_recipients and ZK note_recipients maps for the local workspace (reusing the two stores' sharing config loaders, firm expanded to the registry members). For any FEDERATED edge, also load the referenced peer's decrypted recipient-index policy from the federation cache (federation_cache_root, default the workspace cache).
3. For each edge compute :func:edge_recipients — the INTERSECTION of its two endpoints' recipient sets (the ZK end resolved locally for a same-producer edge, or from the peer policy for a federated one). OMIT the edge when it resolves to no recipient (either side missing/ungranted, a federated endpoint with no resolvable peer policy, or an empty intersection) or when none of its recipients is registered. Never encrypt-then-drop.
4. Encrypt each retained edge's placement-header envelope to its recipients' public keys and write units/<digest>.age.
5. Write the cleartext manifest.json of {kind, path} pointers only.

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

max_policy_age_hours is the OPT-IN fail-closed staleness TTL (default None = disabled). When set to a positive number and any cross-producer edge is present, this raises :class:StalePeerPolicyError — BEFORE any output is written — if a referenced peer's recipient policy is missing or older than the TTL, refusing to publish a possibly over-permissive stale edge set. Left unset, the default is unchanged: publish still ships (forward-only) and only emits a non-fatal staleness warning.

Source code in zettelkasten/synapse/sharing/bundle.py

def build_bundle(
    output_dir: Path | str,
    *,
    registry: IdentityRegistry,
    memory_dir: Path | str | None = None,
    store_root: Path | str | None = None,
    overlay: Mapping[str, Any] | None = None,
    producer_id: str | None = None,
    federation_cache_root: Path | str | None = None,
    max_policy_age_hours: float | None = None,
) -> BuildResult:
    """Filter synapse edges by both-ends grants, encrypt per recipient, write a bundle.

    Pipeline (filtering happens BEFORE any encryption):

    1. Load the local ``.synapse`` overlay edges (``overlay`` arg, else
       :func:`zettelkasten.synapse.overlay.load_overlay`).
    2. Build the memory ``entry_recipients`` and ZK ``note_recipients`` maps for
       the local workspace (reusing the two stores' sharing config loaders,
       ``firm`` expanded to the registry members). For any FEDERATED edge, also
       load the referenced peer's decrypted recipient-index policy from the
       federation cache (``federation_cache_root``, default the workspace cache).
    3. For each edge compute :func:`edge_recipients` — the INTERSECTION of its
       two endpoints' recipient sets (the ZK end resolved locally for a
       same-producer edge, or from the peer policy for a federated one). OMIT the
       edge when it resolves to no recipient (either side missing/ungranted, a
       federated endpoint with no resolvable peer policy, or an empty
       intersection) or when none of its recipients is registered. Never
       encrypt-then-drop.
    4. Encrypt each retained edge's placement-header envelope to its recipients'
       public keys and write ``units/<digest>.age``.
    5. Write the cleartext ``manifest.json`` of ``{kind, path}`` pointers only.

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

    ``max_policy_age_hours`` is the OPT-IN fail-closed staleness TTL (default
    ``None`` = disabled). When set to a positive number and any cross-producer edge
    is present, this raises :class:`StalePeerPolicyError` — BEFORE any output is
    written — if a referenced peer's recipient policy is missing or older than the
    TTL, refusing to publish a possibly over-permissive stale edge set. Left unset,
    the default is unchanged: publish still ships (forward-only) and only emits a
    non-fatal staleness warning.
    """
    from memory import storage

    output_dir = Path(output_dir)
    memory_dir = Path(memory_dir) if memory_dir is not None else storage.MEMORY_DIR
    if store_root is not None:
        store_root = Path(store_root)
    else:
        from zettelkasten.graph import GRAPHS_DIR
        store_root = GRAPHS_DIR

    if overlay is None:
        from zettelkasten.synapse import overlay as overlay_mod
        overlay = overlay_mod.load_overlay()
    edges = overlay.get("edges", []) if isinstance(overlay, Mapping) else []

    entry_recip = _memory_entry_recipients(memory_dir, registry)
    note_recip = _zk_note_recipients_map(store_root, registry)

    # Cross-producer edges resolve their ZK-end recipients from peer policies in
    # the federation cache (fail-closed if absent). Only touch the cache when a
    # federated edge is actually present, so the same-producer path is unchanged.
    has_federated = any(
        isinstance(e, Mapping) and ":" in str(e.get("zk_source") or "")
        for e in edges
    )
    peer_note_recip: dict[str, set[str]] = {}
    if has_federated:
        if federation_cache_root is not None:
            cache_root = Path(federation_cache_root)
        else:
            from zettelkasten.federation import (
                federation_cache_root as _fed_cache_root,
            )
            cache_root = _fed_cache_root()
        peer_note_recip = _peer_note_recipients(edges, cache_root)
        if max_policy_age_hours is not None and max_policy_age_hours > 0:
            # Opt-in HARD gate: refuse to publish on a stale/absent peer policy.
            # Raised BEFORE units/ is rebuilt (below), so an existing bundle is
            # left intact on failure. Deliberately NOT swallowed — aborting is the
            # whole point of the fail-closed TTL.
            _enforce_policy_ttl(
                edges, cache_root, timedelta(hours=float(max_policy_age_hours))
            )
        else:
            # Default: observability nudge (never fatal, never drops an edge). The
            # helper is purely diagnostic, so guard the call site: any unexpected
            # error inside it must NEVER propagate out of build_bundle / abort a
            # publish. Publish correctness must not depend on this observability aid.
            try:
                _warn_stale_peer_policies(edges, cache_root)
            except Exception as exc:  # pragma: no cover - defensive
                print(
                    "[synapse publish] warning: stale-peer-policy diagnostic failed "
                    f"({exc}); continuing publish (this is observability only).",
                    file=sys.stderr,
                )

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

    skipped_unregistered: set[str] = set()

    def _pubkeys_for(recips: set[str]) -> list[str]:
        keys: list[str] = []
        for rid in sorted(recips):
            pk = registry.public_key_for(rid)
            if pk:
                keys.append(pk)
            else:
                skipped_unregistered.add(rid)
        return keys

    units: list[BundleUnit] = []
    all_recipient_ids: set[str] = set()
    edge_count = 0

    for edge in edges:
        if not isinstance(edge, Mapping):
            continue
        recips = edge_recipients(edge, entry_recip, note_recip, peer_note_recip)
        if not recips:
            continue  # no both-ends recipient / unresolvable peer policy — omit
        pubkeys = _pubkeys_for(set(recips))
        if not pubkeys:
            logger.warning(
                "synapse edge %s:%s:%s is shared but no recipient is registered; omitting",
                edge.get("memory_id"), edge.get("zk_source"), edge.get("zk_id"),
            )
            continue
        memory_id = str(edge.get("memory_id") or "").strip()
        zk_source = str(edge.get("zk_source") or "").strip()
        zk_id = str(edge.get("zk_id") or "").strip()
        unit_key = f"{memory_id}:{zk_source}:{zk_id}"
        header = {
            "kind": KIND_SYNAPSE_EDGE,
            "memory_id": memory_id,
            "zk_source": zk_source,
            "zk_id": zk_id,
        }
        payload = json.dumps(
            _shareable_edge_payload(edge), sort_keys=True
        ).encode("utf-8")
        ciphertext = crypto.encrypt(_encode_envelope(header, payload), pubkeys)
        filename = _unit_filename(unit_key)
        atomic_write(units_dir / filename, ciphertext)
        units.append(BundleUnit(kind=KIND_SYNAPSE_EDGE, path=f"{UNITS_DIRNAME}/{filename}"))
        edge_count += 1
        all_recipient_ids.update(
            rid for rid in recips if registry.public_key_for(rid)
        )

    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,
        edge_unit_count=edge_count,
        recipient_ids=tuple(sorted(all_recipient_ids)),
        skipped_unregistered_ids=tuple(sorted(skipped_unregistered)),
    )

decrypt_bundle

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

Decrypt every edge this key can open and REBUILD a .synapse overlay cache.

Edges are decrypted and their records collected, then a valid overlay file is written to <out_root>/.synapse/links/links.json — a federation cache that is ENTIRELY bundle-owned, so a full rebuild-from-decrypted is correct: a revoked edge (dropped from a newer bundle, or no longer openable by this key) simply does not reappear. The rebuild is atomic.

Durability: a MISSING/CORRUPT manifest is treated as an empty bundle (the overlay is rebuilt to zero edges — full revocation). An UNREADABLE manifest, or ANY transient I/O fault reading a unit, ABORTS the rebuild WITHOUT touching the existing cache — a transient blip must never corrupt or erase a good overlay; the next clean sync rebuilds normally.

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

Source code in zettelkasten/synapse/sharing/bundle.py

def decrypt_bundle(
    bundle_dir: Path | str,
    out_root: Path | str,
    *,
    identity: LocalIdentity | Any | None = None,
    private_key: str | None = None,
) -> DecryptResult:
    """Decrypt every edge this key can open and REBUILD a ``.synapse`` overlay cache.

    Edges are decrypted and their records collected, then a valid overlay file is
    written to ``<out_root>/.synapse/links/links.json`` — a federation cache that
    is ENTIRELY bundle-owned, so a full rebuild-from-decrypted is correct: a
    revoked edge (dropped from a newer bundle, or no longer openable by this key)
    simply does not reappear. The rebuild is atomic.

    Durability: a MISSING/CORRUPT manifest is treated as an empty bundle (the
    overlay is rebuilt to zero edges — full revocation). An UNREADABLE manifest,
    or ANY transient I/O fault reading a unit, ABORTS the rebuild WITHOUT touching
    the existing cache — a transient blip must never corrupt or erase a good
    overlay; the next clean sync rebuilds normally.

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

    try:
        manifest = read_manifest(bundle_dir)
    except FileNotFoundError as exc:
        logger.warning("bundle manifest absent (%s); rebuilding overlay as empty", exc)
        manifest = BundleManifest(format="", format_version="", created_at="", units=())
    except (json.JSONDecodeError, ValueError) as exc:
        logger.warning("bundle manifest corrupt (%s); rebuilding overlay as empty", exc)
        manifest = BundleManifest(format="", format_version="", created_at="", units=())
    except OSError as exc:
        logger.warning(
            "bundle manifest unreadable due to transient I/O (%s); aborting decrypt "
            "without touching the cache to preserve the valid overlay", exc
        )
        return DecryptResult(output_dir=overlay_path.parent, edge_count=0, total_units=0)

    edges: list[dict[str, Any]] = []
    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():
            # Genuinely gone from disk (e.g. revoked): a full rebuild correctly
            # omits it. Not a transient fault.
            logger.warning("bundle unit missing on disk: %s", unit_path)
            continue
        try:
            ciphertext = unit_path.read_bytes()
        except OSError as exc:
            # Present but unreadable — a transient I/O fault. Rebuilding now could
            # drop a still-valid edge and clobber a good overlay, so abort the
            # rebuild entirely this run (preserve the cache) and retry next sync.
            logger.warning(
                "could not read bundle unit %s (transient I/O): %s; "
                "aborting overlay rebuild this run to preserve the cache",
                unit_path, exc,
            )
            transient_io_error = True
            break
        try:
            plaintext = crypto.decrypt(ciphertext, key)
        except crypto.DecryptError:
            continue  # not encrypted to this key
        except Exception as exc:  # pragma: no cover - defensive
            logger.warning("failed to decrypt unit %s: %s", unit.path, exc)
            continue

        decoded = _decode_envelope(plaintext)
        if decoded is None:
            logger.warning("decrypted unit %s has a malformed envelope; skipping", unit.path)
            continue
        header, payload = decoded
        placement = _valid_placement(header)
        if placement is None:
            logger.warning("decrypted unit %s has an invalid placement; skipping", unit.path)
            continue
        memory_id, zk_source, zk_id = placement
        try:
            edge = json.loads(payload.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            logger.warning("decrypted unit %s has a malformed edge payload; skipping", unit.path)
            continue
        if not isinstance(edge, dict):
            continue
        # Trust the validated placement header for the id-bearing fields so a
        # rebuilt edge always carries consistent, contained ids.
        edge["memory_id"] = memory_id
        edge["zk_source"] = zk_source
        edge["zk_id"] = zk_id
        edges.append(edge)

    if transient_io_error:
        return DecryptResult(output_dir=overlay_path.parent, edge_count=0, total_units=len(manifest.units))

    overlay = {
        "version": OVERLAY_FORMAT_VERSION,
        "manifest": {
            "source": BUNDLE_FORMAT,
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "edge_count": len(edges),
        },
        "edges": edges,
    }
    atomic_write(overlay_path, json.dumps(overlay, indent=2, ensure_ascii=False).encode("utf-8"))

    return DecryptResult(
        output_dir=overlay_path.parent,
        edge_count=len(edges),
        total_units=len(manifest.units),
    )

open_bundle

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

Alias for :func:decrypt_bundle.

Source code in zettelkasten/synapse/sharing/bundle.py

def open_bundle(
    bundle_dir: Path | str,
    out_root: Path | str,
    *,
    identity: LocalIdentity | Any | None = None,
    private_key: str | None = None,
) -> DecryptResult:
    """Alias for :func:`decrypt_bundle`."""
    return decrypt_bundle(bundle_dir, out_root, identity=identity, private_key=private_key)