symfonic.agent.stream_contract

stream_contract

Task #25 option C — the 11.0 untyped-stream chunk contract.

.claude/docs/2026-08-20-task-25-stream-chunk-payload-decision.md Addendum 2 chose C: the target shape is option B's — StreamChunk.data on event_type="text_delta" is the answer delta as str, at the grain stream_typed already emits, with the kernel's kind set — and the schedule is the whole difference. C ships that shape at the first release that already rejects the old one, and leaves 9.x / 10.x yielding the LangGraph node mappings they have always yielded.

This module is that decision, implemented. Two things live here and nothing else:

• The release-line predicate. Which line publishes the contract, answered from the installed distribution version rather than from a private flag. The 9.x/10.x behaviour is not "old code left behind" — it is the published contract of those lines, so the branch is a first-class part of the design and is exercised by tests on both sides.
• The projection itself (:class:TextDeltaProjection), so the 11.0 body of SymfonicAgent._stream_impl stays a short delegation rather than a second copy of _stream_typed_impl inlined into an already-oversized module.

On the duplicated version marker. 11.0 is not decided here. The authority is symfonic.services.switching.release.FIRST_REJECTING_RETIREMENT_VERSION (itself the normative name from T1.2.6, and the value RET-8 published in retirement-record.toml). That module is a runtime-service internal and this one is facade-compiler, whose only licensed edge into runtime-service is a declared port (dependency-matrix.md); importing it here is the layering violation test_the_facade_reaches_runtime_services_only_through_a_port exists to catch. So the marker is mirrored, and the mirror is machine-checked: tests/agent/test_stream_contract.py fails if the two strings ever disagree. release.py's own warning — "three copies of 10.4 in three files is how a version boundary quietly becomes two boundaries" — is answered by the test, not by the comment.

TextDeltaProjection

TextDeltaProjection()

Project the runtime's typed events onto 11.0 untyped chunks.

Three properties are the acceptance criteria of the decision, and each is a consequence of how this projects rather than of a check bolted on afterwards:

• Concatenated deltas reconstruct final_response exactly. Every delta is appended to :attr:text as it is yielded, and the terminal done payload is built from that same accumulation (:attr:final_response) — so the two cannot drift. The alternative, deriving the terminal text separately from the runtime's last node update, is precisely how a filter change would leave the deltas and the final answer disagreeing with nothing to notice.
• The deltas are JSON-safe. They are str. The 9.x payload was a mapping carrying live AIMessage objects, which json.dumps refuses without a default.
• The kind set matches the kernel's. thinking is emitted only when the model actually produced thinking text, carrying that text — which is what cutover.projection.as_chunk does — instead of the unconditional data=None chunk the 9.x body yields before any I/O. On a turn with no thinking deltas both routes therefore emit {text_delta, done}, and on a reasoning turn both emit thinking with a str. Convergence, not suppression: blunt-dropping thinking would match the measured scenario and diverge again the first time a reasoning model was used.

spreading_activation has no such convergent form — the kernel's EventKind taxonomy excludes it by construction — so 11.0 drops it from the untyped stream. It remains on stream_typed as ActivationEvent, which is where the decision put it.

consolidating is deliberately not touched. Addendum 2 §1 leaves exactly that one kind unsigned, because admit_invocation refuses auto_consolidate=True and no turn in the parity harness can emit it. Dropping it here would sign a kind set that was never observed.

Source code in src/symfonic/agent/stream_contract.py

def __init__(self) -> None:
    self._extraction = ExtractionFilter()
    self._citation = CitationFilter()
    self._texts: list[str] = []
    self._thinking: list[str] = []
    #: The last graph output seen on an ``ExtensionEvent``. The fallback
    #: for a provider that returns a whole response instead of streaming
    #: it, mirroring ``_stream_typed_impl``.
    self.last_chain_result: dict[str, Any] = {}

extracted_ops property

extracted_ops: list[dict[str, Any]]

Memory-extraction ops the filter parsed out of the deltas.

text property

text: str

Everything yielded as a text_delta, concatenated.

