`symfonic.kernel.contracts`¶

contracts ¶

kernel.contracts — the vocabulary every layer is allowed to share.

Standard library only, by rule: this package is the one thing capabilities, services, integrations and the facade may all import, so anything it pulls in becomes a dependency of the entire framework. The single internal import is the error taxonomy in symfonic.core.contracts.errors, which is the kernel contract set's interim home until the package move lands (INV-ADR §2, FERR-1).

AdapterPressure ¶

Bases: Protocol

The BP-12 numbers an adapter exports for one run.

Structural on purpose: kernel.backpressure.AdapterMetrics satisfies it without importing this module, so the contract package keeps its rule of depending on nothing.

BackgroundEntry `dataclass` ¶

BackgroundEntry(owner: str, purpose: str, deadline_seconds: float | None = None)

One registered unit of run-owned work (R7).

An entry, not a bare task: a task set records that something is running, while owner/purpose/deadline record who to ask when it is still running at teardown — which is the question a leaked task always raises.

CheckpointerPort ¶

Bases: Protocol

Durable run state: made ready before effects, flushed before close.

ensure_ready is separate from construction because readiness can open external resources (a pool, a schema migration) and must therefore be lazy, idempotent, and — critically — registered for teardown at the moment it succeeds, not at the call site that happened to trigger it.

close `async` ¶

close() -> None

Release the durable handle, flushed or not.

Source code in src/symfonic/kernel/contracts/lifecycle.py

async def close(self) -> None:
    """Release the durable handle, flushed or not."""

ensure_ready `async` ¶

ensure_ready() -> None

Open whatever durable state this run needs. Idempotent per run.

Source code in src/symfonic/kernel/contracts/lifecycle.py

async def ensure_ready(self) -> None:
    """Open whatever durable state this run needs. Idempotent per run."""

flush `async` ¶

flush() -> None

Push buffered writes. Failure is reported; it never blocks close.

Source code in src/symfonic/kernel/contracts/lifecycle.py

async def flush(self) -> None:
    """Push buffered writes. Failure is reported; it never blocks close."""

CompiledStage `dataclass` ¶

CompiledStage(stage_id: str, phase: Phase, capability: str, priority: int, effects: frozenset[str] = frozenset(), emits: frozenset[str] = frozenset(), kind: StageKind = StageKind.COMPILATION, config: Mapping[str, Any] = (lambda: MappingProxyType({}))(), tie_break: str = 'first-in-phase')

A stage with its position decided and the reason recorded (STG-12).

Coerces kind for the same reason :class:StageDescriptor does, and it matters more here: this is the type the dispatcher actually reads, and its identity comparison is only valid because of a guarantee this class has to make itself. ordering.py copies an already-coerced descriptor, so the shipped path was safe -- but CompiledStage and StageProgram are public, and any other construction site got the vanishing-stage defect back with no refusal anywhere.

ConversationPort ¶

Bases: Protocol

Owns the transcript's shape; the kernel only owns its sequence.

close_round ¶

close_round(transcript: Any, turn: ModelTurn, requests: Sequence[ToolRequest], outcomes: Sequence[Any]) -> None

Append one completed round — assistant reply, then tool observations.

Source code in src/symfonic/kernel/contracts/ports.py

def close_round(
    self,
    transcript: Any,
    turn: ModelTurn,
    requests: Sequence[ToolRequest],
    outcomes: Sequence[Any],
) -> None:
    """Append one completed round — assistant reply, then tool observations."""

messages ¶

messages(transcript: Any) -> tuple[Any, ...]

Return the turn's messages in order, for the invocation outcome.

Source code in src/symfonic/kernel/contracts/ports.py

def messages(self, transcript: Any) -> tuple[Any, ...]:
    """Return the turn's messages in order, for the invocation outcome."""

open_turn ¶

open_turn(assembly: PromptAssembly) -> Any

Return an opaque transcript handle seeded with the assembled prompt.

Source code in src/symfonic/kernel/contracts/ports.py

def open_turn(self, assembly: PromptAssembly) -> Any:
    """Return an opaque transcript handle seeded with the assembled prompt."""

