Skip to content

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 
78
79
80
81
82
83
84
85
86
87
88
89
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 
126
127
128
129
130
131
132
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.

selected_line_candidates property 

selected_line_candidates: tuple[object, ...]

Non-authoritative hydration rows that survived the compiler budget.

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 
91
92
93
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 
 95
 96
 97
 98
 99
100
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.

ExtractionDirectiveSource dataclass 

ExtractionDirectiveSource(template_path: Path | None = None, cadence: int = 1, scope_path: str = '', untrusted: bool = False, offline_safe: bool = True, scope_aware: bool = True)

Render a custom or bundled extraction directive on its declared cadence.

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.

JitManifestSource dataclass 

JitManifestSource(manifest: Sequence[str], policy: RegionalPromptPolicy, untrusted: bool = False, offline_safe: bool = True, scope_aware: bool = False)

Render a fixed tool manifest under the policy's independent JIT cap.

The caller supplies the manifest it has already authorised for the turn. This source deliberately does not discover tools or inspect providers.

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, include_platform: bool = False, 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 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
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 = []

    shared: list = []
    if self.include_platform:
        from symfonic.memory.layers.procedural.sharing import platform_scope

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

    rendered = [render_skill_for_prompt(skill) for skill in skills or ()]
    own_labels = {_skill_label(skill) for skill in skills or ()}
    for skill in shared:
        label = _skill_label(skill)
        if label and label in own_labels:
            continue
        rendered.append(_mark_shared(render_skill_for_prompt(skill)))
    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 
71
72
73
74
75
76
77
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, session_id: str = '', 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, jit_only: 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 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
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 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
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))
        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")
        # 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(turn_request)
        session_id = str(getattr(turn_request, "session_id", "") or "")
        options = _turn_options(self.options, scope_path, session_id)
        # 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
        )
        run_id = getattr(context, "run_id", "")
        receipt = None
        if run_id:
            receipt = PromptAdmissionReceipt(
                run_id=run_id,
                scope_path=scope_path,
                session_id=session_id,
                compiler_scope_path=options.get("scope_path", ""),
                compiler_session_id=options.get("session_id", ""),
                admitted=turn_compiled.budget.admitted,
                dropped=turn_compiled.budget.dropped,
            )
        if not turn_compiled.text:
            return no_change(
                "the prompt compiled to nothing: no contribution survived "
                "its gates, or none was supplied",
                counts=prompt_counts(turn_compiled),
                prompt_admission=receipt,
            )
        assembly = to_prompt_assembly(
            turn_compiled,
            prompt=turn_request.prompt,
            attachments=tuple(turn_request.attachments),
            history=tuple(turn_request.history),
            append_user_message=getattr(turn_request, "append_user_message", True),
        )
        return applied(
            _compose(context, assembly),
            counts=prompt_counts(turn_compiled),
            prompt_admission=receipt,
        )

    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.

RegionalBudgetSource dataclass 

RegionalBudgetSource(source: _ReadableSource, policy: RegionalPromptPolicy, region: str = 'hms_system_prompt')

Apply an advisory or strict boundary to an authored prompt source.

RegionalPromptPolicy dataclass 

RegionalPromptPolicy(hms_system_prompt_token_budget: int | None = None, jit_manifest_token_budget: int = 0, hms_budget_mode: Literal['warn', 'strict'] = 'warn', on_hms_budget_exceeded: BudgetObserver | None = None, domain_description_max_chars: tuple[int, int] = (2000, 400))

The explicit legacy-region limits a native composition can honour.

hms_system_prompt_token_budget guards only the authored HMS source. jit_manifest_token_budget controls only the manifest source. The compiler's own PromptBudget remains a distinct total-prompt ceiling.

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, selected_line_candidates: tuple[object, ...] = ())

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, session_id: str = '', 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 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
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),
    )

bounded_domain_description 

bounded_domain_description(description: str, policy: RegionalPromptPolicy, *, jit: bool) -> str

Return the model-visible domain description for the selected surface.

