`symfonic.capabilities.prompting`¶

prompting ¶

The prompt and context capability (T3.2.1, REQ-S3.2).

One compiler owns everything that decides what the model reads: domain instructions, the JIT and stratified strategies, the stratigraphic layers, standing contributions and their sources, scope boundaries and inheritance, render gates, budgets, and cache annotations. Those nine concerns used to live in five packages that each knew a little about the others' ordering; here they are one pipeline with one entry point, :func:compile_prompt.

The three deliverables of this task map onto three modules:

:mod:.contracts — the prompt contribution contract: what a capability declares, and (just as load-bearing) what it may not.
:mod:.cache — the cache-region model: derived regions, the volatile-content invariant, and the non-increasing TTL ladder.
:mod:.compiler — the prompt/context compiler: accept, validate, admit, resolve, gate, order, budget, region, freeze.

Nothing here imports the kernel runtime, a provider SDK, a memory backend, or a tokenizer download. The one outward edge is :mod:.assembly, which speaks kernel contracts only.

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.

BudgetReport `dataclass` ¶

BudgetReport(limit: int | None = None, total_tokens: int = 0, admitted: tuple[str, ...] = (), dropped: tuple[str, ...] = ())

The budget arithmetic, kept alongside the prompt it explains.

CacheDirective `dataclass` ¶

CacheDirective(cacheable: bool = False, ttl: CacheTtl | None = None)

One contribution's (or region's) cache annotation.

ttl without cacheable is inert by construction rather than by convention: :meth:marker reads cacheable first, so a directive that names a tier it never earned cannot leak a marker onto the wire.

rank `property` ¶

rank: int

TTL tier as a comparable rank; None and 5m share tier 0.

marker ¶

marker() -> dict[str, str] | None

The wire-level cache_control value, or None when uncached.

The marker is reconstructed from ttl rather than passed through from anything a caller built, so the wire shape stays canonical however the directive was assembled.

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

def marker(self) -> dict[str, str] | None:
    """The wire-level ``cache_control`` value, or ``None`` when uncached.

    The marker is *reconstructed* from ``ttl`` rather than passed through
    from anything a caller built, so the wire shape stays canonical however
    the directive was assembled.
    """
    if not self.cacheable:
        return None
    if self.ttl is CacheTtl.ONE_HOUR:
        return {"type": "ephemeral", "ttl": "1h"}
    return {"type": "ephemeral"}

CacheRegion `dataclass` ¶

CacheRegion(index: int, layer: Layer, directive: CacheDirective, text: str, contribution_ids: tuple[str, ...], digest: str)

One cache-addressable span of the compiled prompt.

annotation ¶

annotation() -> dict[str, object]

This region as a provider content block, marker included when cached.

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

def annotation(self) -> dict[str, object]:
    """This region as a provider content block, marker included when cached."""
    block: dict[str, object] = {"type": "text", "text": self.text}
    marker = self.directive.marker()
    if marker is not None:
        block["cache_control"] = marker
    return block

CacheTtl ¶

Bases: StrEnum

The two TTL tiers a cache breakpoint can advertise.

CompiledPrompt `dataclass` ¶

CompiledPrompt(strategy: str, regions: tuple[CacheRegion, ...] = (), contributions: tuple[RenderedContribution, ...] = (), budget: BudgetReport = BudgetReport(), diagnostics: tuple[PromptDiagnostic, ...] = (), digest: str = '')

One compiled prompt: regions, survivors, budget, diagnostics, digest.

text `property` ¶

text: str

The whole prompt as one string, regions joined in ladder order.

cache_annotations ¶

cache_annotations() -> list[dict[str, object]]

The prompt as provider content blocks, cache markers included.

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

def cache_annotations(self) -> list[dict[str, object]]:
    """The prompt as provider content blocks, cache markers included."""
    return [region.annotation() for region in self.regions]

region_of ¶

region_of(contribution_id: str) -> CacheRegion | None

The region carrying contribution_id, or None when it was dropped.

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

def region_of(self, contribution_id: str) -> CacheRegion | None:
    """The region carrying ``contribution_id``, or ``None`` when it was dropped."""
    for region in self.regions:
        if contribution_id in region.contribution_ids:
            return region
    return None

ContextStrategy ¶

Bases: StrEnum

How much context the compiler assembles, and whether it may be cached.

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.

DomainPersona `dataclass` ¶

DomainPersona(name: str, role: str = 'assistant', tone: str = 'helpful', description: str = '', onboarding: Sequence[str] = tuple())

Who a domain's assistant is, in the four fields that reach the model.

Frozen for the reason load_plugin() disappears: a plan compiled from mutable inputs is a plan nobody can trust, and a persona that could be edited after composition would reintroduce that one field at a time.

Only name is required, because only name was. A domain that names itself and says nothing else is the ordinary case, and requiring a checklist to get an identity would make the smallest scaffold carry the largest one's shape.

GateResult `dataclass` ¶

GateResult(text: str | None, diagnostics: tuple[PromptDiagnostic, ...] = ())

What a gate decided for one contribution.

Guardrail `dataclass` ¶

Guardrail(statement: str, sensitive_tags: Sequence[str] = tuple(), priority: int = 0)

One rule a deployment wants honoured, and the tags that watch it.

Frozen, like every other composition input: a plan compiled from mutable rules is a plan whose rules nobody can quote afterwards.

HeuristicTokenEstimator `dataclass` ¶

HeuristicTokenEstimator(chars_per_token: int = 4)

Characters-per-token arithmetic — the offline default.

Deliberately an estimate with a name that says so. A provider-exact counter is a legitimate substitution (T3.2.2 owns that seam), but it must arrive as a bound port rather than as a network call this module makes, which is why the estimator is a protocol and this is only its default.

InMemoryLastKnownGood `dataclass` ¶

InMemoryLastKnownGood(_entries: dict[tuple[str, str], SourceRead] = dict())

A per-process store of the last revision each contribution served.

Keyed by :func:~.boundaries.isolation_key, so one tenant's stored profile can never be served into another tenant's prompt — the failure mode that makes a "just use the cached copy" fallback dangerous in a shared process.

Layer ¶

Bases: StrEnum

The three stratigraphic layers, most stable first.

ProceduralSkillsSource `dataclass` ¶

ProceduralSkillsSource(layer: Any, scope: Any, cue: str = '', limit: int = DEFAULT_LIMIT, include_drafts: bool = False, untrusted: bool = True, offline_safe: bool = False, scope_aware: bool = True)

The active skills for a turn's scope, rendered for the prompt.

aread `async` ¶

aread(request: SourceRequest) -> SourceRead

The skills this scope may act on, or nothing at all.

