The four ports the minimal kernel calls, and nothing else (CON-S).
Narrow on purpose. Each protocol is the smallest surface that lets the loop do
its job while staying ignorant of the wire: the kernel decides when a model
is called, which tool ids are legal, how many rounds are allowed and what
order events carry — the adapter decides what any of that looks like in a
provider's vocabulary.
They are Protocols rather than base classes so an adopter's implementation
owes nothing to our inheritance tree (CON-S, dependency inversion).
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."""
|
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."""
|
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
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."""
|
ResponsePort
Bases: Protocol
Response handling: structured extraction plus result/event adaptation.
structured
property
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(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."""
|
Bases: Protocol
Runs one allowlisted tool call and returns an opaque outcome record.
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."""
|