DiagnosticRecord `dataclass` ¶

DiagnosticRecord(category: str, subject: str, detail: str)

One fact about how this plan was compiled.

EventAdapter `dataclass` ¶

EventAdapter(name: str, buffer: BufferClass = 'rendezvous', policy: BufferPolicy = 'block', capacity: int = 0, byte_capacity: int = 0, terminal_policy: TerminalPolicy = 'reserve', owner: str = 'kernel', emits: frozenset[str] = frozenset(), sheddable: frozenset[str] = frozenset(), coalescible: frozenset[str] = frozenset(), text_reconstruction: Literal['deltas', 'terminal-only', 'transport-dependent'] = 'deltas')

One G9 adapter row, including every BP-1 policy decision.

capacity and byte_capacity are per run. Rendezvous adapters use zero for both. text_reconstruction='transport-dependent' means streaming transport publishes deltas while blocking transport publishes its complete round text as one text_delta. terminal_policy='shed' exists only so compilation can reject it with a useful BP-4 error.

EventProgram `dataclass` ¶

EventProgram(emitted: frozenset[str] = frozenset(), adapters: tuple[EventAdapter, ...] = ())

G9 — the declared event types and every adapter attached to them.

A buffer that is not declared here may not exist at run time (BP-1), which is only enforceable because the declaration is compiled data.

FinalizerFailure `dataclass` ¶

FinalizerFailure(name: str, error: str)

A finalizer that raised or overran its budget, named so it can be fixed.

InvocationOutcome `dataclass` ¶

InvocationOutcome(text: str, output: Any = None, messages: tuple[Any, ...] = (), tool_outcomes: tuple[Any, ...] = (), usage: UsageDelta = UsageDelta(), run_id: str = '', duration_ms: float = 0.0, stop_reason: str | None = None)

Everything one invocation produced, before result-shape adaptation.

messages and tool_outcomes are opaque to the kernel: it collects and orders them, and the bound response port decides what they look like to an adopter. That split is why T2.3.3 can change the public result shape without touching the loop.

InvocationPaused ¶

InvocationPaused(interrupt: PendingInterrupt)

Bases: BaseException

The run stopped at a registered pause point. Not an error (HK1).

Raised by whatever reaches the pause point -- a capability's pre-tool stage today -- and caught in exactly one place: symfonic.kernel.runner.InvocationRunner.events, which turns it into the run's terminal event. It is never converted into an error terminal and never re-raised at the consumer, because a paused run did not fail.

Source code in src/symfonic/kernel/contracts/interrupts.py

def __init__(self, interrupt: PendingInterrupt) -> None:
    super().__init__(f"the run paused at {interrupt.name!r}")
    self.interrupt = interrupt

InvocationPlan `dataclass` ¶

InvocationPlan(identity: PlanIdentity, scope: RequestScope, model: ModelResolution, tool_manifest: ToolManifest, stage_program: StageProgram, bindings: ServiceBindings, effect_grants: frozenset[str], limits: PlanLimits, event_program: EventProgram, diagnostics: PlanDiagnostics)

The ten frozen field groups of INV-ADR §2.

Deep-immutable over G1–G5 and G7–G10; G6 holds live port objects whose set and identity are frozen while their internals are not (IPL-4).

allows_tool ¶

allows_tool(name: str) -> bool

Whether name is on the single manifest (G4). Never recomputed.

Source code in src/symfonic/kernel/contracts/plan.py

def allows_tool(self, name: str) -> bool:
    """Whether ``name`` is on the single manifest (G4). Never recomputed."""
    return name in self.tool_manifest.names()

equality_key ¶

equality_key() -> tuple[Any, ...]

The reproducibility key: everything but G1's ids and G6's objects.

Two compilations of the same normalized config with the same registered capabilities must agree on this tuple. plan_id and compiled_at are excluded because they are per-compilation by construction, and the bindings because object identity is not a property of a configuration.

Source code in src/symfonic/kernel/contracts/plan.py