A store fault renders an empty block rather than raising. The turn has an answer to give and a missing procedure block costs the model a hint; a raised exception costs the user the turn.

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

async def aread(self, request: SourceRequest) -> SourceRead:
    """The skills this scope may act on, or nothing at all.

    A store fault renders an empty block rather than raising. The turn has
    an answer to give and a missing procedure block costs the model a hint;
    a raised exception costs the user the turn.
    """
    from symfonic.memory.layers.procedural.skill_render import (
        render_skill_for_prompt,
    )

    try:
        skills = await self.layer.query_skills(
            self.scope,
            self.cue,
            top_k=self.limit,
            include_drafts=self.include_drafts,
        )
    except Exception:  # noqa: BLE001 - a store fault is not a turn fault
        skills = []

    rendered = [render_skill_for_prompt(skill) for skill in skills or ()]
    body = "\n".join(line for line in rendered if line and line.strip())
    text = f"{HEADING}\n{body}" if body else ""
    return SourceRead(
        text=text,
        revision=_digest(text, request.contribution_id),
        untrusted=self.untrusted,
    )

PromptBudget `dataclass` ¶

PromptBudget(max_total_tokens: int | None = None)

The ceiling one compiled prompt runs under.

None means unbounded, which is the honest default: inventing a ceiling the adopter did not ask for would silently truncate prompts that fit.

narrows ¶

narrows(parent: PromptBudget) -> bool

True when this budget is the parent's or a strictly tighter one.

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

def narrows(self, parent: PromptBudget) -> bool:
    """``True`` when this budget is the parent's or a strictly tighter one."""
    if parent.max_total_tokens is None:
        return True
    if self.max_total_tokens is None:
        return False
    return self.max_total_tokens <= parent.max_total_tokens

PromptCompileRequest `dataclass` ¶

PromptCompileRequest(instructions: str | None = None, contributions: tuple[PromptContribution, ...] = (), strategy: ContextStrategy = ContextStrategy.STRATIFIED, budget: PromptBudget = PromptBudget(), policy: RenderPolicy = DEFAULT_POLICY, scope_path: str = '', turn: int = 0, last_known_good: InMemoryLastKnownGood | None = None, estimator: object | None = None, principal_grants: frozenset[str] = frozenset(), operator_overrides: Mapping[str, str] | None = None)

Everything one prompt compile is allowed to depend on.

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.

PromptDiagnostic `dataclass` ¶

PromptDiagnostic(kind: str, subject: str, detail: str)

One recorded compile decision.

kind is the stage that made it (input, strategy, source, gate, budget, cache), subject the contribution it concerns. Diagnostics are data rather than log lines because the caller that needs them most — a test asserting a block was dropped for the right reason — has no access to a log handler.

PromptingCapability `dataclass` ¶

PromptingCapability(sources: Sequence[Any] = (), priority: int = -900, options: dict[str, Any] = dict())

Compile the layered prompt and contribute it to the turn's assembly.

Usage is the whole point of the phase::

Agent(provider, capabilities=[PromptingCapability(sources=[...])])

sources are the prompt contributions this capability compiles — the same values the capability's own compiler has always taken. They are held on the config rather than read from a global, so two agents in one process do not share a prompt.

contribute ¶

contribute(request: CapabilityRequest) -> CapabilityContribution

Declare the stage and supply the handler that answers it.

Both in one value, which is what lets fold_contributions refuse a declaration nothing runs. The compile happens here, at contribution time, so the stage descriptor can carry the compiled digest — that is what makes "this plan compiled this prompt" checkable from the plan alone, without the plan carrying the prompt text.

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

def contribute(self, request: CapabilityRequest) -> CapabilityContribution:
    """Declare the stage and supply the handler that answers it.

    Both in one value, which is what lets ``fold_contributions`` refuse a
    declaration nothing runs. The compile happens *here*, at contribution
    time, so the stage descriptor can carry the compiled digest — that is
    what makes "this plan compiled this prompt" checkable from the plan
    alone, without the plan carrying the prompt text.
    """
    static, awaited = split_by_door(as_contributions(self.sources))
    digest_options, deferred_gate = _digest_options(self.options)
    compiled = compile_prompt(
        PromptCompileRequest(contributions=static, **digest_options)
    )
    descriptor = prompt_stage_descriptor(
        compiled,
        priority=self.priority,
        awaited_sources=len(awaited),
        deferred_render_gate=deferred_gate,
    )

    async def handle(context: Any) -> StageResult[Any]:
        """Project the compiled prompt onto the turn being assembled.

        Reports ``NO_CHANGE`` with a reason when the compile produced
        nothing, rather than contributing an empty assembly. An empty
        contribution and a capability that decided not to act are different
        facts, and only one of them is a reason to look at the sources.
        """
        # Resolution ran first (STG-7), so the snapshot is complete and
        # frozen. Recompiling with it is what makes this capability compile
        # the *turn's* prompt rather than only its static configuration.
        resolved = _contributions_from(getattr(context, "resolved", None))
        # The turn's scope reaches every source through the compile request:
        # ``resolve_source`` passes ``scope_path`` down, which is how a
        # source bound at construction still renders for the tenant being
        # served. Without it the system prompt named whichever scope the
        # bundle was folded with, on every turn.
        scope_path = _scope_path_of(getattr(context, "request", None))
        options = dict(self.options)
        if scope_path:
            options.setdefault("scope_path", scope_path)
        # The asynchronous door (S01). The stage handler is already a
        # coroutine the kernel awaits, so awaiting the compile here costs no
        # new concurrency surface -- and it is what lets a contribution whose
        # content lives behind an ``await`` reach the prompt at all. A
        # source that only offers the synchronous ``read`` is still called,
        # not wrapped: ``resolve_source_async`` calls it directly.
        turn_compiled = (
            await compile_prompt_async(
                PromptCompileRequest(
                    contributions=as_contributions(self.sources) + resolved,
                    **options)
            )
            if resolved or self.sources
            else compiled
        )
        if not turn_compiled.text:
            return no_change(
                "the prompt compiled to nothing: no contribution survived "
                "its gates, or none was supplied"
            )
        turn_request = getattr(context, "request", None)
        if turn_request is None:  # pragma: no cover - defensive
            return no_change("no turn request on the stage context")
        assembly = to_prompt_assembly(
            turn_compiled,
            prompt=turn_request.prompt,
            attachments=tuple(turn_request.attachments),
            history=tuple(turn_request.history),
        )
        return applied(_compose(context, assembly))

    return CapabilityContribution(
        capability="prompting",
        stages=(descriptor,),
        handlers={PROMPTING_STAGE: handle},
        # No effects: prompt assembly is a pure function of the plan and the
        # request (STG-7). Claiming one would widen this capability's own
        # admission for nothing.
        effect_grants=frozenset(),
    )

