symfonic.capabilities.prompting.contracts

contracts

The prompt contribution contract — what a contributor declares about content.

A contribution declares four things and nothing else: what text it supplies (a source), how stable it is (a layer), how much authority it carries (a tier), and how widely its content is shared (a scope). Everything a contributor might want to state instead — its position in the prompt, which cache region holds it, how many tokens it may spend — is a compiler decision, because those are global properties and a contributor can only see itself.

Trust is declared on both halves. The tier says what the operator believes about the content; the read says what the source knows about it. When they disagree the render gate refuses (see :mod:.gates) — the disagreement is the signal, and resolving it in favour of either side would erase it.

AsyncContributionSource

Bases: Protocol

The asynchronous half of the same port (S01, TA8.51).

read stays exactly what it was. This is a second member rather than a coroutine version of the first, and the choice is the whole design:

• every source written against :class:ContributionSource keeps working, and the synchronous :func:~.compiler.compile_prompt keeps compiling them without an event loop;
• a source whose content lives behind an await -- the shape every symfonic.core.prompt.blocks source already has, where BlockSource.load is a coroutine function PromptBlockResolver awaits -- declares aread and is awaited by :func:~.compiler.compile_prompt_async.

A synchronous bridge was the other option and it was refused on purpose: asyncio.run inside read explodes on a loop that is already running, and a thread hop per contribution per turn would trade a missing port for a latency defect on every turn that has nothing to do with blocks.

A source may declare both. Declaring both means the synchronous door can still compile it; declaring only aread means the synchronous door refuses it by name rather than dropping it silently.

ContributionScope

Bases: StrEnum

How widely one contribution's content is shared.

ContributionSource

Bases: Protocol

Reads the current content of one contribution for one scope.

scope_aware and offline_safe are declared members, so an isinstance check requires them to be present: a source that never decided whether it keys on scope does not satisfy this protocol.

PromptContribution dataclass

PromptContribution(contribution_id: str, source: ContributionSource, capability: str = 'prompting', layer: Layer = Layer.L1, tier: TrustTier = TrustTier.OPERATING, scope: ContributionScope = ContributionScope.DEPLOYMENT, order: int = 0, inherit: bool = True, pinned: bool = False, requires_hydration: bool = False, cache: CacheDirective | None = None, on_source_failure: SourceFailurePolicy | None = None, agent_permissions: frozenset[str] = frozenset(), operator_editable: bool = False, label_prefix: str = '', profile_fields: frozenset[str] = frozenset())

One capability's declaration of context it contributes to the prompt.

failure_policy property

failure_policy: SourceFailurePolicy

The declared policy, or the tier's default when none was declared.

is_learned property

is_learned: bool

True when this content is aggregated rather than authored.

validate

validate() -> None

Shape validation, performed the moment a contribution is declared.

Resolution against the whole contribution set (duplicate ids, ordering, budget) happens later in the compiler, where the set is knowable.

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

def validate(self) -> None:
    """Shape validation, performed the moment a contribution is declared.

    Resolution against the whole contribution set (duplicate ids, ordering,
    budget) happens later in the compiler, where the set is knowable.
    """
    if not self.contribution_id:
        raise ContributionContractError(
            "a prompt contribution must declare a non-empty contribution_id."
        )
    if not _ID_CHARSET.match(self.contribution_id):
        raise ContributionContractError(
            f"contribution id {self.contribution_id!r} is outside the permitted charset "
            "[A-Za-z0-9_.-]; ids appear in isolation keys and rendered delimiters, where "
            "a separator or an angle bracket forges a boundary."
        )
    if is_coroutine_read(self.source) and not is_async_source(self.source):
        raise ContributionContractError(
            f"contribution {self.contribution_id!r} declares "
            f"{type(self.source).__name__}.read as a coroutine function. The awaited "
            "member of this port is named aread(); a coroutine read() cannot be called "
            "on the synchronous door and would reach the renderer unawaited on the "
            "asynchronous one. Rename it to aread()."
        )
    if not is_sync_source(self.source) and not is_async_source(self.source):
        raise ContributionContractError(
            f"contribution {self.contribution_id!r} declares a source that cannot read: "
            f"{type(self.source).__name__} has neither a callable read() nor a callable "
            "aread(). One of the two is the port; declaring neither is a source nothing "
            "can ask."
        )
    if self.agent_permissions - {"read"} and self.tier in AUTHORED_TIERS:
        raise ContributionContractError(
            f"contribution {self.contribution_id!r} grants "
            f"{sorted(self.agent_permissions - {'read'})!r} at the authored tier "
            f"{self.tier.value!r}. Authored content is what the operator wrote; a verb "
            "beyond 'read' there would let the agent edit its own instructions. Move it "
            "to a learned tier ('profile' or 'session')."
        )
    if self.cache is not None and self.cache.cacheable and is_volatile(self.layer):
        raise ContributionContractError(
            f"contribution {self.contribution_id!r} declares itself cacheable on the "
            f"volatile layer {self.layer.value}; per-turn content in a cached region "
            "invalidates the prefix every turn."
        )
    if self.scope is not ContributionScope.DEPLOYMENT and not getattr(
        self.source, "scope_aware", False
    ):
        raise ContributionContractError(
            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."
        )