def equality_key(self) -> tuple[Any, ...]:
    """The reproducibility key: everything but G1's ids and G6's objects.

    Two compilations of the same normalized config with the same registered
    capabilities must agree on this tuple. ``plan_id`` and ``compiled_at``
    are excluded because they are per-compilation by construction, and the
    bindings because object identity is not a property of a configuration.
    """
    return (
        self.identity.config_digest,
        self.identity.schema_version,
        self.scope,
        self.model,
        self.tool_manifest,
        self.stage_program,
        self.effect_grants,
        self.limits,
        self.event_program,
        self.diagnostics,
    )

grants ¶

grants(effect: str) -> bool

Whether this invocation may exercise effect (STG-8, fail-closed).

Source code in src/symfonic/kernel/contracts/plan.py

def grants(self, effect: str) -> bool:
    """Whether this invocation may exercise ``effect`` (STG-8, fail-closed)."""
    return effect in self.effect_grants

KernelEvent `dataclass` ¶

KernelEvent(kind: str, index: int, run_id: str, text: str | None = None, tool_request: ToolRequest | None = None, tool_outcome: Any = None, outcome: InvocationOutcome | None = None, error: str | None = None, stage_id: str | None = None, phase: str | None = None, capability: str | None = None, stage_outcome: str | None = None, stage_reason: str | None = None, counts: Mapping[str, int] = (lambda: EMPTY_COUNTS)(), dropped_kind: str | None = None, dropped_count: int = 0, first_dropped_index: int | None = None, last_dropped_index: int | None = None, termination: str | None = None, interrupt: PendingInterrupt | None = None)

One event on the single internal pipeline (CADR-10, EVT-1…EVT-10).

index is dense and monotonic within a run, and the kernel is the only thing that assigns it — an adapter that renumbered events would break the ordering guarantee every consumer joins on.

ModelDelta `dataclass` ¶

ModelDelta(kind: str, text: str)

One incremental piece of a streaming round.

ModelPort ¶

Bases: Protocol

Executes the model decision already made in plan group G3 (CADR-02).

It never re-picks a model, and it never silently downgrades a pinned one.

invoke `async` ¶

invoke(transcript: Any) -> ModelTurn

Run one non-streaming round.

Source code in src/symfonic/kernel/contracts/ports.py

async def invoke(self, transcript: Any) -> ModelTurn:
    """Run one non-streaming round."""

stream ¶

stream(transcript: Any) -> ModelRound

Begin one streaming round.

Source code in src/symfonic/kernel/contracts/ports.py

def stream(self, transcript: Any) -> ModelRound:
    """Begin one streaming round."""

ModelResolution `dataclass` ¶

ModelResolution(provider_family: str = 'unknown', model_name: str | None = None, sampling: Mapping[str, Any] = (lambda: MappingProxyType({}))(), response_format: ResponseFormat = ResponseFormat())

G3 — the model decision, already made. The router executes it (CADR-02).

ModelRound ¶

Bases: Protocol

One streaming round: deltas first, then the aggregate they add up to.

deltas ¶

deltas() -> AsyncIterator[ModelDelta]

Yield incremental pieces as the provider produces them.

Source code in src/symfonic/kernel/contracts/ports.py

def deltas(self) -> AsyncIterator[ModelDelta]:
    """Yield incremental pieces as the provider produces them."""

result ¶

result() -> ModelTurn

The completed round. Valid only once deltas() is exhausted.

Source code in src/symfonic/kernel/contracts/ports.py

def result(self) -> ModelTurn:
    """The completed round. Valid only once ``deltas()`` is exhausted."""

ModelTurn `dataclass` ¶

ModelTurn(text: str = '', tool_requests: tuple[ToolRequest, ...] = (), usage: UsageDelta = UsageDelta(), stop_reason: str | None = None, payload: Any = None)

One completed provider round, in kernel vocabulary.

payload is the adapter's own object — the raw provider message, or whatever else it needs back when the kernel asks it to record the round. The kernel carries it and never reads it: a field the loop cannot inspect is a field the loop cannot come to depend on.

PendingInterrupt `dataclass` ¶