RegionRow ¶

Bases: NamedTuple

One rendered contribution, as the region planner sees it.

A narrow tuple rather than the full rendered value: the planner must not be able to read a tier, a trust flag, or a source, because any of those would become a second place cache decisions get made.

RenderPolicy `dataclass` ¶

RenderPolicy(max_learned_chars: int = 500, render_when: Callable[[PromptContribution], Any] | None = None)

The limits applied to learned content, and the set-wide render gate.

Authored content is deliberately uncapped: it is operator configuration, already reviewed by whoever deployed it, and a cap there would silently delete instructions the operator can see in their own config file.

render_when is S01's counterpart to the legacy per-block predicate, and it lives here rather than on the contribution for the reason PROMPT_BLOCK_CONTRACT published ahead of the port: the kernel gate is a policy over the whole set, so a host predicate migrates as a policy rule. It is consulted before any source is read, so a gated-off contribution costs no I/O -- the same ordering PromptBlockResolver.resolve_block documents. It may return an awaitable, in which case only the asynchronous door can honour it; the synchronous one refuses by name.

RenderedContribution `dataclass` ¶

RenderedContribution(contribution_id: str, layer: Layer, tier: TrustTier, text: str, revision: str, tokens: int, directive: CacheDirective, pinned: bool, degraded: bool = False)

One contribution after gating, with its cost and cache annotation fixed.

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.

SourceResolution `dataclass` ¶

SourceResolution(read: SourceRead | None, diagnostics: tuple[PromptDiagnostic, ...] = (), degraded: bool = False)

The outcome of reading one contribution's source.

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.

TokenEstimator ¶

Bases: Protocol

Estimates the token cost of a rendered string.

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.

admit_within_budget ¶

admit_within_budget(rows: Sequence[BudgetRow], budget: PromptBudget) -> tuple[tuple[BudgetRow, ...], BudgetReport]

Admit as much as the budget allows, dropping in a fixed, total order.

The returned rows keep their original (compiled) order; only membership is decided here. Reordering the prompt to fit would change what the model reads for reasons the adopter never declared.

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

def admit_within_budget(
    rows: Sequence[BudgetRow], budget: PromptBudget
) -> tuple[tuple[BudgetRow, ...], BudgetReport]:
    """Admit as much as the budget allows, dropping in a fixed, total order.

    The returned rows keep their original (compiled) order; only membership is
    decided here. Reordering the prompt to fit would change what the model
    reads for reasons the adopter never declared.
    """
    total = sum(row.tokens for row in rows)
    limit = budget.max_total_tokens
    if limit is None or total <= limit:
        return tuple(rows), BudgetReport(
            limit=limit,
            total_tokens=total,
            admitted=tuple(row.contribution_id for row in rows),
        )

    pinned_total = sum(row.tokens for row in rows if row.pinned)
    if pinned_total > limit:
        pinned = ", ".join(row.contribution_id for row in rows if row.pinned)
        raise BudgetExceededError(
            f"pinned contributions ({pinned}) need {pinned_total} tokens but the budget is "
            f"{limit}. Pinned content is never dropped and never truncated: raise the budget "
            "or unpin something, but the compiler will not ship a prompt missing it."
        )

    dropped: set[str] = set()
    running = total
    for row in _most_droppable_first(rows):
        if running <= limit:
            break
        if row.pinned:
            continue
        dropped.add(row.contribution_id)
        running -= row.tokens

    admitted = tuple(row for row in rows if row.contribution_id not in dropped)
    return admitted, BudgetReport(
        limit=limit,
        total_tokens=sum(row.tokens for row in admitted),
        admitted=tuple(row.contribution_id for row in admitted),
        dropped=tuple(row.contribution_id for row in rows if row.contribution_id in dropped),
    )

cache_directive_for ¶

cache_directive_for(strategy: ContextStrategy, contribution: PromptContribution) -> CacheDirective

The cache annotation this contribution carries under strategy.

JIT caches nothing — including a contribution that declared a directive, since a prefix that changes shape per turn cannot be a cache hit and the breakpoint would be pure cost. Under stratified a declared directive wins, and the default is derived from volatility: stable layers cache at the provider's default tier, the volatile layer never does.

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

def cache_directive_for(
    strategy: ContextStrategy, contribution: PromptContribution
) -> CacheDirective:
    """The cache annotation this contribution carries under ``strategy``.

    JIT caches nothing — including a contribution that declared a directive,
    since a prefix that changes shape per turn cannot be a cache hit and the
    breakpoint would be pure cost. Under ``stratified`` a declared directive
    wins, and the default is derived from volatility: stable layers cache at
    the provider's default tier, the volatile layer never does.
    """
    if strategy is ContextStrategy.JIT:
        return UNCACHED
    if contribution.cache is not None:
        return contribution.cache
    if is_volatile(contribution.layer):
        return UNCACHED
    return CacheDirective(cacheable=True)

compile_prompt ¶

compile_prompt(request: PromptCompileRequest) -> CompiledPrompt

Compile one prompt from one request. Reads sources; performs nothing else.

The synchronous door onto the one sequence. A contribution whose source only offers aread is refused here by name (see :func:~.sources.resolve_source) rather than dropped.

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

def compile_prompt(request: PromptCompileRequest) -> CompiledPrompt:
    """Compile one prompt from one request. Reads sources; performs nothing else.

    The synchronous door onto the one sequence. A contribution whose source
    only offers ``aread`` is refused here by name (see
    :func:`~.sources.resolve_source`) rather than dropped.
    """
    diagnostics: list[PromptDiagnostic] = []
    estimator, admitted = _prepare(request, diagnostics)
    admitted = gated(request, admitted, diagnostics)
    rendered: list[RenderedContribution] = []
    for contribution in admitted:
        resolution = overridden(request, contribution, diagnostics)
        if resolution is None:
            resolution = resolve_source(
                contribution,
                scope_path=request.scope_path,
                turn=request.turn,
                store=request.last_known_good,
            )
        row = _render_row(request, contribution, resolution, estimator, diagnostics)
        if row is not None:
            rendered.append(row)
    return _finish(request, tuple(rendered), diagnostics)

compile_prompt_async `async` ¶

compile_prompt_async(request: PromptCompileRequest) -> CompiledPrompt

The same sequence, with step 4 awaited (S01, TA8.51).

This is not the forbidden second pipeline, and the difference is worth stating because the module docstring above forbids one by name. A second pipeline is a second place ordering, gating, budgeting and region planning are decided; those steps are :func:_prepare and :func:_finish here, and both doors call the same two functions with the same arguments. What differs is one line: whether the source read is called or awaited. A contribution's position in the prompt cannot depend on which door compiled it, because neither door decides positions.

