memory.sharing.projection

memory.sharing.projection

Per-caller PLAINTEXT projection of a shared memory store.

The Iceberg hosted-federation broker (see documents/260708_iceberg-revocation-and-binding.md, Part D) owns the plaintext store and must hand each authenticated caller a .memory-shaped view containing ONLY the units that caller is entitled to see — the same filtering and reference-scrubbing a shared bundle applies, but WITHOUT the encrypt/decrypt round-trip (and without ever holding the caller's private key).

:func:project_for_caller produces exactly that. It reuses :func:memory.sharing.bundle._collect_shared_units — the single source of truth for the filter + per-unit scrub that :func:memory.sharing.bundle.build_bundle also uses — so the result satisfies the PARITY CONTRACT:

project_for_caller(caller) == build_bundle(...) then
decrypt_bundle(..., private_key=caller_key)

i.e. a caller's plaintext projection is byte-for-byte the SAME set of units, and the SAME scrubbed content, that caller would obtain by decrypting the encrypted bundle.

Why the per-unit SUPERSET scrub — not a per-caller "everything I can see" set

It is tempting to scrub each projected unit against {x : caller in recipients(x)} (everything the caller can see). That is WRONG and breaks parity. build_bundle scrubs each unit exactly ONCE, against allowed_ids = the bundled units whose recipient set is a SUPERSET of THAT unit's recipient set. That scrub is caller-independent and is baked into the single ciphertext every recipient of the unit decrypts. So when a caller decrypts the bundle they see each unit scrubbed to that SUPERSET set — which can be strictly NARROWER than the full set of units the caller can see elsewhere.

Concretely: a session shared to {alice, bob} keeps only refs whose recipient set is a superset of {alice, bob} (e.g. a firm-wide entry). Alice's decrypted copy therefore omits an alice-only ref even though alice can open that entry in her own unit. Scrubbing the projection with the caller's global visible set would RETAIN that ref and diverge from the decrypt. Reusing the shared collector (which applies the superset rule per unit, once) guarantees identical scrubbed bytes on both paths — so the projection and the decrypt agree byte-for-byte.

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

@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

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

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