PendingInterrupt(name: str, kind: str = 'interrupt', payload: Any = None, token: str = '', tool_call_id: str = '', interrupt_id: str = '', run_id: str = '', session_id: str = '', expires_at: float | None = None, resumable: bool = False)

One pause: what was asked, who may answer, and with which token.

payload is carried, never interpreted. The kernel does not know what an ask_user request looks like and must not learn: the capability that registered the interaction validated the payload against its own schema before minting, and the projection that publishes it is the layer that knows which public event shape it becomes.

resumable is a declaration, not a wish. TA8.34 mints a pause and publishes it; nothing in this build redeems one on the kernel route -- resume, rehydration and the checkpointer a resume needs are TA8.35. A pause that cannot be resumed has to say so, because the alternative is a consumer holding a token that looks live and answering into nothing.

Phase ¶

Bases: StrEnum

The eight phases, in the only order they ever run (STG-1).

PlanDiagnostics `dataclass` ¶

PlanDiagnostics(records: tuple[DiagnosticRecord, ...] = ())

The whole compile record, ordered as the compiler produced it.

explain ¶

explain() -> str

Render the record as safe-to-log text (ERR-5).

Source code in src/symfonic/kernel/contracts/diagnostics.py

def explain(self) -> str:
    """Render the record as safe-to-log text (ERR-5)."""
    return "\n".join(record.render() for record in self.records)

PlanIdentity `dataclass` ¶

PlanIdentity(plan_id: str, config_digest: str, compiled_at: float, schema_version: int = 1, parent_plan_id: str | None = None)

G1 — who this plan is, and which plan it was derived from.

PlanLimits `dataclass` ¶

PlanLimits(max_model_rounds: int = 10, max_recursion_depth: int = 1, deadline_seconds: float | None = None, max_event_buffer: int = 256, teardown_grace_seconds: float = 5.0)

G8 — every ceiling this invocation runs under, including the round bound.

max_model_rounds lives here rather than in a module constant on purpose: a bound that a plan cannot express is a bound no adopter can tune and no test can shorten.

teardown_grace_seconds is the same argument applied to unwinding: it bounds each finalizer, the background-work drain, and terminal delivery (RCX-10, BP-8, BP-10). A run whose consumer has gone away must still finish dying in a knowable amount of time.

PressureRecord `dataclass` ¶

PressureRecord(adapter: str, high_watermark: int = 0, byte_high_watermark: int = 0, blocked_seconds: float = 0.0, events_shed: Mapping[str, int] = (lambda: MappingProxyType({}))(), terminal_delivery_failed: bool = False, abandoned: bool = False)

One adapter's per-run pressure, frozen into the report (BP-12).

PromptAssembly `dataclass` ¶

PromptAssembly(instructions: str | None, prompt: str, attachments: tuple[Any, ...] = (), history: tuple[Any, ...] = (), system_blocks: tuple[Any, ...] = ())

The output of the prompt-assembly phase (STG-7: a pure function).

It is a value, not a side effect, precisely so the phase stays pure and the assembled prompt is reproducible from the plan and the request alone.

RequestScope `dataclass` ¶

RequestScope(tenant: str | None = None, principal: str | None = None, grants: frozenset[str] = frozenset())

G2 — the scope the platform derived, carried and never recomputed.

narrows ¶

narrows(parent: RequestScope) -> bool

True when this scope is the parent's or a strict narrowing of it.

Source code in src/symfonic/kernel/contracts/groups.py

def narrows(self, parent: RequestScope) -> bool:
    """True when this scope is the parent's or a strict narrowing of it."""
    if self == parent:
        return True
    tenant_ok = parent.tenant is None or self.tenant == parent.tenant
    principal_ok = parent.principal is None or self.principal == parent.principal
    return tenant_ok and principal_ok and self.grants <= parent.grants

ResourcePort ¶

Bases: Protocol

Anything a run acquires and must give back — a pool, a socket, a lease.

One method. A resource that also needed flushing is a checkpointer, and a resource that needed a health check is a resource plus a stage; widening this port would let every future collaborator negotiate its own teardown, which is precisely the sprawl this task removes.

name `property` ¶

name: str