The alternative -- asyncio.run or a thread hop inside a synchronous read -- was refused for the reason S01 states: it would block the loop every turn, trading a missing port for a latency defect.

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

async def compile_prompt_async(request: PromptCompileRequest) -> CompiledPrompt:
    """The same sequence, with step 4 awaited (S01, TA8.51).

    **This is not the forbidden second pipeline**, and the difference is worth
    stating because the module docstring above forbids one by name. A second
    pipeline is a second place ordering, gating, budgeting and region planning
    are decided; those steps are :func:`_prepare` and :func:`_finish` here, and
    both doors call the same two functions with the same arguments. What
    differs is one line: whether the source read is called or awaited. A
    contribution's position in the prompt cannot depend on which door compiled
    it, because neither door decides positions.

    The alternative -- ``asyncio.run`` or a thread hop inside a synchronous
    ``read`` -- was refused for the reason S01 states: it would block the loop
    every turn, trading a missing port for a latency defect.
    """
    diagnostics: list[PromptDiagnostic] = []
    estimator, admitted = _prepare(request, diagnostics)
    admitted = await gated_async(request, admitted, diagnostics)
    rendered: list[RenderedContribution] = []
    for contribution in admitted:
        resolution = overridden(request, contribution, diagnostics)
        if resolution is None:
            resolution = await resolve_source_async(
                contribution,
                scope_path=request.scope_path,
                turn=request.turn,
                store=request.last_known_good,
            )
        row = _render_row(request, contribution, resolution, estimator, diagnostics)
        if row is not None:
            rendered.append(row)
    return _finish(request, tuple(rendered), diagnostics)

compiled_instructions ¶

compiled_instructions(compiled: CompiledPrompt) -> str | None

The compiled prompt as kernel instructions, or None when it is empty.

None rather than "" because the kernel's instructions field is optional, and an empty string is a system prompt that says nothing — indistinguishable downstream from a deployment that meant to send one and lost it.

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

def compiled_instructions(compiled: CompiledPrompt) -> str | None:
    """The compiled prompt as kernel instructions, or ``None`` when it is empty.

    ``None`` rather than ``""`` because the kernel's ``instructions`` field is
    optional, and an empty string is a system prompt that says nothing —
    indistinguishable downstream from a deployment that meant to send one and
    lost it.
    """
    text = compiled.text
    return text or None

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
    )

derive_child_request ¶

derive_child_request(parent: PromptCompileRequest, *, contributions: Sequence[PromptContribution] | None = None, budget: PromptBudget | None = None, scope_path: str | None = None, instructions: str | None = None, principal_grants: frozenset[str] | None = None, operator_overrides: Mapping[str, str] | None = None) -> PromptCompileRequest

Derive a delegated child's request by narrowing parent.

Omitting contributions inherits the parent's inheritable set — the safe default, since a child silently losing its boundaries is the failure that matters. Naming them narrows: every named id must already be inheritable on the parent, so a child cannot promote a contribution its parent marked non-inheritable by re-declaring it under the same name.

The two S01 access axes inherit, and the rule is stated rather than implied. Delegation here is the same principal on the same turn, so a child that dropped principal_grants would silently withhold from the subagent a contribution the parent was entitled to, and one that dropped operator_overrides would serve the source's own text where the operator had replaced it — a divergence between what the parent and the child read that no diagnostic would explain. Inheriting is therefore the default, and both axes narrow the same way everything else here does:

principal_grants must be a subset of the parent's. A child cannot hand itself a verb the turn's principal was not granted.
operator_overrides must be a sub-mapping of the parent's — the same text under a subset of the same ids. A child may drop an override; it may not introduce one, nor change the text of one, because either would make the operator's decision for the parent turn something the child rewrites.

Delegation is a block-visibility boundary elsewhere in this repo (tests/agent/subagents/test_child_block_lockdown.py), and that boundary is drawn by inherit on the contribution, which is enforced above. These two axes are about the principal and the operator, not about which blocks a child may see.

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

def derive_child_request(
    parent: PromptCompileRequest,
    *,
    contributions: Sequence[PromptContribution] | None = None,
    budget: PromptBudget | None = None,
    scope_path: str | None = None,
    instructions: str | None = None,
    principal_grants: frozenset[str] | None = None,
    operator_overrides: Mapping[str, str] | None = None,
) -> PromptCompileRequest:
    """Derive a delegated child's request by narrowing ``parent``.

    Omitting ``contributions`` inherits the parent's inheritable set — the safe
    default, since a child silently losing its boundaries is the failure that
    matters. Naming them narrows: every named id must already be inheritable on
    the parent, so a child cannot promote a contribution its parent marked
    non-inheritable by re-declaring it under the same name.

    **The two S01 access axes inherit, and the rule is stated rather than
    implied.** Delegation here is the same principal on the same turn, so a
    child that dropped ``principal_grants`` would silently withhold from the
    subagent a contribution the parent was entitled to, and one that dropped
    ``operator_overrides`` would serve the source's own text where the operator
    had replaced it — a divergence between what the parent and the child read
    that no diagnostic would explain. Inheriting is therefore the default, and
    both axes narrow the same way everything else here does:

    * ``principal_grants`` must be a subset of the parent's. A child cannot
      hand itself a verb the turn's principal was not granted.
    * ``operator_overrides`` must be a sub-mapping of the parent's — the same
      text under a subset of the same ids. A child may drop an override; it may
      not introduce one, nor change the text of one, because either would make
      the operator's decision for the parent turn something the child rewrites.

    Delegation is a block-*visibility* boundary elsewhere in this repo
    (``tests/agent/subagents/test_child_block_lockdown.py``), and that boundary
    is drawn by ``inherit`` on the contribution, which is enforced above. These
    two axes are about the principal and the operator, not about which blocks a
    child may see.
    """
    available = {c.contribution_id: c for c in inheritable(parent.contributions)}
    if contributions is None:
        selected: tuple[PromptContribution, ...] = tuple(available.values())
    else:
        unknown = [c.contribution_id for c in contributions if c.contribution_id not in available]
        if unknown:
            raise InheritanceError(
                f"child request declares {unknown!r}, which the parent did not contribute "
                "(or marked non-inheritable). A child narrows its parent's standing context; "
                "it never widens it."
            )
        selected = tuple(contributions)

    child_budget = parent.budget if budget is None else budget
    if not child_budget.narrows(parent.budget):
        raise InheritanceError(
            f"child budget {child_budget.max_total_tokens} does not narrow the parent's "
            f"{parent.budget.max_total_tokens}; a delegated run may not spend more than the "
            "run that delegated to it."
        )

    return replace(
        parent,
        contributions=selected,
        budget=child_budget,
        scope_path=parent.scope_path if scope_path is None else scope_path,
        instructions=parent.instructions if instructions is None else instructions,
        principal_grants=_child_grants(parent, principal_grants),
        operator_overrides=_child_overrides(parent, operator_overrides),
    )

