symfonic.agent.cutover.messages

messages

Kernel messages in and out of the legacy AgentResponse.messages shape.

The outbound dump, :func:as_legacy_messages, belongs to :mod:symfonic.agent.response_messages (CORE-414): the native compatibility HTTP routes call it without loading cutover, and this module re-exports the same function for the delegate. Its module docstring carries the translation contract and the two named losses (no message id, no response_metadata), pinned by tests/agent/cutover/ test_facade_parity.py::test_the_projection_does_not_carry_provider_message_metadata.

What stays here is the inbound direction the delegate alone needs: trimming and converting caller history into typed kernel messages (:func:as_kernel_history), the :class:ReplayedMessage that keeps what Message has no field for, and the history cap.

ReplayedMessage

Bases: Message

A replayed history message that keeps what Message has no field for.

Message.content is a str (RES-2), so a history message whose LangChain content is a list of typed blocks -- an image, a document, a provider reasoning block -- has nowhere to land. Flattening it to its text blocks and stopping there is a silent loss at the wire: legacy replays the caller's message verbatim, so dropping blocks hides the image from the model. blocks keeps the list beside the flattened text, and :func:symfonic.agent.backend.messages.to_langchain rebuilds the wire message from it, while content stays the str readers expect.

tool_name is the same argument for a replayed ToolMessage.name: legacy sends and reports the caller's value verbatim, None included. Every replayed tool message is a ReplayedMessage, so its name is never re-derived from an earlier call (CORE-20); only a tool message the kernel produced itself is named that way, by :mod:.tool_names.

as_kernel_history

as_kernel_history(history: Any, *, cap: int) -> tuple[Any, ...]

Trim and convert caller history into the kernel's typed messages.

The inbound direction, and it is deliberately not the mirror image of :func:as_legacy_messages. Three things happen, in the order _legacy_run_impl does them:

1. Trim, on the LangChain messages, with the shared function. The slice is :func:symfonic.agent.history_trim.pair_aware_history_slice, which the engine's compatibility wrapper delegates to — so the two routes cannot disagree about which messages survive. A second copy of Anthropic's tool-pairing rule would fail as an HTTP 400 on the next provider call rather than as a visible parity difference, which is the worst available failure mode.
2. Convert each survivor into the typed :class:~symfonic.agent.facade_types.Message the kernel carries.
3. Nothing else. No reordering and no system-message filtering: the replayed history reaches build_request exactly as legacy replays it into invocation_state["messages"].

cap is config.agent.max_conversation_messages, threaded from the engine rather than written down here. CORE-177 admits plain integer caps; fresh history still uses the shared slice. Authenticated checkpoint history is converted without trimming: its initial turn already applied the cap.

The named loss, and its exact boundary. Message has no message id and no response_metadata, so a replayed message's two do not survive — the same two :func:as_legacy_messages cannot invent in the other direction. A replayed AI message's usage_metadata does survive, on Message.usage (CORE-21), so it is reported back as legacy reports it. None of the three reaches the model: to_langchain rebuilds the role, content and tool calls, which is what the provider is handed. Pinned in tests/agent/cutover/test_history_parity.py.

Block content is a different case and must not be filed under the same claim. Message.content is a str, so a block list has no field to land in — but it does reach the model on the legacy route, which replays the caller's BaseMessage verbatim. Flattening it here would drop a replayed image at the wire while the A/B taken at the wire showed it. So the blocks are carried on :class:ReplayedMessage instead of dropped, and content keeps the flattened text for readers that want a str.

Telemetry. :func:symfonic.agent.history_trim.log_history_trim is the shared implementation behind the legacy body's DEBUG record, so a trim that drops the oldest turns is as diagnosable on this route as on that one.

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

