symfonic.agent.cutover.typed_close_state

typed_close_state

The accumulation a typed turn must have ready at every exit path.

Split out of :mod:~symfonic.agent.cutover.typed_projection because the two are different responsibilities: that module is a stateless translation from kernel events to typed events, and this one is the per-turn state a body carries so the answer it persists does not depend on how the turn ended.

TypedCloseState

TypedCloseState()

The accumulation a typed turn must have ready at every exit path.

ST1's duplicated_filtering_and_accumulation and generator_exit_consolidation_state drift rows are one defect seen twice: the legacy typed body filters and accumulates in the main loop and then filters and accumulates again, differently, inside its GeneratorExit handler. The two copies decide independently whether a buffered tag remainder counts, so an abandoned turn and a completed turn could write different text to memory for the same deltas.

One accumulator, shared by both the legacy body and the kernel route, is the fix. :meth:flush is idempotent, so "the consumer left" and "the stream ended" reach the same final text by construction rather than by two code paths agreeing.

Source code in src/symfonic/agent/cutover/typed_close_state.py

def __init__(self) -> None:
    self._extraction = ExtractionFilter()
    self._citation = CitationFilter()
    self._clean: list[str] = []
    self._thinking: list[str] = []
    self._flushed = False
    #: The last raw result mapping seen on an extension payload. The
    #: fallback for providers that never emit a text delta.
    self.last_chain_result: dict[str, Any] = {}

clean_text property

clean_text: str

Everything the consumer was shown, concatenated, in order.

feed_text

feed_text(text: str) -> str | None

Filter one text delta; return what may be emitted, or None.

None means a filter is buffering a potential tag prefix. It is not "nothing arrived" -- the characters are held and will surface either in a later delta or in :meth:flush.

Source code in src/symfonic/agent/cutover/typed_close_state.py

def feed_text(self, text: str) -> str | None:
    """Filter one text delta; return what may be emitted, or ``None``.

    ``None`` means a filter is buffering a potential tag prefix. It is not
    "nothing arrived" -- the characters are held and will surface either in
    a later delta or in :meth:`flush`.
    """
    clean = self._extraction.feed(text)
    if not clean:
        return None
    emitted = self._citation.feed(clean)
    if not emitted:
        return None
    self._clean.append(emitted)
    return emitted

flush

flush() -> tuple[str, ...]

Drain both filters once; return the tail deltas still owed.

Idempotent: a second call returns () and changes nothing. That is what lets the abandoned path and the completed path call it without either double-counting the tail or disagreeing about it.

Source code in src/symfonic/agent/cutover/typed_close_state.py

def flush(self) -> tuple[str, ...]:
    """Drain both filters once; return the tail deltas still owed.

    Idempotent: a second call returns ``()`` and changes nothing. That is
    what lets the abandoned path and the completed path call it without
    either double-counting the tail or disagreeing about it.
    """
    if self._flushed:
        return ()
    self._flushed = True
    tail: list[str] = []
    remainder = self._extraction.flush()
    if remainder:
        remainder = self._citation.feed(remainder)
        if remainder:
            tail.append(remainder)
    citation_tail = self._citation.flush()
    if citation_tail:
        tail.append(citation_tail)
    self._clean.extend(tail)
    return tuple(tail)

memory_response

memory_response(fallback: str = '') -> str

The response consolidation should persist for this turn.

The same expression on both exit paths, which is the point: the text written to memory must not depend on whether the consumer stayed.

Source code in src/symfonic/agent/cutover/typed_close_state.py

def memory_response(self, fallback: str = "") -> str:
    """The response consolidation should persist for this turn.

    The same expression on both exit paths, which is the point: the text
    written to memory must not depend on whether the consumer stayed.
    """
    return self.clean_text or fallback

observe_result

observe_result(payload: Any) -> None

Record a chain result carried on an extension payload, if any.

Source code in src/symfonic/agent/cutover/typed_close_state.py

def observe_result(self, payload: Any) -> None:
    """Record a chain result carried on an extension payload, if any."""
    if isinstance(payload, dict):
        output = payload.get("output")
        if isinstance(output, dict):
            self.last_chain_result = output