guardrail_sources ¶

guardrail_sources(guardrails: Iterable[Guardrail]) -> tuple[Any, ...]

The prompt contributions guardrails render through.

Ordered by descending priority, because a budget drops from the end and priority already existed on this value while deciding nothing. If two rules cannot both fit, the deployment has said which one matters.

No guardrails contribute no source -- not an empty section. A heading with nothing under it reads as "this deployment has no rules", which is a claim rather than an absence.

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

def guardrail_sources(guardrails: Iterable[Guardrail]) -> tuple[Any, ...]:
    """The prompt contributions ``guardrails`` render through.

    Ordered by descending priority, because a budget drops from the end and
    priority already existed on this value while deciding nothing. If two rules
    cannot both fit, the deployment has said which one matters.

    No guardrails contribute no source -- not an empty section. A heading with
    nothing under it reads as "this deployment has no rules", which is a claim
    rather than an absence.
    """
    ordered = sorted(guardrails, key=lambda rule: -rule.priority)
    return tuple(
        StaticSource(
            text=rule.statement.strip(),
            untrusted=False,
            offline_safe=True,
            scope_aware=False,
        )
        for rule in ordered
    )

inheritable ¶

inheritable(contributions: Iterable[PromptContribution]) -> tuple[PromptContribution, ...]

The subset of contributions a delegated child may see.

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

def inheritable(
    contributions: Iterable[PromptContribution],
) -> tuple[PromptContribution, ...]:
    """The subset of ``contributions`` a delegated child may see."""
    return tuple(c for c in contributions if c.inherit)

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

is_volatile ¶

is_volatile(layer: Layer) -> bool

True when content in layer may change between turns of one session.

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

def is_volatile(layer: Layer) -> bool:
    """``True`` when content in ``layer`` may change between turns of one session."""
    return layer not in STABLE_LAYERS

isolation_key ¶

isolation_key(scope_path: str, contribution_id: str) -> tuple[str, str]

The canonical storage key for one contribution in one scope.

The scope half is the full root-first path verbatim, not a tenant id: keying on the tenant alone would collapse every brand and conversation under an organisation into one bucket, so two sibling scopes would share — and overwrite — one another's stored revisions.

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

def isolation_key(scope_path: str, contribution_id: str) -> tuple[str, str]:
    """The canonical storage key for one contribution in one scope.

    The scope half is the full root-first path verbatim, not a tenant id:
    keying on the tenant alone would collapse every brand and conversation
    under an organisation into one bucket, so two sibling scopes would share —
    and overwrite — one another's stored revisions.
    """
    return (scope_path, contribution_id)

layer_index ¶

layer_index(layer: Layer) -> int

Position of layer on the ladder; lower renders earlier.

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

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

neutralise_delimiters ¶

neutralise_delimiters(value: str) -> str

Replace every delimiter-shaped sequence in value.

Two rules, because "looks like the closing delimiter" is a question with a lexical answer and a visual one, and content that defeats either has escaped the wrapper.

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

def neutralise_delimiters(value: str) -> str:
    """Replace every delimiter-shaped sequence in ``value``.

    Two rules, because "looks like the closing delimiter" is a question with a
    lexical answer and a visual one, and content that defeats either has
    escaped the wrapper.
    """
    neutralised = _DELIMITER_LOOKALIKE.sub(NEUTRALISED, value)
    return _TAGGISH_SPAN.sub(
        lambda match: match.group(0) if match.group(0).isascii() else NEUTRALISED,
        neutralised,
    )

normalise_ttl_ladder ¶

normalise_ttl_ladder(regions: Sequence[CacheRegion]) -> tuple[CacheRegion, ...]

Promote earlier cached regions to the highest tier appearing to their right.

Right-to-left scan tracking the high-water tier. Uncached regions are skipped — they carry no marker and so do not participate in the provider's ladder — which is why a volatile region between two cached ones does not reset the rule.

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

def normalise_ttl_ladder(regions: Sequence[CacheRegion]) -> tuple[CacheRegion, ...]:
    """Promote earlier cached regions to the highest tier appearing to their right.

    Right-to-left scan tracking the high-water tier. Uncached regions are
    skipped — they carry no marker and so do not participate in the provider's
    ladder — which is why a volatile region between two cached ones does not
    reset the rule.
    """
    result = list(regions)
    highest = -1
    for position in range(len(result) - 1, -1, -1):
        region = result[position]
        if not region.directive.cacheable:
            continue
        rank = region.directive.rank
        if rank < highest:
            promoted = replace(region.directive, ttl=_TTL_AT[highest])
            result[position] = replace(region, directive=promoted)
        else:
            highest = rank
    return tuple(result)

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)

persona_sources ¶

persona_sources(persona: DomainPersona) -> tuple[Any, ...]

The contribution sources persona renders through.

Ordered identity first: it is the shortest, the most stable, and the one whose absence changes how every other contribution reads. A budget that has to drop something should drop the description before it drops who the assistant is.

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

def persona_sources(persona: DomainPersona) -> tuple[Any, ...]:
    """The contribution sources ``persona`` renders through.

    Ordered identity first: it is the shortest, the most stable, and the one
    whose absence changes how every other contribution reads. A budget that
    has to drop something should drop the description before it drops who the
    assistant is.
    """
    sources: list[Any] = [
        AgentIdentitySource(
            identity=AgentIdentity.from_domain_name(
                persona.name, role=persona.role, tone=persona.tone
            )
        )
    ]

    directive = OnboardingDirective.from_checklist(persona.onboarding)
    if directive is not None:
        sources.append(OnboardingSource(directive=directive))

    if persona.description.strip():
        # Authored and offline, like the two above: it is text the deployment
        # wrote, not something fetched or supplied by a user, so the compiler
        # may place it at an authored tier rather than delimiting it as
        # untrusted.
        sources.append(
            StaticSource(
                text=persona.description.strip(),
                untrusted=False,
                offline_safe=True,
                scope_aware=False,
            )
        )

    return tuple(sources)

plan_regions ¶

plan_regions(rows: Sequence[RegionRow]) -> tuple[CacheRegion, ...]

Group rows into regions, breaking wherever the annotation changes.

Rows arrive already ordered by the compiler. Empty rows are skipped rather than emitted: a region whose only content is an empty string still costs a breakpoint, and breakpoints are the scarce resource here.

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