thinking_text property

thinking_text: str

Everything yielded as a thinking chunk, concatenated.

chunks async

chunks(events: AsyncIterator[StreamEvent], *, run_id: str) -> AsyncIterator[StreamChunk]

Yield 11.0 chunks for one runtime event stream.

Does not close events; the caller owns that, because the caller is the one that has to schedule consolidation on GeneratorExit.

Source code in src/symfonic/agent/stream_contract.py

async def chunks(
    self, events: AsyncIterator[StreamEvent], *, run_id: str
) -> AsyncIterator[StreamChunk]:
    """Yield ``11.0`` chunks for one runtime event stream.

    Does not close ``events``; the caller owns that, because the caller is
    the one that has to schedule consolidation on ``GeneratorExit``.
    """
    async for event in events:
        if isinstance(event, TextDeltaEvent):
            clean = self._extraction.feed(event.text)
            if not clean:
                continue
            citation_clean = self._citation.feed(clean)
            if not citation_clean:
                continue
            self._texts.append(citation_clean)
            yield StreamChunk(
                event_type="text_delta", data=citation_clean, run_id=run_id
            )
        elif isinstance(event, ThinkingDeltaEvent):
            self._thinking.append(event.text)
            yield StreamChunk(
                event_type="thinking", data=event.text, run_id=run_id
            )
        elif isinstance(event, ExtensionEvent) and isinstance(event.payload, dict):
            output = event.payload.get("output")
            if isinstance(output, dict):
                self.last_chain_result = output

final_response

final_response(*, fallback: str = '') -> str

The terminal final_response.

Identical to :attr:text whenever anything streamed, which is what makes reconstruction exact. fallback covers a provider that emitted no TextDeltaEvent at all: on such a turn there is nothing to reconstruct from, and answering with the graph's own final text is strictly better than answering with "". That turn is the one shape where concatenation does not reproduce the terminal payload, and it is the same shape on which stream_typed and stream_text already yield nothing — this contract does not make it worse, and inventing a synthetic whole-answer delta to paper over it would break the grain half of the target.

The fallback is passed in rather than derived here so that this module stays free of an import back into engine, whose _find_final_response is the function that knows how to walk a LangGraph state dict.

Source code in src/symfonic/agent/stream_contract.py

def final_response(self, *, fallback: str = "") -> str:
    """The terminal ``final_response``.

    Identical to :attr:`text` whenever anything streamed, which is what
    makes reconstruction exact. ``fallback`` covers a provider that emitted
    no ``TextDeltaEvent`` at all: on such a turn there is nothing to
    reconstruct *from*, and answering with the graph's own final text is
    strictly better than answering with ``""``. That turn is the one shape
    where concatenation does not reproduce the terminal payload, and it is
    the same shape on which ``stream_typed`` and ``stream_text`` already
    yield nothing — this contract does not make it worse, and inventing a
    synthetic whole-answer delta to paper over it would break the grain
    half of the target.

    The fallback is passed in rather than derived here so that this module
    stays free of an import back into ``engine``, whose
    ``_find_final_response`` is the function that knows how to walk a
    LangGraph state dict.
    """
    return self.text or fallback

flush

flush(*, run_id: str) -> list[StreamChunk]

Drain both filters into any deltas they were still withholding.

Order matters and is the same as _stream_typed_impl's: the extraction filter may release real prose, that prose still has to pass the citation filter, and only then does the citation filter drain. A flush that skipped the middle step would emit an unscrubbed citation tag on exactly the turns where a tag landed on a chunk boundary.

Source code in src/symfonic/agent/stream_contract.py

