symfonic.agent.cutover.messages

messages

Projecting kernel messages onto the legacy AgentResponse.messages shape.

AgentResponse.messages is a public payload: adopters index it, log it, and feed it back. The legacy engine fills it with LangChain message dumps ({"type": "ai", "content": ..., "tool_calls": [...]}) because the graph runtime hands it LangChain messages; the kernel hands the delegate typed :class:~symfonic.agent.facade_types.Message values whose dump is {"role": ..., "tool_call_id": ...}.

Left alone, that difference is a breaking change hiding behind a switch that is advertised as an implementation swap — message["type"] raises KeyError the day the switch flips. T3.5.2's parity suite measured it, and this module is the translation that closes it: the kernel's typed message is rebuilt as the LangChain message the legacy path would have produced, then dumped by the same serializer.

Translation, not interpretation. Nothing here invents a field the kernel did not report — and the price of that is named rather than claimed away. :class:~symfonic.agent.facade_types.Message carries role, content, tool_calls and tool_call_id and nothing else, so the rebuilt LangChain message's id, response_metadata and usage_metadata are LangChain's constructor defaults (None, {}, None). On the legacy path a real provider populates all three from the wire, and usage_metadata is how per-message token usage reaches adopters.

Those three are therefore the fields known not to survive the flip, pinned executably by tests/agent/cutover/test_facade_parity.py:: test_the_projection_does_not_carry_provider_message_metadata. Parity is claimed for the key set and for the values adopters index — type, content, tool_calls — not for provider-populated metadata.

ReplayedMessage

Bases: Message

A replayed history message that also carries its original block list.

:class:~symfonic.agent.facade_types.Message declares content as 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 no field to land in. Flattening it to its text blocks and stopping there is a silent loss at the wire, not a bookkeeping one: the legacy body replays the caller's BaseMessage verbatim into invocation_state["messages"], so on that route the image reaches the model and on the migrated route it would not. The model then answers about a picture it was never shown, plausibly, which is the failure mode this guard's parity evidence exists to exclude.

blocks keeps the original list beside the flattened text. :func:symfonic.agent.backend.messages.to_langchain rebuilds the wire message from blocks when they are present, so the provider is handed the same content list on both routes, while content stays the str every other reader of Message already expects.

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 legacy function. The slice is :func:symfonic.agent.engine._pair_aware_history_slice itself — imported, not reimplemented — 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. The envelope refuses that field, so on every admitted turn it holds its stock value and both routes trim at the same number by construction; threading it is what keeps that true the day the field is admitted, where a constant would silently disagree with the configuration and still produce a valid message list.

The named loss, and its exact boundary. Message carries role, content, tool_calls and tool_call_id and nothing else, so a replayed message's id, response_metadata and usage_metadata do not survive — the same three fields :func:as_legacy_messages cannot invent in the other direction. Those three do not reach the model: the wire message to_langchain rebuilds carries the role, the content and the tool calls, which is what the provider is handed. They are pinned executably 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.engine._log_history_trim is the same DEBUG record the legacy body emits before its slice — the function itself, not a second copy — 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 legacy function.** The slice
       is :func:`symfonic.agent.engine._pair_aware_history_slice` itself —
       imported, not reimplemented — 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. The envelope refuses that field, so
    on every admitted turn it holds its stock value and both routes trim at the
    same number by construction; threading it is what keeps that true the day
    the field is admitted, where a constant would silently disagree with the
    configuration and still produce a valid message list.

    **The named loss, and its exact boundary.** ``Message`` carries ``role``,
    ``content``, ``tool_calls`` and ``tool_call_id`` and nothing else, so a
    replayed message's ``id``, ``response_metadata`` and ``usage_metadata`` do
    not survive — the same three fields :func:`as_legacy_messages` cannot
    invent in the other direction. *Those three* do not reach the model: the
    wire message ``to_langchain`` rebuilds carries the role, the content and
    the tool calls, which is what the provider is handed. They are pinned
    executably 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.engine._log_history_trim` is the same
    DEBUG record the legacy body emits before its slice — the function itself,
    not a second copy — so a trim that drops the oldest turns is as
    diagnosable on this route as on that one.
    """
    if not history:
        return ()
    from symfonic.agent.engine 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.

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

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

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