def plan_regions(rows: Sequence[RegionRow]) -> tuple[CacheRegion, ...]:
    """Group ``rows`` into regions, breaking wherever the annotation changes.

    Rows arrive already ordered by the compiler. Empty rows are skipped rather
    than emitted: a region whose only content is an empty string still costs a
    breakpoint, and breakpoints are the scarce resource here.
    """
    regions: list[CacheRegion] = []
    bucket: list[RegionRow] = []

    def flush() -> None:
        if not bucket:
            return
        text = REGION_SEPARATOR.join(row.text for row in bucket)
        regions.append(
            CacheRegion(
                index=len(regions),
                layer=bucket[0].layer,
                directive=bucket[0].directive,
                text=text,
                contribution_ids=tuple(row.contribution_id for row in bucket),
                digest=_digest(text),
            )
        )
        bucket.clear()

    for row in rows:
        if not row.text:
            continue
        if row.directive.cacheable and is_volatile(row.layer):
            raise CacheRegionError(
                f"contribution {row.contribution_id!r} is volatile ({row.layer.value}) and "
                "cacheable; per-turn content inside a cached region invalidates the whole "
                "prefix on every turn. Move it to a stable layer or leave it uncached."
            )
        if bucket and (bucket[0].layer is not row.layer or bucket[0].directive != row.directive):
            flush()
        bucket.append(row)
    flush()

    breakpoints = sum(1 for region in regions if region.directive.cacheable)
    if breakpoints > MAX_CACHE_BREAKPOINTS:
        raise CacheRegionError(
            f"the compiled prompt declares {breakpoints} cache breakpoints; providers accept "
            f"at most {MAX_CACHE_BREAKPOINTS}. Coalesce contributions onto shared TTL tiers."
        )
    return tuple(regions)

prompt_stage_descriptor ¶

prompt_stage_descriptor(compiled: CompiledPrompt, *, priority: int = -900, awaited_sources: int = 0, deferred_render_gate: bool = False) -> StageDescriptor

The stage this capability contributes, carrying the compile's identity.

No effects and no emitted events are declared: this is a compilation stage (STG-7), a pure function of plan, request and the resolved-input snapshot. A stage that claimed an effect it does not perform would widen its own admission for nothing, and under the reformulated STG-7 it would also be refused outright.

static_prompt_digest identifies this compile, not the final prompt. The name is deliberate and the distinction is not pedantic. The digest covers sources — what the capability was configured with, compiled here at contribution time. It does not cover:

Agent(instructions=...), which the kernel's own stage puts on the assembly and this capability composes onto afterwards;
anything a resolution stage put in the turn's snapshot, which by definition does not exist yet when this descriptor is built;
any contribution whose source must be awaited (S01). contribute() is synchronous, so an aread-only source cannot be read here at all. awaited_sources counts them in the same config, because "the digest does not cover N of my sources" is a fact a plan reader needs and a silently narrower digest is exactly the false "checkable from the plan alone" claim the paragraph below retires.
the request's render_when policy, when one is set (S01). deferred_render_gate records that the digest compile ran without it. The gate is a turn-time decision over the whole set and may be awaited, so the synchronous declaration-time compile cannot honour it; running it here would refuse the plan outright for a policy the turn supports.

The field was called prompt_digest and read as "this plan compiled this prompt, checkable from the plan alone". That claim was already false before memory existed — the adopter's instructions were outside it — and the reformulated STG-7 makes the gap structural rather than accidental. Naming it for what it actually covers is the honest half; recording the final digest is a runtime artifact and needs a carrier the kernel does not have yet (see the debt table in the phase-4 sizing note).

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

def prompt_stage_descriptor(
    compiled: CompiledPrompt,
    *,
    priority: int = -900,
    awaited_sources: int = 0,
    deferred_render_gate: bool = False,
) -> StageDescriptor:
    """The stage this capability contributes, carrying the compile's identity.

    No effects and no emitted events are declared: this is a *compilation*
    stage (STG-7), a pure function of plan, request and the resolved-input
    snapshot. A stage that claimed an effect it does not perform would widen
    its own admission for nothing, and under the reformulated STG-7 it would
    also be refused outright.

    **``static_prompt_digest`` identifies this compile, not the final prompt.**
    The name is deliberate and the distinction is not pedantic. The digest
    covers ``sources`` — what the capability was configured with, compiled here
    at contribution time. It does *not* cover:

    * ``Agent(instructions=...)``, which the kernel's own stage puts on the
      assembly and this capability composes onto afterwards;
    * anything a resolution stage put in the turn's snapshot, which by
      definition does not exist yet when this descriptor is built;
    * any contribution whose source must be **awaited** (S01). ``contribute()``
      is synchronous, so an ``aread``-only source cannot be read here at all.
      ``awaited_sources`` counts them in the same config, because "the digest
      does not cover N of my sources" is a fact a plan reader needs and a
      silently narrower digest is exactly the false "checkable from the plan
      alone" claim the paragraph below retires.
    * the request's ``render_when`` policy, when one is set (S01).
      ``deferred_render_gate`` records that the digest compile ran without it.
      The gate is a turn-time decision over the whole set and may be awaited,
      so the synchronous declaration-time compile cannot honour it; running it
      here would refuse the plan outright for a policy the turn supports.

    The field was called ``prompt_digest`` and read as "this plan compiled this
    prompt, checkable from the plan alone". That claim was already false before
    memory existed — the adopter's instructions were outside it — and the
    reformulated STG-7 makes the gap structural rather than accidental. Naming
    it for what it actually covers is the honest half; recording the *final*
    digest is a runtime artifact and needs a carrier the kernel does not have
    yet (see the debt table in the phase-4 sizing note).
    """
    return StageDescriptor(
        stage_id=PROMPTING_STAGE,
        phase=Phase.PROMPT_ASSEMBLY,
        capability="prompting",
        priority=priority,
        config={
            "static_prompt_digest": compiled.digest,
            "awaited_sources": awaited_sources,
            "deferred_render_gate": deferred_render_gate,
            "strategy": compiled.strategy,
            "regions": len(compiled.regions),
            "cached_regions": sum(1 for r in compiled.regions if r.directive.cacheable),
        },
    )

render_contribution ¶

render_contribution(contribution: PromptContribution, read: SourceRead, policy: RenderPolicy = DEFAULT_POLICY) -> GateResult

Gate one read and return the text that may render, or None.

Raises only on a trust/tier mismatch: that is the one failure where continuing means putting attacker-influenced text where the model reads operator instruction. Every other refusal drops with a diagnostic.

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