def flush(self, *, run_id: str) -> list[StreamChunk]:
    """Drain both filters into any deltas they were still withholding.

    Order matters and is the same as ``_stream_typed_impl``'s: the
    extraction filter may release real prose, that prose still has to pass
    the citation filter, and only then does the citation filter drain. A
    flush that skipped the middle step would emit an unscrubbed citation
    tag on exactly the turns where a tag landed on a chunk boundary.
    """
    drained: list[StreamChunk] = []
    remainder = self._extraction.flush()
    if remainder:
        remainder = self._citation.feed(remainder)
        if remainder:
            self._texts.append(remainder)
            drained.append(
                StreamChunk(event_type="text_delta", data=remainder, run_id=run_id)
            )
    tail = self._citation.flush()
    if tail:
        self._texts.append(tail)
        drained.append(
            StreamChunk(event_type="text_delta", data=tail, run_id=run_id)
        )
    return drained

installed_release_line

installed_release_line(distribution: str = DISTRIBUTION_NAME) -> str

The installed distribution version, or a conservative fallback.

Source code in src/symfonic/agent/stream_contract.py

def installed_release_line(distribution: str = DISTRIBUTION_NAME) -> str:
    """The installed distribution version, or a conservative fallback."""
    try:
        from importlib.metadata import version  # noqa: PLC0415

        return version(distribution)
    except Exception:  # noqa: BLE001 - any metadata failure means "unknown"
        return _UNKNOWN_VERSION

numeric_release_line

numeric_release_line(version: str) -> str

The leading numeric release segment of version, dotted, or "0".

Tolerant on purpose: a local build may carry 11.0.0.dev3+local or 11.0rc1, and refusing to parse those would silently drop such a build onto the pre-11.0 branch — the same wrong answer as the fallback, arrived at without anyone noticing. Only the digits before the first non-numeric component decide the line.

Public since TA8.54, which needed the string rather than the tuple: services.switching.release.parse_version is strict (int(part), so 9.13.0.dev0 raises ConfigurationError), and every agent build now hands it a version. Normalising through this one function is what keeps "which line is this build on?" a single answer — a second tolerant parser would be free to disagree with this one about a dev build, and the disagreement would only show up on a release candidate.

Source code in src/symfonic/agent/stream_contract.py

def numeric_release_line(version: str) -> str:
    """The leading numeric release segment of ``version``, dotted, or ``"0"``.

    Tolerant on purpose: a local build may carry ``11.0.0.dev3+local`` or
    ``11.0rc1``, and refusing to parse those would silently drop such a build
    onto the pre-11.0 branch — the same wrong answer as the fallback, arrived
    at without anyone noticing. Only the digits before the first non-numeric
    component decide the line.

    Public since TA8.54, which needed the *string* rather than the tuple:
    ``services.switching.release.parse_version`` is strict (``int(part)``, so
    ``9.13.0.dev0`` raises ``ConfigurationError``), and every agent build now
    hands it a version. Normalising through this one function is what keeps
    "which line is this build on?" a single answer — a second tolerant parser
    would be free to disagree with this one about a dev build, and the
    disagreement would only show up on a release candidate.
    """
    parts: list[str] = []
    for raw in version.split("."):
        digits = ""
        for character in raw:
            if not character.isdigit():
                break
            digits += character
        if not digits:
            break
        parts.append(digits)
        if digits != raw:
            break
    return ".".join(parts) or "0"

publishes_text_deltas

publishes_text_deltas(package_version: str) -> bool

Does package_version ship the text-delta chunk contract?

Pure, and separately testable from whatever the checkout happens to be installed as. >= rather than == so 11.1 and 12.0 keep the contract they inherited.

Source code in src/symfonic/agent/stream_contract.py

def publishes_text_deltas(package_version: str) -> bool:
    """Does ``package_version`` ship the text-delta chunk contract?

    Pure, and separately testable from whatever the checkout happens to be
    installed as. ``>=`` rather than ``==`` so ``11.1`` and ``12.0`` keep the
    contract they inherited.
    """
    return _parse(package_version) >= _parse(TEXT_DELTA_CONTRACT_VERSION)

text_delta_contract_is_published

text_delta_contract_is_published() -> bool

Does this installation publish it? The seam _stream_impl reads.

Source code in src/symfonic/agent/stream_contract.py

def text_delta_contract_is_published() -> bool:
    """Does *this* installation publish it? The seam ``_stream_impl`` reads."""
    return publishes_text_deltas(installed_release_line())