symfonic.capabilities.prompting.declarations

declarations

Turning what a composition root declares into what the compiler compiles.

persona_sources and guardrail_sources return sources -- the objects with a read. The compiler works on contributions, which wrap a source with where it sits and what may drop it. A composition root handed the first had to know the second existed, and the mismatch did not fail where it was made: it surfaced inside the compiler as a missing validate on a type that file never names.

One function, so both doors of :class:PromptingCapability normalise the same way and a declaration list means the same thing at declaration time and at stage time.

as_contributions

as_contributions(declared: Sequence[Any]) -> tuple[Any, ...]

Accept a declaration or a bare source; return declarations.

Identity comes from the source's own type and its position, which keeps two personas in one list distinct and keeps the declared order the compile order -- the ordering persona_sources documents (identity first) is a promise about this list, so it must survive the wrapping.

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

def as_contributions(declared: Sequence[Any]) -> tuple[Any, ...]:
    """Accept a declaration or a bare source; return declarations.

    Identity comes from the source's own type and its position, which keeps
    two personas in one list distinct and keeps the declared order the compile
    order -- the ordering ``persona_sources`` documents (identity first) is a
    promise about this list, so it must survive the wrapping.
    """
    wrapped: list[Any] = []
    for index, item in enumerate(declared):
        if getattr(item, "source", None) is not None:
            wrapped.append(item)
            continue
        wrapped.append(
            PromptContribution(
                contribution_id=f"prompting.{type(item).__name__.lower()}.{index}",
                source=item,
                order=index,
                tier=_tier_for(item),
            )
        )
    return tuple(wrapped)

split_by_door

split_by_door(sources: Sequence[Any]) -> tuple[tuple[Any, ...], tuple[Any, ...]]

Partition declared sources into ones the sync door can read, and awaited ones.

Needed because :meth:PromptingCapability.contribute is synchronous — a capability declares its stage before any loop is running — while an awaited source can only be read once the stage runs. The partition is what keeps the descriptor honest: static_prompt_digest covers what was actually compiled here, and awaited_sources counts what it could not cover.

A source offering both members is not deferred: the synchronous door can read it, so the digest covers it and the async door still awaits it at stage time.

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

def split_by_door(sources: Sequence[Any]) -> tuple[tuple[Any, ...], tuple[Any, ...]]:
    """Partition declared sources into ones the sync door can read, and awaited ones.

    Needed because :meth:`PromptingCapability.contribute` is synchronous — a
    capability declares its stage before any loop is running — while an awaited
    source can only be read once the stage runs. The partition is what keeps the
    descriptor honest: ``static_prompt_digest`` covers what was actually
    compiled here, and ``awaited_sources`` counts what it could not cover.

    A source offering **both** members is not deferred: the synchronous door can
    read it, so the digest covers it and the async door still awaits it at stage
    time.
    """
    static: list[Any] = []
    awaited: list[Any] = []
    for declared in sources:
        source = getattr(declared, "source", None)
        target = (
            awaited
            if is_async_source(source) and not is_sync_source(source)
            else static
        )
        target.append(declared)
    return tuple(static), tuple(awaited)