def render_contribution(
    contribution: PromptContribution,
    read: SourceRead,
    policy: RenderPolicy = DEFAULT_POLICY,
) -> GateResult:
    """Gate one read and return the text that may render, or ``None``.

    Raises only on a trust/tier mismatch: that is the one failure where
    continuing means putting attacker-influenced text where the model reads
    operator instruction. Every other refusal drops with a diagnostic.
    """
    name = contribution.contribution_id
    if read.untrusted and contribution.tier in {TrustTier.PLATFORM, TrustTier.OPERATING}:
        raise RenderGateError(
            f"contribution {name!r} renders at the authored tier "
            f"{contribution.tier.value!r}, but its source declared the payload untrusted. "
            "Authored tiers render verbatim; move this contribution to a learned tier "
            "('profile' or 'session') so it renders inside the untrusted-data wrapper."
        )

    selected = select_body(contribution, read)
    if selected is None:
        return GateResult(
            None,
            (
                PromptDiagnostic(
                    "gate",
                    name,
                    f"dropped: profile_fields {sorted(contribution.profile_fields)!r} named "
                    "no value this source returned, so the contribution would render its "
                    "prose instead of the fields it declared",
                ),
            ),
        )

    if not contribution.is_learned:
        if not selected.strip():
            return GateResult(None, (PromptDiagnostic("gate", name, "empty; nothing to render"),))
        return GateResult(_prefixed(contribution, selected))

    body = _prefixed(contribution, normalise_learned(selected))
    if not body.strip():
        return GateResult(None, (PromptDiagnostic("gate", name, "empty; nothing to render"),))
    if len(body) > policy.max_learned_chars:
        return GateResult(
            None,
            (
                PromptDiagnostic(
                    "gate",
                    name,
                    f"dropped: {len(body)} chars exceeds the learned-content cap of "
                    f"{policy.max_learned_chars}; learned content is dropped, never truncated",
                ),
            ),
        )
    return GateResult(f"{untrusted_open_tag(name)}\n{body}\n{UNTRUSTED_CLOSE}")

resolve_source ¶

resolve_source(contribution: PromptContribution, *, scope_path: str = '', turn: int = 0, store: InMemoryLastKnownGood | None = None) -> SourceResolution

Read one contribution's source, applying its declared failure policy.

The synchronous door. A source that offers only aread is refused by name here rather than dropped: a silent drop is the defect this whole seam exists to avoid, and a compile that accepted the declaration and rendered none of it would be exactly the "accepted, validated, and never resolved" failure TA8.38 refused to ship.

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

def resolve_source(
    contribution: PromptContribution,
    *,
    scope_path: str = "",
    turn: int = 0,
    store: InMemoryLastKnownGood | None = None,
) -> SourceResolution:
    """Read one contribution's source, applying its declared failure policy.

    The **synchronous** door. A source that offers only ``aread`` is refused by
    name here rather than dropped: a silent drop is the defect this whole seam
    exists to avoid, and a compile that accepted the declaration and rendered
    none of it would be exactly the "accepted, validated, and never resolved"
    failure TA8.38 refused to ship.
    """
    request = source_request(contribution, scope_path=scope_path, turn=turn)
    key = isolation_key(scope_path, contribution.contribution_id)
    if not is_sync_source(contribution.source):
        raise ContributionContractError(
            f"contribution {contribution.contribution_id!r} declares an asynchronous source "
            f"({type(contribution.source).__name__} offers aread() and no read()), and this "
            "compile went through the synchronous door. Await compile_prompt_async(request) "
            "instead; the synchronous compiler cannot read it and will not pretend it did."
        )
    try:
        read = contribution.source.read(request)
    except Exception as exc:  # noqa: BLE001 - every backend failure is one policy decision
        return _apply_policy(contribution, key, exc, store)
    if store is not None and contribution.failure_policy is SourceFailurePolicy.LAST_KNOWN_GOOD:
        store.put(key, read)
    return SourceResolution(read=read)

resolve_source_async `async` ¶

resolve_source_async(contribution: PromptContribution, *, scope_path: str = '', turn: int = 0, store: InMemoryLastKnownGood | None = None) -> SourceResolution

Read one contribution's source, awaiting it when it asks to be awaited.

The asynchronous door, and the only new failure mode it introduces is cancellation, which is named ahead of the policy arm on purpose. CancelledError derives from BaseException rather than Exception, so the arm below would not have caught it anyway -- it is written out because a later edit widening that arm to BaseException would otherwise absorb a cancelled turn into on_source_failure and leave the turn running. PromptBlockResolver.resolve_block names the same hazard in the same words, and this seam is the one that brings that hazard onto the kernel line.

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

async def resolve_source_async(
    contribution: PromptContribution,
    *,
    scope_path: str = "",
    turn: int = 0,
    store: InMemoryLastKnownGood | None = None,
) -> SourceResolution:
    """Read one contribution's source, awaiting it when it asks to be awaited.

    The **asynchronous** door, and the only new failure mode it introduces is
    cancellation, which is named ahead of the policy arm on purpose.
    ``CancelledError`` derives from ``BaseException`` rather than ``Exception``,
    so the arm below would not have caught it anyway -- it is written out
    because a later edit widening that arm to ``BaseException`` would otherwise
    absorb a cancelled turn into ``on_source_failure`` and leave the turn
    running. ``PromptBlockResolver.resolve_block`` names the same hazard in the
    same words, and this seam is the one that brings that hazard onto the kernel
    line.
    """
    request = source_request(contribution, scope_path=scope_path, turn=turn)
    key = isolation_key(scope_path, contribution.contribution_id)
    try:
        if is_async_source(contribution.source):
            read = await contribution.source.aread(request)
        else:
            read = contribution.source.read(request)
    except asyncio.CancelledError:
        # Cancellation is not a source failure. Swallowing it under ``omit``
        # would report a compiled prompt for a turn that is being torn down.
        raise
    except Exception as exc:  # noqa: BLE001 - every backend failure is one policy decision
        return _apply_policy(contribution, key, exc, store)
    if store is not None and contribution.failure_policy is SourceFailurePolicy.LAST_KNOWN_GOOD:
        store.put(key, read)
    return SourceResolution(read=read)

resolve_strategy ¶

resolve_strategy(name: str | None = None, *, jit_context: bool | None = None) -> ContextStrategy

Resolve the strategy from an explicit name, falling back to the legacy flag.

Precedence is stated once, here, so the facade and the compiler cannot disagree about which strategy a given configuration selects — the classic "two factories, two answers" defect.

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

def resolve_strategy(
    name: str | None = None, *, jit_context: bool | None = None
) -> ContextStrategy:
    """Resolve the strategy from an explicit name, falling back to the legacy flag.

    Precedence is stated once, here, so the facade and the compiler cannot
    disagree about which strategy a given configuration selects — the classic
    "two factories, two answers" defect.
    """
    if name is None:
        if jit_context is None:
            return ContextStrategy.STRATIFIED
        return ContextStrategy.JIT if jit_context else ContextStrategy.STRATIFIED
    try:
        return ContextStrategy(name)
    except ValueError as exc:
        expected = ", ".join(repr(member.value) for member in ContextStrategy)
        raise ValueError(
            f"unknown context strategy {name!r}; expected one of {expected}."
        ) from exc