def as_kernel_history(history: Any, *, cap: int) -> tuple[Any, ...]:
    """Trim and convert caller ``history`` into the kernel's typed messages.

    The inbound direction, and it is deliberately *not* the mirror image of
    :func:`as_legacy_messages`. Three things happen, in the order
    ``_legacy_run_impl`` does them:

    1. **Trim, on the LangChain messages, with the shared function.** The slice
       is :func:`symfonic.agent.history_trim.pair_aware_history_slice`, which
       the engine's compatibility wrapper delegates to — so the two routes
       cannot disagree about
       which messages survive. A second copy of Anthropic's tool-pairing rule
       would fail as an HTTP 400 on the next provider call rather than as a
       visible parity difference, which is the worst available failure mode.
    2. **Convert each survivor** into the typed
       :class:`~symfonic.agent.facade_types.Message` the kernel carries.
    3. **Nothing else.** No reordering and no system-message filtering: the
       replayed history reaches ``build_request`` exactly as legacy replays it
       into ``invocation_state["messages"]``.

    ``cap`` is ``config.agent.max_conversation_messages``, threaded from the
    engine rather than written down here. CORE-177 admits plain integer caps;
    fresh history still uses the shared slice. Authenticated checkpoint history
    is converted without trimming: its initial turn already applied the cap.

    **The named loss, and its exact boundary.** ``Message`` has no message
    ``id`` and no ``response_metadata``, so a replayed message's two do not
    survive — the same two :func:`as_legacy_messages` cannot invent in the
    other direction. A replayed AI message's ``usage_metadata`` *does* survive,
    on ``Message.usage`` (CORE-21), so it is reported back as legacy reports it.
    None of the three reaches the model: ``to_langchain`` rebuilds the role,
    content and tool calls, which is what the provider is handed. Pinned in
    ``tests/agent/cutover/test_history_parity.py``.

    Block content is a different case and must not be filed under the same
    claim. ``Message.content`` is a ``str``, so a block list has no field to
    land in — but it *does* reach the model on the legacy route, which replays
    the caller's ``BaseMessage`` verbatim. Flattening it here would drop a
    replayed image at the wire while the A/B taken at the wire showed it. So
    the blocks are carried on :class:`ReplayedMessage` instead of dropped, and
    ``content`` keeps the flattened text for readers that want a ``str``.

    **Telemetry.** :func:`symfonic.agent.history_trim.log_history_trim` is the
    shared implementation behind the legacy body's DEBUG record, so a trim
    that drops the oldest turns is as
    diagnosable on this route as on that one.
    """
    if isinstance(history, _CheckpointHistory):
        return tuple(_as_typed(message) for message in history)
    if not history:
        return ()
    from symfonic.agent.history_trim import (
        log_history_trim,
        pair_aware_history_slice,
    )

    messages = list(history)
    log_history_trim(messages, cap)
    retained = pair_aware_history_slice(messages, cap)
    return tuple(_as_typed(message) for message in retained)

as_legacy_messages

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

Dump kernel messages in the shape the legacy engine returns, tools named.

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

def as_legacy_messages(messages: Iterable[Any]) -> list[dict[str, Any]]:
    """Dump kernel messages in the shape the legacy engine returns, tools named."""
    typed = list(messages)
    return named_tool_messages(typed, [_as_langchain(message).model_dump() for message in typed])

history_cap

history_cap(stated: Any) -> int

The cap :func:as_kernel_history trims at, given what was stated.

Beside the trim it feeds rather than on the delegate that passes it: the delegate is translation, and "what does an unstated cap mean?" is a question about this module's own function. bool is excluded explicitly because it is an int and max_conversation_messages=True would otherwise trim every replayed conversation to one message.

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

def history_cap(stated: Any) -> int:
    """The cap :func:`as_kernel_history` trims at, given what was stated.

    Beside the trim it feeds rather than on the delegate that passes it: the
    delegate is translation, and "what does an unstated cap mean?" is a question
    about this module's own function. ``bool`` is excluded explicitly because it
    is an ``int`` and ``max_conversation_messages=True`` would otherwise trim
    every replayed conversation to one message.
    """
    if not isinstance(stated, int) or isinstance(stated, bool):
        return _STOCK_HISTORY_CAP
    return stated