`memory.sharing.config`¶

Grants / sharing-scope config parsing and shareability resolution.

This module reads the owner-authored sharing policy from .memory/config.yaml (loaded via :func:memory.storage.read_config, a plain dict) and turns it into typed accessors. It is the policy half of firm-wide sharing: it decides which memory entries are shareable to which recipient id. It performs no crypto and no I/O beyond parsing an already-loaded dict.

Two independent, additive mechanisms combine to decide shareability:

sharing.scopes — a visibility scope attached to a subtree node (a subproject/phase node id). The scope is inherited by every descendant entry and may be overridden on a deeper node (or the entry itself). Vocabulary:
private (the default) — shared with nobody.
firm — shared with every registered firm member.
bilateral— shared with an explicit audience list of recipient ids.
grants — an owner-authored node_id -> [recipient_ids] map that grants specific recipients access to a node and its descendants, regardless of scope.

An entry is shareable to a recipient if EITHER the effective scope admits them (firm for anyone, bilateral if listed) OR an ancestor grant lists them.

Fully additive / opt-in: when both keys are absent the resolver returns the empty set — i.e. today's behaviour (nothing is shared). No existing config is affected.

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

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

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

@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() -> bool

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

Source code in memory/sharing/config.py

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
    )

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

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

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

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)

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

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_transport(entry: Mapping[str, Any], entries_by_id: Mapping[str, Mapping[str, Any]], sharing: SharingConfig) -> str

Resolve the transport tier for entry after subtree inheritance.

Returns offline or hosted. Parallels :func:effective_scope: it walks the same ancestor chain and honours inheritance identically, so a child under a hosted ancestor is hosted unless it (or a nearer ancestor) overrides back to offline. Defaults to :data:TRANSPORT_OFFLINE for an entry with no transport anywhere on its chain (deny-by-default).

Source code in memory/sharing/config.py

def effective_transport(
    entry: Mapping[str, Any],
    entries_by_id: Mapping[str, Mapping[str, Any]],
    sharing: SharingConfig,
) -> str:
    """Resolve the transport tier for ``entry`` after subtree inheritance.

    Returns ``offline`` or ``hosted``. Parallels :func:`effective_scope`: it
    walks the same ancestor chain and honours inheritance identically, so a child
    under a ``hosted`` ancestor is ``hosted`` unless it (or a nearer ancestor)
    overrides back to ``offline``. Defaults to :data:`TRANSPORT_OFFLINE` for an
    entry with no transport anywhere on its chain (deny-by-default).
    """
    entry_id = str(entry.get("id") or "")
    if not entry_id:
        return TRANSPORT_OFFLINE
    return _transport_for_chain(_ancestor_ids(entry_id, entries_by_id), sharing)

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

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

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

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

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

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

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

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

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

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

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

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

memory.sharing.config¶

memory.sharing.config ¶

Scope dataclass ¶

SharingConfig dataclass ¶

is_empty ¶

parse_sharing_config ¶

effective_scope ¶

effective_grants ¶

effective_transport ¶

is_shareable_to ¶

shareable_entries ¶

entry_recipients ¶

document_recipients ¶

session_recipients ¶

skill_recipients ¶

`memory.sharing.config`¶

Scope `dataclass` ¶

SharingConfig `dataclass` ¶