A stable label, used when the release is reported as having failed.

release `async` ¶

release() -> None

Give the resource back. Called exactly once, on every exit path.

Source code in src/symfonic/kernel/contracts/lifecycle.py

async def release(self) -> None:
    """Give the resource back. Called exactly once, on every exit path."""

ResponseFormat `dataclass` ¶

ResponseFormat(mode: Literal['text', 'structured'] = 'text', schema_name: str | None = None)

How this invocation's response is handled.

A descriptor, not a schema object: the live Pydantic model (or whatever the adopter's schema library produces) is reachable only through the bound response port, which is what keeps IPL-7's "a plan is serializable minus G6" true when structured output is in play.

ResponsePort ¶

Bases: Protocol

Response handling: structured extraction plus result/event adaptation.

structured `property` ¶

structured: bool

Whether a terminal structured-output pass is owed (RES-8).

build_event ¶

build_event(event: KernelEvent) -> Any

Project one kernel event onto the caller's event type.

Source code in src/symfonic/kernel/contracts/ports.py

def build_event(self, event: KernelEvent) -> Any:
    """Project one kernel event onto the caller's event type."""

build_result ¶

build_result(outcome: InvocationOutcome) -> Any

Project the kernel outcome onto the caller's result type.

Source code in src/symfonic/kernel/contracts/ports.py

def build_result(self, outcome: InvocationOutcome) -> Any:
    """Project the kernel outcome onto the caller's result type."""

extract `async` ¶

extract(transcript: Any) -> Any

Run the terminal extraction pass against the final transcript.

Source code in src/symfonic/kernel/contracts/ports.py

async def extract(self, transcript: Any) -> Any:
    """Run the terminal extraction pass against the final transcript."""

ServiceBindings `dataclass` ¶

ServiceBindings(conversation: Any = None, model: Any = None, palette: Any = None, tools: Any = None, response: Any = None, event_sink: Any = None, stage_handlers: Mapping[str, Any] = (lambda: MappingProxyType({}))(), tool_preconditions: tuple[Any, ...] = ())

G6 — live port objects, fixed in identity at freeze time (IPL-4).

These are the only collaborators the kernel ever calls. Rebinding requires a new plan; there is no setter, and no code path that swaps one mid-run.

StageDescriptor `dataclass` ¶

StageDescriptor(stage_id: str, phase: Phase, capability: str = _KERNEL_CAPABILITY, priority: int = 0, after: tuple[str, ...] = (), before: tuple[str, ...] = (), optional_after: tuple[str, ...] = (), optional_before: tuple[str, ...] = (), effects: frozenset[str] = frozenset(), emits: frozenset[str] = frozenset(), kind: StageKind = StageKind.COMPILATION, config: Mapping[str, Any] = (lambda: MappingProxyType({}))())

One capability's declaration of a stage it contributes (CON-C-1).

validate ¶

validate() -> None

Shape validation, performed the moment a capability registers (STG-4).

Resolution of the constraints against the full stage set happens later, at compile time, when the whole set is finally knowable.

Source code in src/symfonic/kernel/contracts/stages.py

def validate(self) -> None:
    """Shape validation, performed the moment a capability registers (STG-4).

    Resolution of the constraints against the full stage set happens later,
    at compile time, when the whole set is finally knowable.
    """
    if not self.stage_id:
        raise ConfigurationError("a stage descriptor must declare a non-empty stage_id.")
    if self.phase not in PHASE_LADDER:
        raise ConfigurationError(
            f"stage {self.stage_id!r} declares unknown phase {self.phase!r}; "
            f"the ladder is {[phase.value for phase in PHASE_LADDER]}."
        )
    if self.phase in KERNEL_OWNED_PHASES and self.capability != _KERNEL_CAPABILITY:
        raise ConfigurationError(
            f"stage {self.stage_id!r} (capability {self.capability!r}) targets the "
            f"kernel-owned phase {self.phase.value!r}; only the kernel registers "
            "into 'bind' and 'teardown'."
        )
    constrained = (
        *self.after,
        *self.before,
        *self.optional_after,
        *self.optional_before,
    )
    if self.stage_id in constrained:
        raise ConfigurationError(
            f"stage {self.stage_id!r} declares a constraint on itself."
        )
    # Checked here rather than at dispatch: an unknown family that reaches
    # dispatch is indistinguishable from one that was simply not granted,
    # so a typo would read as a policy refusal.
    require_known_families(self.effects, subject=f"stage {self.stage_id!r}")
    # Scoped to PROMPT_ASSEMBLY on purpose. Only that phase is split today,
    # because only that phase has a pure consumer that a prior effect must
    # not contaminate. A POST_MODEL stage writing to memory is effectful and
    # has no compiler downstream to keep pure, so the distinction has
    # nothing to say about it -- and applying the rule there would have made
    # every effectful stage outside P1 declare a kind that means nothing.
    if (
        self.phase is Phase.PROMPT_ASSEMBLY
        and self.kind is StageKind.COMPILATION
        and self.effects
    ):
        raise ConfigurationError(
            f"stage {self.stage_id!r} (capability {self.capability!r}) is a "
            f"compilation stage and declares effect(s) {sorted(self.effects)}. "
            "STG-7: compilation is a pure function of (plan, request, "
            "snapshot). A stage that reaches outside declares "
            "kind=StageKind.RESOLUTION, runs before every compilation stage "
            "in its phase, and contributes to the snapshot rather than to "
            "the assembly."
        )

StageProgram `dataclass` ¶

StageProgram(stages: tuple[CompiledStage, ...] = ())

G5 — the compiled, totally ordered stage list.

instructions `property` ¶

instructions: str | None

The system prompt the kernel-owned prompt stage carries (STG-7).

TeardownReport `dataclass` ¶

TeardownReport(run_id: str, reason: TeardownReason, elapsed_ms: float = 0.0, finalizers_run: int = 0, finalizer_failures: tuple[FinalizerFailure, ...] = (), tasks_awaited: int = 0, tasks_cancelled: int = 0, background_failures: tuple[BackgroundFailure, ...] = (), terminal_kind: str | None = None, terminal_delivered: bool = False, pressure: tuple[PressureRecord, ...] = ())

What teardown did, recorded once (RCX-11).

A run that dropped events or cancelled background work without saying so is indistinguishable from a healthy one. This value is the difference. It carries no exception objects and no live handles — only facts safe to log (ERR-5) — so it can cross a diagnostics or metrics boundary unchanged.

clean `property` ¶

clean: bool

True when nothing had to be forced and nothing owned by the run failed.

A terminal-delivery failure counts. BP-10 case 3 — an attached consumer that never received done — is the loudest thing a run can do wrong, and a report that called such a run clean because the task bookkeeping happened to balance would be exactly the indistinguishability RCX-11 exists to prevent. Case 2 (the consumer left) is not a failure and is not recorded as one by any adapter.

explain ¶

explain() -> str

Render the record as safe-to-log text (ERR-5).

Source code in src/symfonic/kernel/contracts/lifecycle.py

def explain(self) -> str:
    """Render the record as safe-to-log text (ERR-5)."""
    lines = [
        f"run {self.run_id} torn down: reason={self.reason} "
        f"elapsed_ms={self.elapsed_ms:.3f} finalizers={self.finalizers_run} "
        f"awaited={self.tasks_awaited} cancelled={self.tasks_cancelled} "
        f"terminal={self.terminal_kind} delivered={self.terminal_delivered}"
    ]
    lines.extend(
        f"finalizer failed: {failure.name} — {failure.error}"
        for failure in self.finalizer_failures
    )
    lines.extend(
        f"background failed: {failure.owner}:{failure.purpose} — {failure.error}"
        for failure in self.background_failures
    )
    lines.extend(record.render() for record in self.pressure)
    return "\n".join(lines)

ToolDescriptor `dataclass` ¶

ToolDescriptor(name: str, description: str = '', policy: Mapping[str, Any] = (lambda: MappingProxyType({}))())

One allowlisted tool, described rather than imported.

ToolManifest `dataclass` ¶

ToolManifest(tools: tuple[ToolDescriptor, ...] = ())

G4 — the single manifest source. Enforcement reads it; nothing rebuilds it.

ToolPort ¶

Bases: Protocol

Runs one allowlisted tool call and returns an opaque outcome record.

execute `async` ¶

execute(request: ToolRequest) -> Any

Execute request. A failing tool is reported, never hidden.

Source code in src/symfonic/kernel/contracts/ports.py

async def execute(self, request: ToolRequest) -> Any:
    """Execute ``request``. A failing tool is *reported*, never hidden."""

ToolRequest `dataclass` ¶

ToolRequest(call_id: str, name: str, arguments: Mapping[str, Any] = (lambda: MappingProxyType({}))())

One tool the model asked for, with a kernel-owned join key.

call_id is assigned by the kernel, not trusted from the provider: providers omit ids and reuse them across rounds, and an ambiguous join key between a tool_call and its tool_result is not a defect a consumer can work around (RES-3).

TurnRequest `dataclass` ¶

TurnRequest(prompt: str, attachments: tuple[Any, ...] = (), history: tuple[Any, ...] = (), run_id: str = '', root_run_id: str = '', parent_run_id: str | None = None, session_id: str = '', scope: Any = None, properties: Mapping[str, Any] = (lambda: EMPTY_PROPERTIES)())

The caller's half of an invocation: prompt, attachments, history.

Instructions are deliberately absent. They are a plan decision (the kernel-owned kernel.prompt stage carries them), so a caller cannot change the system prompt per call without recompiling — which is IPL-1's compile-once rule expressed in a signature.

UsageDelta `dataclass` ¶

UsageDelta(input_tokens: int = 0, output_tokens: int = 0, total_tokens: int = 0, cache_read_tokens: int = 0, cache_creation_tokens: int = 0, reasoning_tokens: int = 0, cache_ttl: str | None = None)

What one provider round reported (RES-4).

0 means unreported, never free: the kernel adds what it is told and estimates nothing, so a silent provider leaves the zero value standing.

deep_freeze ¶

deep_freeze(value: Any) -> Any

Return an immutable equivalent of value, transitively.

list/tuple become tuple, set/frozenset become frozenset, mappings become read-only proxies over a fresh dict, and everything else is returned unchanged.

Source code in src/symfonic/kernel/contracts/freeze.py

def deep_freeze(value: Any) -> Any:
    """Return an immutable equivalent of ``value``, transitively.

    ``list``/``tuple`` become ``tuple``, ``set``/``frozenset`` become
    ``frozenset``, mappings become read-only proxies over a fresh dict, and
    everything else is returned unchanged.
    """
    if isinstance(value, MappingProxyType):
        return MappingProxyType({key: deep_freeze(item) for key, item in value.items()})
    if isinstance(value, Mapping):
        return MappingProxyType({key: deep_freeze(item) for key, item in value.items()})
    if isinstance(value, list | tuple):
        return tuple(deep_freeze(item) for item in value)
    if isinstance(value, set | frozenset):
        return frozenset(deep_freeze(item) for item in value)
    if isinstance(value, bytearray):
        return bytes(value)
    return value

freeze_mapping ¶

freeze_mapping(value: Mapping[str, Any] | None) -> Mapping[str, Any]

Deep-freeze a mapping, treating None as empty.

Source code in src/symfonic/kernel/contracts/freeze.py

def freeze_mapping(value: Mapping[str, Any] | None) -> Mapping[str, Any]:
    """Deep-freeze a mapping, treating ``None`` as empty."""
    if not value:
        return MappingProxyType({})
    frozen = deep_freeze(dict(value))
    return frozen  # type: ignore[return-value]

is_deeply_frozen ¶

is_deeply_frozen(value: Any) -> bool

Return True when no mutable container is reachable from value.

"Frozen at the top, list underneath" is exactly what this rejects.

Source code in src/symfonic/kernel/contracts/freeze.py

def is_deeply_frozen(value: Any) -> bool:
    """Return ``True`` when no mutable container is reachable from ``value``.

    "Frozen at the top, ``list`` underneath" is exactly what this rejects.
    """
    if isinstance(value, MappingProxyType):
        return all(is_deeply_frozen(item) for item in value.values())
    if isinstance(value, _MUTABLE_CONTAINERS):
        return False
    if isinstance(value, str | bytes):
        return True
    if isinstance(value, tuple | frozenset):
        return all(is_deeply_frozen(item) for item in value)
    if dataclasses.is_dataclass(value) and not isinstance(value, type):
        params = getattr(type(value), "__dataclass_params__", None)
        if params is not None and not params.frozen:
            return False
        return all(
            is_deeply_frozen(getattr(value, field.name))
            for field in dataclasses.fields(value)
        )
    return True

symfonic.kernel.contracts¶

contracts ¶

AdapterPressure ¶

BackgroundEntry dataclass ¶

CheckpointerPort ¶

close async ¶

ensure_ready async ¶

flush async ¶

CompiledStage dataclass ¶

ConversationPort ¶

close_round ¶

messages ¶

open_turn ¶

DiagnosticRecord dataclass ¶

EventAdapter dataclass ¶

EventProgram dataclass ¶

FinalizerFailure dataclass ¶

InvocationOutcome dataclass ¶

InvocationPaused ¶

InvocationPlan dataclass ¶

allows_tool ¶

equality_key ¶

grants ¶

KernelEvent dataclass ¶

ModelDelta dataclass ¶

ModelPort ¶

invoke async ¶

stream ¶

ModelResolution dataclass ¶

ModelRound ¶

deltas ¶

result ¶

ModelTurn dataclass ¶

PendingInterrupt dataclass ¶

Phase ¶

PlanDiagnostics dataclass ¶

explain ¶

PlanIdentity dataclass ¶

PlanLimits dataclass ¶

PressureRecord dataclass ¶

PromptAssembly dataclass ¶

RequestScope dataclass ¶

narrows ¶

ResourcePort ¶

name property ¶

release async ¶

ResponseFormat dataclass ¶

ResponsePort ¶

structured property ¶

build_event ¶

build_result ¶

extract async ¶

ServiceBindings dataclass ¶

StageDescriptor dataclass ¶

validate ¶

StageProgram dataclass ¶

instructions property ¶

TeardownReport dataclass ¶

clean property ¶

explain ¶

ToolDescriptor dataclass ¶

ToolManifest dataclass ¶

ToolPort ¶

execute async ¶

ToolRequest dataclass ¶

TurnRequest dataclass ¶

UsageDelta dataclass ¶

deep_freeze ¶

freeze_mapping ¶

is_deeply_frozen ¶

`symfonic.kernel.contracts`¶

BackgroundEntry `dataclass` ¶

close `async` ¶

ensure_ready `async` ¶

flush `async` ¶

CompiledStage `dataclass` ¶

DiagnosticRecord `dataclass` ¶

EventAdapter `dataclass` ¶

EventProgram `dataclass` ¶

FinalizerFailure `dataclass` ¶

InvocationOutcome `dataclass` ¶

InvocationPlan `dataclass` ¶

KernelEvent `dataclass` ¶

ModelDelta `dataclass` ¶

invoke `async` ¶

ModelResolution `dataclass` ¶

ModelTurn `dataclass` ¶

PendingInterrupt `dataclass` ¶

PlanDiagnostics `dataclass` ¶

PlanIdentity `dataclass` ¶

PlanLimits `dataclass` ¶

PressureRecord `dataclass` ¶

PromptAssembly `dataclass` ¶

RequestScope `dataclass` ¶

name `property` ¶

release `async` ¶

ResponseFormat `dataclass` ¶

structured `property` ¶

extract `async` ¶

ServiceBindings `dataclass` ¶

StageDescriptor `dataclass` ¶

StageProgram `dataclass` ¶

instructions `property` ¶

TeardownReport `dataclass` ¶

clean `property` ¶

ToolDescriptor `dataclass` ¶

ToolManifest `dataclass` ¶

execute `async` ¶

ToolRequest `dataclass` ¶

TurnRequest `dataclass` ¶

UsageDelta `dataclass` ¶