Source code in src/symfonic/capabilities/prompting/regional.py 
80
81
82
83
84
85
86
87
def bounded_domain_description(
    description: str, policy: RegionalPromptPolicy, *, jit: bool
) -> str:
    """Return the model-visible domain description for the selected surface."""
    if not description:
        return "(none)" if jit else "(no domain-specific directives)"
    cap = policy.domain_description_max_chars[1 if jit else 0]
    return description if len(description) <= cap else description[:cap] + "... [truncated]"

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 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
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,
                session_id=request.session_id,
                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 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
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,
                session_id=request.session_id,
                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 
36
37
38
39
40
41
42
43
44
45
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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),
    )

extraction_contribution 

extraction_contribution(source: ExtractionDirectiveSource, *, order: int = 900) -> Any

Declare the directive in L1, after stable system instructions.

Source code in src/symfonic/capabilities/prompting/extraction.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def extraction_contribution(source: ExtractionDirectiveSource, *, order: int = 900) -> Any:
    """Declare the directive in L1, after stable system instructions."""
    from symfonic.capabilities.prompting.contracts import (
        ContributionScope,
        PromptContribution,
        TrustTier,
    )
    from symfonic.capabilities.prompting.layers import Layer

    return PromptContribution(
        contribution_id="memory.extraction.directive",
        source=source,
        capability="memory",
        layer=Layer.L1,
        tier=TrustTier.PLATFORM,
        scope=ContributionScope.TENANT,
        order=order,
    )

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 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
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
    )

hms_source_for_domain 

hms_source_for_domain(state: Mapping[str, object], domain: object, policy: RegionalPromptPolicy) -> _ReadableSource

Bind the supported domain.description field into an HMS source.

The caller still supplies the template's other authored values. Domain fields that belong to tools or governance are intentionally not projected into this prompt source.

Source code in src/symfonic/capabilities/prompting/regional.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def hms_source_for_domain(
    state: Mapping[str, object], domain: object, policy: RegionalPromptPolicy
) -> _ReadableSource:
    """Bind the supported ``domain.description`` field into an HMS source.

    The caller still supplies the template's other authored values.  Domain
    fields that belong to tools or governance are intentionally not projected
    into this prompt source.
    """
    return _DomainHmsSource(
        dict(state), str(getattr(domain, "description", "") or ""), policy
    )

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 
41
42
43
44
45
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 
137
138
139
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 
159
160
161
162
163
164
165
166
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 
49
50
51
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 
30
31
32
33
34
35
36
37
38
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)

jit_manifest_contribution 

jit_manifest_contribution(source: JitManifestSource, *, order: int = 100) -> PromptContribution

Declare the manifest in L1; callers use it only in a JIT composition.

Source code in src/symfonic/capabilities/prompting/regional.py 
226
227
228
229
230
231
232
233
234
235
236
def jit_manifest_contribution(source: JitManifestSource, *, order: int = 100) -> PromptContribution:
    """Declare the manifest in L1; callers use it only in a JIT composition."""
    return PromptContribution(
        contribution_id="jit.tool_manifest",
        source=source,
        capability="prompting",
        layer=Layer.L1,
        tier=TrustTier.PLATFORM,
        scope=ContributionScope.DEPLOYMENT,
        order=order,
    )

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 
44
45
46
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 
120
121
122
123
124
125
126
127
128
129
130
131
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 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
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 
231
232
233
234
235
236
237
238
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 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
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 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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),
        },
    )

regional_hms_contribution 

regional_hms_contribution(source: _ReadableSource, policy: RegionalPromptPolicy, *, order: int = -1000) -> PromptContribution

Declare the HMS section as a pinned authored L0 region.

Source code in src/symfonic/capabilities/prompting/regional.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def regional_hms_contribution(
    source: _ReadableSource,
    policy: RegionalPromptPolicy,
    *,
    order: int = -1000,
) -> PromptContribution:
    """Declare the HMS section as a pinned authored L0 region."""
    return PromptContribution(
        contribution_id="hms.system",
        source=RegionalBudgetSource(source=source, policy=policy),
        capability="prompting",
        layer=Layer.L0,
        tier=TrustTier.PLATFORM,
        scope=ContributionScope.DEPLOYMENT,
        order=order,
        pinned=True,
    )