PromptContributor

Bases: Protocol

What a capability registers with the prompt compiler.

A contributor is asked once per compile and answers with declarations, not with text: the compiler decides ordering, gating, budgeting, and caching over the whole set, which it can only do if nothing has rendered yet.

SourceFailurePolicy

Bases: StrEnum

What the compiler does when a contribution's source cannot be read.

SourceRead dataclass

SourceRead(text: str, revision: str = '', untrusted: bool = False, fields: Mapping[str, str] | None = None)

What a source answered.

untrusted is the source's own declaration about its payload. A source that fetches a web page or reads a user-writable row says so here, and the render gate then refuses to place it at an authored tier.

SourceRequest dataclass

SourceRequest(contribution_id: str, scope_path: str = '', turn: int = 0, scope: ContributionScope = ContributionScope.DEPLOYMENT)

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

StaticSource dataclass

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

A source whose content is fixed at declaration time.

Deployment-global by construction: scope_aware is False because one literal string is the same for every tenant, and saying so is what lets the scope-pairing check reject a tenant-scoped block backed by it.

TrustTier

Bases: StrEnum

Authority tiers, highest first.

platform and operating are authored: a human wrote that text. profile and session are learned: their content is aggregated from what the system recorded about a user, so it is attacker-influenced input.

default_failure_policy

default_failure_policy(tier: TrustTier) -> SourceFailurePolicy

The policy a contribution of tier gets when it declares none.

fail_closed for the authored tiers: losing BOUNDARIES or RULES removes the agent's constraints. omit for the learned tiers: losing a profile costs personalisation for a turn, which does not justify failing the turn.

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

def default_failure_policy(tier: TrustTier) -> SourceFailurePolicy:
    """The policy a contribution of ``tier`` gets when it declares none.

    ``fail_closed`` for the authored tiers: losing BOUNDARIES or RULES removes
    the agent's constraints. ``omit`` for the learned tiers: losing a profile
    costs personalisation for a turn, which does not justify failing the turn.
    """
    return (
        SourceFailurePolicy.FAIL_CLOSED
        if tier in AUTHORED_TIERS
        else SourceFailurePolicy.OMIT
    )

is_async_source

is_async_source(source: object) -> bool

True when source offers the awaited member.

Source code in src/symfonic/capabilities/prompting/ports.py

def is_async_source(source: object) -> bool:
    """``True`` when ``source`` offers the awaited member."""
    return callable(getattr(source, "aread", None))

is_coroutine_read

is_coroutine_read(source: object) -> bool

True when source wrote async def read instead of aread.

The mistake is one keystroke from the intended shape and it used to be invisible: :func:is_sync_source matched on callable(read) alone, so such a source passed as synchronous, resolve_source_async called it without awaiting, and the coroutine object travelled into the renderer as if it were a :class:SourceRead. The failure surfaced as an AttributeError on .revision two modules from its cause, plus a "coroutine was never awaited" warning.

The mirrored mistake -- aread only, taken to the synchronous door -- has always been refused by name. This makes the pair symmetric.

Source code in src/symfonic/capabilities/prompting/ports.py

def is_coroutine_read(source: object) -> bool:
    """``True`` when ``source`` wrote ``async def read`` instead of ``aread``.

    The mistake is one keystroke from the intended shape and it used to be
    invisible: :func:`is_sync_source` matched on ``callable(read)`` alone, so
    such a source passed as synchronous, ``resolve_source_async`` called it
    without awaiting, and the coroutine object travelled into the renderer as
    if it were a :class:`SourceRead`. The failure surfaced as an
    ``AttributeError`` on ``.revision`` two modules from its cause, plus a
    "coroutine was never awaited" warning.

    The mirrored mistake -- ``aread`` only, taken to the synchronous door -- has
    always been refused by name. This makes the pair symmetric.
    """
    return inspect.iscoroutinefunction(getattr(source, "read", None))

is_sync_source

is_sync_source(source: object) -> bool

True when source offers the synchronous member.

A coroutine read is deliberately not one: it cannot be called on the synchronous door and calling it on the asynchronous one would produce a coroutine where the compiler expects bytes. See :func:is_coroutine_read.

Source code in src/symfonic/capabilities/prompting/ports.py

def is_sync_source(source: object) -> bool:
    """``True`` when ``source`` offers the synchronous member.

    A coroutine ``read`` is deliberately **not** one: it cannot be called on
    the synchronous door and calling it on the asynchronous one would produce a
    coroutine where the compiler expects bytes. See :func:`is_coroutine_read`.
    """
    return callable(getattr(source, "read", None)) and not is_coroutine_read(source)

ordering_key

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

Total order over contributions: 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 the compiled prompt is byte-stable across runs and across dict/set iteration orders upstream.

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

def ordering_key(contribution: PromptContribution) -> tuple[int, int, str]:
    """Total order over contributions: 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 the compiled prompt is byte-stable
    across runs and across dict/set iteration orders upstream.
    """
    return (layer_index(contribution.layer), contribution.order, contribution.contribution_id)