symfonic.platform.transport_citation

transport_citation

Citation presentation at the native HTTP boundary.

BufferedCitationStream

BufferedCitationStream()

Project each display channel through its canonical filter in event order.

Text and thinking have independent filters. The filter's held characters keep their source event pending; every later event waits behind it. Once those characters are decided, their event and any intervening tool events are released in original order. A bounded queue refuses instead of allowing unresolved text to move across an unlimited number of events.

Source code in src/symfonic/platform/transport_citation.py

def __init__(self) -> None:
    self._filters: dict[str, CitationTagFilter] = {}
    self._chars: dict[str, list[tuple[str, _PendingEvent]]] = {}
    self._events: list[_PendingEvent] = []
    self._bytes = 0

terminal

terminal(source: Any) -> list[tuple[Any, str | None]]

Settle held display text before an authoritative terminal payload.

Source code in src/symfonic/platform/transport_citation.py

def terminal(self, source: Any) -> list[tuple[Any, str | None]]:
    """Settle held display text before an authoritative terminal payload."""
    return [*self.finish(), (source, None)]

CitationBufferOverflow

Bases: UnusableFinal

The citation boundary could not be resolved within a bounded queue.

CitationTagFilter

CitationTagFilter(layer_names: Sequence[str] = DEFAULT_LAYER_NAMES)

Streaming filter that strips internal layer tags across chunk boundaries.

Feed chunks in order and concatenate what comes back; call :meth:flush once at stream end. A partial prefix that never completes is dropped at flush rather than emitted: it was only held back because it matched a known layer name, so emitting it would leak a fragment of an internal annotation.

Source code in src/symfonic/capabilities/knowledge/citations.py

def __init__(self, layer_names: Sequence[str] = DEFAULT_LAYER_NAMES) -> None:
    if not layer_names:
        raise ValueError(
            "a citation filter needs at least one layer name; an empty vocabulary "
            "is a filter that strips nothing, which is better written as no filter."
        )
    self._tags = tuple(f"[{name.lower()}]" for name in layer_names)
    # T3.2.4 finding 3: horizontal whitespace only, never ``\s*``.
    #
    # ``\s*`` is greedy over newlines too, so a tag on its own indented line
    # took the preceding line break with it and welded two lines together:
    # ``"code:\n    [working]    indented"`` became ``"code:    indented"``.
    # That was the most destructive of the three answers one input could
    # produce, and it is a *whole-buffer* defect — under chunked delivery
    # the newline had already been emitted and survived, which is why it is
    # a different finding from the chunk-invariance one and not the same one
    # seen twice.
    self._tag_re = re.compile(
        r"[ \t]*\[(?:" + "|".join(re.escape(name) for name in layer_names) + r")\]",
        re.IGNORECASE,
    )
    self._buffer = ""
    self._last_emitted_was_space = False
    self._pending_seam = False

feed

feed(text: str) -> str

Append text and return the prefix that is safe to emit.

Source code in src/symfonic/capabilities/knowledge/citations.py

def feed(self, text: str) -> str:
    """Append ``text`` and return the prefix that is safe to emit."""
    if not text:
        return ""
    self._buffer += text
    return self._drain(final=False)

flush

flush() -> str

Drain the buffer at stream end, dropping any unfinished tag prefix.

Source code in src/symfonic/capabilities/knowledge/citations.py

def flush(self) -> str:
    """Drain the buffer at stream end, dropping any unfinished tag prefix."""
    return self._drain(final=True)

visible_answer

visible_answer(text: str) -> str

Require a usable answer after removing internal citation tags.

Source code in src/symfonic/platform/transport_citation.py

def visible_answer(text: str) -> str:
    """Require a usable answer after removing internal citation tags."""
    answer = visible_text(text)
    if not answer.strip():
        raise UnusableFinal("empty visible answer")
    return answer

visible_assistant_messages

visible_assistant_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]

Copy assistant display text without changing tool or caller payloads.

Source code in src/symfonic/platform/transport_citation.py

def visible_assistant_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Copy assistant display text without changing tool or caller payloads."""
    return [
        {**message, "content": visible_text(message["content"])}
        if message.get("type") == "ai" and isinstance(message.get("content"), str)
        else message
        for message in messages
    ]

visible_text

visible_text(text: str) -> str

Remove only the closed vocabulary of internal layer tags.

Source code in src/symfonic/platform/transport_citation.py

def visible_text(text: str) -> str:
    """Remove only the closed vocabulary of internal layer tags."""
    citation_filter = CitationTagFilter()
    return citation_filter.feed(text) + citation_filter.flush()