symfonic.capabilities.knowledge.contracts

contracts

The ingested-context contract: what a knowledge bridge is allowed to declare.

This is a deliberate narrowing of T3.2.1's prompt contribution contract, not a copy of it. The prompting capability accepts contributions at any tier because some of them are the operator's own instructions; this capability only ever carries content that came from a store, a document, or a file, and AS-ING-1 says that content is untrusted by definition. So three declarations the prompt contract permits are refused here — authored tier, the L0 prefix, and pinning — and the refusal is a constructor check rather than a review note.

The two contracts are kept aligned by vocabulary, not by an import. A capability may import the kernel's contracts and its own package; two capabilities that import each other are one capability wearing two names. The enum values here are the same strings the prompt contract uses, and :mod:.assembly emits a spec the composition root feeds to the compiler.

ContextContribution dataclass

ContextContribution(contribution_id: str, source: ContextSource, capability: str = 'knowledge', layer: ContextLayer = ContextLayer.L2, tier: ContextTier = ContextTier.SESSION, scope: ContextScope = ContextScope.DEPLOYMENT, order: int = 0, inherit: bool = True, pinned: bool = False, requires_hydration: bool = True)

One bridge's declaration of ingested context it contributes.

validate

validate() -> None

Refuse every declaration this capability is not allowed to make.

Source code in src/symfonic/capabilities/knowledge/contracts.py

def validate(self) -> None:
    """Refuse every declaration this capability is not allowed to make."""
    if not self.contribution_id:
        raise BridgeContractError(
            "an ingested-context contribution must declare a non-empty contribution_id."
        )
    if not _ID_CHARSET.match(self.contribution_id):
        raise BridgeContractError(
            f"contribution id {self.contribution_id!r} is outside the permitted charset "
            "[A-Za-z0-9_.-]; ids appear in isolation keys and in the rendered untrusted "
            "delimiter, where a separator or an angle bracket forges a boundary."
        )
    if not callable(getattr(self.source, "read", None)):
        raise BridgeContractError(
            f"contribution {self.contribution_id!r} declares a source that cannot read: "
            f"{type(self.source).__name__} has no callable read()."
        )
    if self.tier in AUTHORED_TIERS:
        raise BridgeContractError(
            f"contribution {self.contribution_id!r} declares the authored tier "
            f"{self.tier.value!r}. Retrieved, stored, and extracted content is untrusted "
            "by definition (AS-ING-1) and renders as data; an authored tier renders it "
            "verbatim beside the operator's own instructions."
        )
    if self.layer is ContextLayer.L0:
        raise BridgeContractError(
            f"contribution {self.contribution_id!r} declares layer L0. L0 is the cached "
            "authored prefix: ingested content placed there is served back to the model "
            "on every later turn of the session, so one poisoned retrieval outlives the "
            "turn that fetched it. Declare L1 (standing) or L2 (per turn)."
        )
    if self.pinned:
        raise BridgeContractError(
            f"contribution {self.contribution_id!r} declares itself pinned. Pinned content "
            "fails the compile rather than being dropped for budget; ingested content is "
            "evidence, not instruction, and a turn without it is still a correct turn."
        )
    if self.scope is not ContextScope.DEPLOYMENT and not getattr(
        self.source, "scope_aware", False
    ):
        raise BridgeContractError(
            f"contribution {self.contribution_id!r} declares scope={self.scope.value!r} "
            f"but its source {type(self.source).__name__} is not scope_aware: it serves "
            "one value for every tenant. Declare scope='deployment', or supply a source "
            "that keys on the scope path."
        )

ContextContributor

Bases: Protocol

What a knowledge-side capability registers with the composition root.

ContextLayer

Bases: StrEnum

The stratigraphic layer a contribution renders on.

ContextRead dataclass

ContextRead(text: str, revision: str = '', untrusted: bool = True)

What an ingested-context source answered.

untrusted defaults to True — the inverse of the prompt contract's default, and the whole point of a separate value type. In the general contract a source must remember to declare its payload untrusted; here it would have to remember to declare it trusted, which nothing in this capability ever does.

ContextRequest dataclass

