def project_for_caller(
caller_id: str,
*,
registry: IdentityRegistry,
out_root: Path | str,
store_root: Path | str | None = None,
sharing: SharingConfig | None = None,
allowed_transports: set[str] | None = None,
) -> ProjectionResult:
"""Materialize the PLAINTEXT units ``caller_id`` may open under ``out_root``.
Writes a ``.zettelkasten``-shaped plaintext store into
``<out_root>/.zettelkasten/`` — the SAME layout
:func:`~zettelkasten.sharing.bundle.decrypt_bundle` produces — containing only
the units ``caller_id`` is entitled to, with references scrubbed exactly as
:func:`~zettelkasten.sharing.bundle.build_bundle` scrubs them.
``store_root`` defaults to the local ``.zettelkasten`` store and ``sharing``
to the policy parsed from ``<store_root>/config.yaml`` (both resolved inside
the shared :func:`~zettelkasten.sharing.bundle._collect_pending`).
``allowed_transports`` restricts the projection to units of a given transport
tier (design decision D3). The hosted-federation broker passes ``{"hosted"}``
so it serves ONLY hosted units — structurally, a unit whose effective
transport is ``offline`` can never be projected to a broker caller, so an
offline-only store is never served over the network. The default ``None``
applies no transport filter (every entitled unit is projected), preserving
the pre-transport-tier behaviour and keeping the projection byte-identical to
an offline ``build_bundle`` + ``decrypt_bundle`` when the store has no
transport annotations.
This function OWNS ``<out_root>/.zettelkasten`` and RECONCILES it on every
call, exactly like :func:`~zettelkasten.sharing.bundle.decrypt_bundle`: a
re-projection (e.g. after a grant change) surgically evicts only the units this
projection previously produced that the caller is no longer entitled to, while
PRESERVING foreign/unowned files and honoring the same transient-I/O guard
(a read glitch skips eviction rather than destroying a valid cache). The result
matches a fresh decrypt of the caller's CURRENT grants even when projecting
into a previously-populated store.
"""
out_root = Path(out_root)
zettel_dir = out_root / ".zettelkasten"
# Enumerate + recipient-resolve + filter — IDENTICAL to the bundle path
# (save the transport tier: the broker passes ``{"hosted"}`` to serve only
# hosted units; ``None`` projects every entitled unit, unchanged behaviour).
pending, bundled_recip, _skipped = _collect_pending(
store_root, sharing, registry, allowed_transports
)
# A caller opens exactly the units its key can DECRYPT. build_bundle encrypts
# each unit to a set of age PUBLIC KEYS, so the caller opens a unit iff its own
# resolvable public key is among that unit's ``pubkeys`` — NOT merely iff its
# registry id appears in the recipient set. Selecting by public key (rather
# than id) keeps projection == decrypt even when two distinct registry ids
# resolve to the SAME age key: an id-based check could withhold a unit the
# caller's key CAN decrypt (or project one it cannot), diverging from the
# decrypt. An unregistered caller (no resolvable key) opens nothing — parity
# with a decrypt that opens no unit.
caller_pubkey = registry.public_key_for(caller_id)
# ---- Phase 1: select this caller's units, BUFFERING writes -------------
# Mirror decrypt_bundle: buffer ``(dest, payload, ident)`` per kind and record
# the destination names/paths we will own (``seen_*``), so Phase 2 can evict
# ONLY the stale units this projection previously produced — never foreign
# drop-ins. Writing happens after reconcile so eviction+write is one pass.
note_writes: list[tuple[str, Path, bytes, str]] = [] # (box, dest, payload, ident)
meta_writes: list[tuple[str, Path, bytes, str]] = [] # (box, dest, payload, ident)
project_writes: list[tuple[Path, bytes, str]] = []
citation_writes: list[tuple[Path, bytes, str]] = []
review_writes: list[tuple[Path, bytes, str]] = []
org_writes: list[tuple[Path, bytes, str]] = []
table_writes: list[tuple[Path, bytes, str]] = []
outline_writes: list[tuple[Path, bytes, str]] = []
sidecar_writes: list[tuple[Path, bytes, str]] = []
seen_notes: dict[str, set[str]] = {} # box -> {filename}
seen_meta_boxes: set[str] = set()
seen_projects: set[str] = set()
seen_citations: set[str] = set()
seen_reviews: set[str] = set() # <name>.yaml
seen_tables: set[str] = set() # <name>.tables.json
seen_outlines: set[str] = set() # <name>.outlines.json
seen_org_paths: set[Path] = set() # resolved <id>.json dest paths
if caller_pubkey is not None:
for _kind, _unit_key, header, payload, _recips, pubkeys in _scrub_pending(
pending, bundled_recip
):
# A unit is decryptable by this caller iff their resolved public key is
# among the unit's recipient keys (the SAME ``pubkeys`` build_bundle
# encrypts to) — key-based, matching decrypt_bundle exactly. Using the
# SUPERSET-scrubbed ``payload`` (not a caller-relative re-scrub) is what
# makes the plaintext byte-identical to what the caller would decrypt
# from the bundle.
if caller_pubkey not in pubkeys:
continue
dest, kind, ident = _unit_dest(header, zettel_dir)
if dest is None:
# A malformed/unsafe placement is skipped exactly as decrypt does.
logger.warning("skipping unit with no usable placement: %r", header)
continue
if kind == KIND_NOTE:
box = str(header.get("box"))
note_writes.append((box, dest, payload, ident))
seen_notes.setdefault(box, set()).add(dest.name)
elif kind == KIND_SOURCE_META:
box = str(header.get("box"))
meta_writes.append((box, dest, payload, ident))
seen_meta_boxes.add(box)
elif kind == KIND_PROJECT:
project_writes.append((dest, payload, ident))
seen_projects.add(dest.name)
elif kind == KIND_CITATION:
citation_writes.append((dest, payload, ident))
seen_citations.add(dest.name)
elif kind == KIND_REVIEW:
review_writes.append((dest, payload, ident))
seen_reviews.add(dest.name)
elif kind == KIND_TABLE:
table_writes.append((dest, payload, ident))
seen_tables.add(dest.name)
elif kind == KIND_OUTLINE:
outline_writes.append((dest, payload, ident))
seen_outlines.add(dest.name)
elif kind == KIND_ORG:
org_writes.append((dest, payload, ident))
seen_org_paths.add(dest.resolve())
elif kind == KIND_SIDECAR:
sidecar_writes.append((dest, payload, ident))
# ---- Phase 2: reconcile (evict stale) then write (additive) -----------
# This is decrypt_bundle's Phase 2 verbatim (minus the manifest/ciphertext
# bits it does not apply here), so a projection and a decrypt reconcile a
# populated store IDENTICALLY. The scoped store is EXCLUSIVELY produced by this
# projection, so notes are evicted by filename (like memory's ``entries/``),
# while the shared ``_projects``/``_citations``/``_reviews``/``_organizations``
# dirs evict only well-formed owned units (foreign drop-ins preserved). Binary/
# dataset sidecars carry no self-identifying frontmatter, so their ownership is
# tracked by the ``.owned-sidecars.json`` provenance record.
current_shipped = {ident for _d, _p, ident in sidecar_writes}
# Load the prior owned-sidecars provenance record BEFORE reconciling. A
# transient read fault aborts ALL eviction this run (never deletes valid cached
# plaintext); a MISSING/corrupt record yields an empty prior set (evict nothing
# this run, just start recording).
transient_io_error = False
prior_owned, record_ok = _load_owned_sidecars(zettel_dir)
if not record_ok:
transient_io_error = True
# Eviction is gated on a clean pass so a transient listing fault never deletes
# valid cached plaintext (revocation still works on any later clean pass).
if not transient_io_error:
try:
boxes_to_reconcile = (
set(seen_notes) | seen_meta_boxes | set(_note_graph_dirs(zettel_dir))
)
for box in boxes_to_reconcile:
try:
box_dir = safe_join(zettel_dir, box)
except ValueError:
continue
if not box_dir.is_dir():
continue
keep = seen_notes.get(box, set())
for existing in box_dir.glob("*.md"):
if existing.name.startswith("_") or existing.name in keep:
continue
_unlink_quietly(existing)
# A box's _meta.yaml is stale when this box has no shared meta now.
if box not in seen_meta_boxes:
_unlink_quietly(box_dir / "_meta.yaml")
_reconcile_dir(zettel_dir / "_projects", seen_projects, KIND_PROJECT)
_reconcile_dir(zettel_dir / "_citations", seen_citations, KIND_CITATION)
reviews_dir = zettel_dir / "_reviews"
if reviews_dir.is_dir():
for existing in reviews_dir.glob("*.yaml"):
if existing.name not in seen_reviews and _is_owned_unit(existing, KIND_REVIEW):
_unlink_quietly(existing)
for existing in reviews_dir.glob("*.tables.json"):
if existing.name not in seen_tables and _is_owned_unit(existing, KIND_TABLE):
_unlink_quietly(existing)
for existing in reviews_dir.glob("*.outlines.json"):
if existing.name not in seen_outlines and _is_owned_unit(existing, KIND_OUTLINE):
_unlink_quietly(existing)
_reconcile_orgs(zettel_dir / "_organizations", seen_org_paths)
except OSError as exc:
logger.warning(
"cache listing failed during projection reconcile (transient I/O): "
"%s; skipping eviction and record persist this run to preserve valid "
"cached plaintext (revocation retries on a later clean pass)", exc
)
transient_io_error = True
# Evict binary/dataset sidecars by PROVENANCE (not shape): exactly the paths we
# materialized before that the caller no longer receives. A path we never
# produced (a DVC-pulled data file, a foreign drop-in) is never in
# ``prior_owned`` and so is always preserved. All-or-nothing under a fault.
if not transient_io_error:
for stale_ident in prior_owned - current_shipped:
try:
stale_path = safe_join(zettel_dir, *stale_ident.split("/"))
except ValueError:
continue
_unlink_quietly(stale_path)
# Materialize the caller's current units (additive). ``atomic_write`` creates
# parent dirs, so an empty projection writes nothing (no dir churn).
for _box, dest, payload, _ident in note_writes:
atomic_write(dest, payload)
for _box, dest, payload, _ident in meta_writes:
atomic_write(dest, payload)
for dest, payload, _ident in project_writes:
atomic_write(dest, payload)
for dest, payload, _ident in citation_writes:
atomic_write(dest, payload)
for dest, payload, _ident in review_writes:
atomic_write(dest, payload)
for dest, payload, _ident in table_writes:
atomic_write(dest, payload)
for dest, payload, _ident in outline_writes:
atomic_write(dest, payload)
for dest, payload, _ident in org_writes:
atomic_write(dest, payload)
for dest, payload, _ident in sidecar_writes:
atomic_write(dest, payload)
# Persist this run's owned-sidecar provenance so the NEXT projection evicts
# exactly what this run stops shipping. Only on a clean pass; an empty set
# removes the record (no spurious empty cache dir).
if not transient_io_error:
_persist_owned_sidecars(zettel_dir, current_shipped)
note_keys = [ident for _b, _d, _p, ident in note_writes]
source_boxes = [ident for _b, _d, _p, ident in meta_writes]
project_names = [ident for _d, _p, ident in project_writes]
citation_ids = [ident for _d, _p, ident in citation_writes]
review_names = [ident for _d, _p, ident in review_writes]
org_ids = [ident for _d, _p, ident in org_writes]
table_reviews = [ident for _d, _p, ident in table_writes]
outline_reviews = [ident for _d, _p, ident in outline_writes]
sidecar_paths = [ident for _d, _p, ident in sidecar_writes]
total = (
len(note_keys) + len(source_boxes) + len(project_names) + len(citation_ids)
+ len(review_names) + len(org_ids) + len(table_reviews) + len(outline_reviews)
+ len(sidecar_paths)
)
return ProjectionResult(
caller_id=caller_id,
output_dir=zettel_dir,
note_keys=tuple(note_keys),
source_boxes=tuple(source_boxes),
project_names=tuple(project_names),
citation_ids=tuple(citation_ids),
review_names=tuple(review_names),
org_ids=tuple(org_ids),
table_reviews=tuple(table_reviews),
outline_reviews=tuple(outline_reviews),
sidecar_paths=tuple(sidecar_paths),
note_unit_count=len(note_keys),
source_unit_count=len(source_boxes),
project_unit_count=len(project_names),
citation_unit_count=len(citation_ids),
review_unit_count=len(review_names),
org_unit_count=len(org_ids),
table_unit_count=len(table_reviews),
outline_unit_count=len(outline_reviews),
sidecar_unit_count=len(sidecar_paths),
total_units=total,
transient_io_error=transient_io_error,
)