sensitive_terms ¶

sensitive_terms(guardrails: Iterable[Guardrail]) -> tuple[str, ...]

The terms a governance stage should watch, from the rules themselves.

MetacognitionStage takes terms and guardrails carry tags; deriving one from the other is what stops them drifting. Threaded by hand, a tag added to a rule and forgotten in the stage's configuration is a rule that renders and is never enforced.

Sorted and deduplicated, because an unordered set makes a stage's configuration differ between runs for no reason a reader can see.

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

def sensitive_terms(guardrails: Iterable[Guardrail]) -> tuple[str, ...]:
    """The terms a governance stage should watch, from the rules themselves.

    ``MetacognitionStage`` takes terms and guardrails carry tags; deriving one
    from the other is what stops them drifting. Threaded by hand, a tag added
    to a rule and forgotten in the stage's configuration is a rule that renders
    and is never enforced.

    Sorted and deduplicated, because an unordered set makes a stage's
    configuration differ between runs for no reason a reader can see.
    """
    return tuple(
        sorted({tag.strip() for rule in guardrails for tag in rule.sensitive_tags if tag.strip()})
    )

strategy_admits ¶

strategy_admits(strategy: ContextStrategy, contribution: PromptContribution) -> bool

True when strategy assembles this contribution at all.

The only axis is hydration: JIT declines contributions whose content only exists after a retrieval pass, because performing that pass is exactly what JIT exists to avoid.

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

def strategy_admits(strategy: ContextStrategy, contribution: PromptContribution) -> bool:
    """``True`` when ``strategy`` assembles this contribution at all.

    The only axis is hydration: JIT declines contributions whose content only
    exists after a retrieval pass, because performing that pass is exactly what
    JIT exists to avoid.
    """
    return not (strategy is ContextStrategy.JIT and contribution.requires_hydration)

to_prompt_assembly ¶

to_prompt_assembly(compiled: CompiledPrompt, *, prompt: str, attachments: tuple[Any, ...] = (), history: tuple[Any, ...] = ()) -> PromptAssembly

Project a compiled prompt into the kernel's prompt-assembly value.

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

def to_prompt_assembly(
    compiled: CompiledPrompt,
    *,
    prompt: str,
    attachments: tuple[Any, ...] = (),
    history: tuple[Any, ...] = (),
) -> PromptAssembly:
    """Project a compiled prompt into the kernel's ``prompt-assembly`` value."""
    return PromptAssembly(
        instructions=compiled_instructions(compiled),
        prompt=prompt,
        attachments=attachments,
        history=history,
        # ``CompiledPrompt.text`` remains the portable, readable projection.
        # The blocks retain the separate cache annotations until the provider
        # adapter makes its dialect decision.
        system_blocks=tuple(compiled.cache_annotations()),
    )

untrusted_open_tag ¶

untrusted_open_tag(contribution_id: str) -> str

The opening delimiter for contribution_id.

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

def untrusted_open_tag(contribution_id: str) -> str:
    """The opening delimiter for ``contribution_id``."""
    return f'<{UNTRUSTED_TAG} source="{sanitise_name(contribution_id)}">'

symfonic.capabilities.prompting¶

prompting ¶

AsyncContributionSource ¶

BudgetReport dataclass ¶

CacheDirective dataclass ¶

rank property ¶

marker ¶

CacheRegion dataclass ¶

annotation ¶

CacheTtl ¶

CompiledPrompt dataclass ¶

text property ¶

cache_annotations ¶

region_of ¶

ContextStrategy ¶

ContributionScope ¶

ContributionSource ¶

DomainPersona dataclass ¶

GateResult dataclass ¶

Guardrail dataclass ¶

HeuristicTokenEstimator dataclass ¶

InMemoryLastKnownGood dataclass ¶

Layer ¶

ProceduralSkillsSource dataclass ¶

aread async ¶

PromptBudget dataclass ¶

narrows ¶

PromptCompileRequest dataclass ¶

PromptContribution dataclass ¶

failure_policy property ¶

is_learned property ¶

validate ¶

PromptContributor ¶

PromptDiagnostic dataclass ¶

PromptingCapability dataclass ¶

contribute ¶

RegionRow ¶

RenderPolicy dataclass ¶

RenderedContribution dataclass ¶

SourceFailurePolicy ¶

SourceRead dataclass ¶

SourceRequest dataclass ¶

SourceResolution dataclass ¶

StaticSource dataclass ¶

TokenEstimator ¶

TrustTier ¶

admit_within_budget ¶

cache_directive_for ¶

compile_prompt ¶

compile_prompt_async async ¶

compiled_instructions ¶

default_failure_policy ¶

derive_child_request ¶

guardrail_sources ¶

inheritable ¶

is_async_source ¶

is_sync_source ¶

is_volatile ¶

isolation_key ¶

layer_index ¶

neutralise_delimiters ¶

normalise_ttl_ladder ¶

ordering_key ¶

persona_sources ¶

plan_regions ¶

prompt_stage_descriptor ¶

render_contribution ¶

resolve_source ¶

resolve_source_async async ¶

resolve_strategy ¶

sensitive_terms ¶

strategy_admits ¶

to_prompt_assembly ¶

untrusted_open_tag ¶

`symfonic.capabilities.prompting`¶

BudgetReport `dataclass` ¶

CacheDirective `dataclass` ¶

rank `property` ¶

CacheRegion `dataclass` ¶

CompiledPrompt `dataclass` ¶

text `property` ¶

DomainPersona `dataclass` ¶

GateResult `dataclass` ¶

Guardrail `dataclass` ¶

HeuristicTokenEstimator `dataclass` ¶

InMemoryLastKnownGood `dataclass` ¶

ProceduralSkillsSource `dataclass` ¶

aread `async` ¶

PromptBudget `dataclass` ¶

PromptCompileRequest `dataclass` ¶

PromptContribution `dataclass` ¶

failure_policy `property` ¶

is_learned `property` ¶

PromptDiagnostic `dataclass` ¶

PromptingCapability `dataclass` ¶

RenderPolicy `dataclass` ¶

RenderedContribution `dataclass` ¶

SourceRead `dataclass` ¶

SourceRequest `dataclass` ¶

SourceResolution `dataclass` ¶

StaticSource `dataclass` ¶

compile_prompt_async `async` ¶

resolve_source_async `async` ¶