ContextRequest(contribution_id: str, scope_path: str = '', turn: int = 0)

What a source is asked for: one contribution, one scope, one turn.

ContextScope

Bases: StrEnum

How widely one contribution's content is shared.

ContextSource

Bases: Protocol

Reads the current ingested content for one contribution and scope.

ContextTier

Bases: StrEnum

Authority tiers, mirroring the prompt contract's vocabulary.

IngestSelection dataclass

IngestSelection(text: str = '', revision: str = '', dropped: tuple[tuple[str, str], ...] = ())

What a source emitted, plus a reason for everything it left out.

The knowledge bridge already answers this way — FragmentSelection.dropped carries a (source, reason) pair per refused fragment — and the document and attachment bridges refuse content for the same kinds of reason. Without the channel, an operator asking "why is the handbook not in the prompt?" cannot tell a store that returned None from a document over the per-document ceiling from one evicted by the aggregate cap: the read's text shows the same absence in all three cases and its revision lists only what survived.

:meth:as_read is what the :class:ContextSource protocol consumes; the selection itself is what a composition root reads when it wants to surface the omissions the way the prompting capability surfaces PromptDiagnostic.

as_read

as_read() -> ContextRead

Project the selection into the value :meth:ContextSource.read returns.

Source code in src/symfonic/capabilities/knowledge/contracts.py

def as_read(self) -> ContextRead:
    """Project the selection into the value :meth:`ContextSource.read` returns."""
    return ContextRead(text=self.text, revision=self.revision, untrusted=True)

StaticContextSource dataclass

StaticContextSource(text: str, revision: str = 'static', scope_aware: bool = False, offline_safe: bool = True)

A source whose content is fixed at declaration time. Test and seed use.

layer_index

layer_index(layer: ContextLayer) -> int

Position of layer on the ladder; lower renders earlier.

Source code in src/symfonic/capabilities/knowledge/contracts.py

def layer_index(layer: ContextLayer) -> int:
    """Position of ``layer`` on the ladder; lower renders earlier."""
    return LAYER_LADDER.index(layer)

ordering_key

ordering_key(contribution: ContextContribution) -> tuple[int, int, str]

Total order: layer, then declared order, then id.

Totality is the point — two contributions that tie on layer and order still have a defined relative position, so what the bridge hands the compiler is byte-stable across runs and across upstream dict iteration orders.

Source code in src/symfonic/capabilities/knowledge/contracts.py

def ordering_key(contribution: ContextContribution) -> tuple[int, int, str]:
    """Total order: layer, then declared order, then id.

    Totality is the point — two contributions that tie on layer and order still
    have a defined relative position, so what the bridge hands the compiler is
    byte-stable across runs and across upstream dict iteration orders.
    """
    return (
        layer_index(contribution.layer),
        contribution.order,
        contribution.contribution_id,
    )

resolve_scope

resolve_scope(scope: str | None) -> ContextScope

Turn a caller's scope string into a member, or refuse it in-hierarchy.

ContextScope('tenant-a') raises a bare :class:ValueError, which is the one refusal in this capability a caller catching :class:~.errors.BridgeContractError around declaration building would miss. The bridge factories go through here so that every way of getting a declaration wrong reports the same way.

Source code in src/symfonic/capabilities/knowledge/contracts.py

def resolve_scope(scope: str | None) -> ContextScope:
    """Turn a caller's scope string into a member, or refuse it in-hierarchy.

    ``ContextScope('tenant-a')`` raises a bare :class:`ValueError`, which is the
    one refusal in this capability a caller catching
    :class:`~.errors.BridgeContractError` around declaration building would
    miss. The bridge factories go through here so that every way of getting a
    declaration wrong reports the same way.
    """
    if scope is None:
        return ContextScope.DEPLOYMENT
    try:
        return ContextScope(scope)
    except ValueError as exc:
        raise BridgeContractError(
            f"scope {scope!r} is not a context scope; permitted values are "
            f"{[member.value for member in ContextScope]}. A scope decides how widely "
            "one contribution's content is shared, so an unrecognised one is refused "
            "rather than guessed at."
        ) from exc