regional_prompting_capability 

regional_prompting_capability(hms_source: _ReadableSource, policy: RegionalPromptPolicy, *, strategy: Literal['jit', 'stratified'] = 'stratified', manifest: Sequence[str] = ()) -> object

Build the supported regional composition with explicit JIT intent.

A non-zero manifest budget is only effective when a JIT manifest is supplied and the composition declares strategy='jit'. Refusing the other combinations prevents a deployment from carrying a configuration value that cannot affect the model request.

Source code in src/symfonic/capabilities/prompting/regional.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def regional_prompting_capability(
    hms_source: _ReadableSource,
    policy: RegionalPromptPolicy,
    *,
    strategy: Literal["jit", "stratified"] = "stratified",
    manifest: Sequence[str] = (),
) -> object:
    """Build the supported regional composition with explicit JIT intent.

    A non-zero manifest budget is only effective when a JIT manifest is
    supplied and the composition declares ``strategy='jit'``.  Refusing the
    other combinations prevents a deployment from carrying a configuration
    value that cannot affect the model request.
    """
    if policy.jit_manifest_token_budget and not manifest:
        raise ValueError(
            "jit_manifest_token_budget needs a manifest; set it to zero when "
            "the JIT manifest is not composed"
        )
    if manifest and strategy != "jit":
        raise ValueError("a JIT manifest requires strategy='jit'")
    from symfonic.capabilities.prompting.capability import PromptingCapability
    from symfonic.capabilities.prompting.strategies import ContextStrategy

    if isinstance(hms_source, _DomainHmsSource):
        # A composition gets its own notice lifetime; never mutate a shared source.
        hms_source = _DomainHmsSource(
            dict(hms_source.state), hms_source.description, hms_source.policy, jit=strategy == "jit"
        )
    sources: list[PromptContribution] = [regional_hms_contribution(hms_source, policy)]
    if manifest:
        sources.append(jit_manifest_contribution(JitManifestSource(manifest, policy)))
    return PromptingCapability(
        sources=sources,
        options={"strategy": ContextStrategy(strategy)},
    )

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 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
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, session_id: str = '', 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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def resolve_source(
    contribution: PromptContribution,
    *,
    scope_path: str = "",
    turn: int = 0,
    session_id: str = "",
    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, session_id=session_id
    )
    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, session_id: str = '', 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 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
async def resolve_source_async(
    contribution: PromptContribution,
    *,
    scope_path: str = "",
    turn: int = 0,
    session_id: str = "",
    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, session_id=session_id
    )
    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 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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 
60
61
62
63
64
65
66
67
68
69
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.
    """
    if strategy is ContextStrategy.JIT:
        return not contribution.requires_hydration
    return not contribution.jit_only

to_prompt_assembly 

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

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

Source code in src/symfonic/capabilities/prompting/assembly.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def to_prompt_assembly(
    compiled: CompiledPrompt,
    *,
    prompt: str,
    attachments: tuple[Any, ...] = (),
    history: tuple[Any, ...] = (),
    append_user_message: bool = True,
) -> PromptAssembly:
    """Project a compiled prompt into the kernel's ``prompt-assembly`` value."""
    return PromptAssembly(
        instructions=compiled_instructions(compiled),
        prompt=prompt,
        attachments=attachments,
        history=history,
        append_user_message=append_user_message,
        # ``CompiledPrompt.text`` remains the portable, readable projection.
        # The blocks retain the separate cache annotations until the provider
        # adapter makes its dialect decision.
        _selected_line_candidates=compiled.selected_line_candidates,
        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 
115
116
117
def untrusted_open_tag(contribution_id: str) -> str:
    """The opening delimiter for ``contribution_id``."""
    return f'<{UNTRUSTED_TAG} source="{sanitise_name(contribution_id)}">'