Skip to content

symfonic.platform.transport_conversation

transport_conversation

Request-bound conversation seam for native HTTP turns.

The transport never treats a caller's session id or browser-supplied history as proof of ownership. Deployments that persist chats provide this port and authorize a session against the already verified principal before a model call.

ConversationPort

Bases: Protocol

prepare async 

prepare(principal: Any, session_id: str, query: str) -> PreparedConversation

Refuse an unowned session before returning its trusted transcript.

Source code in src/symfonic/platform/transport_conversation.py 
26
27
28
29
async def prepare(
    self, principal: Any, session_id: str, query: str,
) -> PreparedConversation:
    """Refuse an unowned session before returning its trusted transcript."""

PreparedConversation dataclass 

PreparedConversation(history: tuple[Any, ...], publish: Callable[[str], Awaitable[None]])

Authorized history and final-only publication for one HTTP request.

UnusableFinal

Bases: ValueError

A turn completed without a publishable answer.

answer_of_result 

answer_of_result(result: Any, *, synthesis: bool = False) -> str

Read only explicit text fields; never stringify an AgentResult.

A tool_limit result is refused unless terminal synthesis closed it, and then only that reply is published (:func:_synthesized_reply). When the caller's agent has synthesis enabled and synthesis declined because the last round spoke, that round's text is published (:func:_interim_reply). The empty-answer check below still applies.

Source code in src/symfonic/platform/transport_conversation.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def answer_of_result(result: Any, *, synthesis: bool = False) -> str:
    """Read only explicit text fields; never stringify an AgentResult.

    A ``tool_limit`` result is refused unless terminal synthesis closed it, and
    then only that reply is published (:func:`_synthesized_reply`). When the
    caller's agent has synthesis enabled and synthesis declined because the
    last round spoke, that round's text is published (:func:`_interim_reply`).
    The empty-answer check below still applies.
    """
    reason = getattr(result, "stop_reason", None)
    if getattr(reason, "value", reason) == "tool_limit":
        answer = _synthesized_reply(result)
        if answer is None and synthesis:
            answer = _interim_reply(result)
        if answer is None:
            raise UnusableFinal("tool limit")
    elif isinstance(result, str):
        answer = result
    else:
        answer = next(
            (value for name in ("text", "final_response", "content")
             if isinstance((value := getattr(result, name, None)), str)),
            None,
        )
    if not answer or not answer.strip():
        raise UnusableFinal("empty answer")
    return answer

answer_of_run 

answer_of_run(agent: Any, result: Any) -> str

:func:answer_of_result for a run of agent, aware of its synthesis.

Source code in src/symfonic/platform/transport_conversation.py 
131
132
133
def answer_of_run(agent: Any, result: Any) -> str:
    """:func:`answer_of_result` for a ``run`` of ``agent``, aware of its synthesis."""
    return answer_of_result(result, synthesis=terminal_synthesis_enabled(agent))

final_answer_of 

final_answer_of(event: Any, deltas: str) -> str

Prefer the authoritative completed result to streaming deltas.

Source code in src/symfonic/platform/transport_conversation.py 
136
137
138
139
140
141
142
143
144
145
def final_answer_of(event: Any, deltas: str) -> str:
    """Prefer the authoritative completed result to streaming deltas."""
    reason = getattr(event, "stop_reason", None)
    if getattr(reason, "value", reason) == "tool_limit":
        raise UnusableFinal("tool limit")
    result = getattr(event, "result", None)
    if result is not None:
        return answer_of_result(result)
    text = getattr(event, "text", None)
    return answer_of_result(text if isinstance(text, str) and text else deltas)

terminal_synthesis_enabled 

terminal_synthesis_enabled(agent: Any) -> bool

Whether the agent's plan binds the TerminalSynthesisCapability it composed.

The folded name alone is not proof: any capability may contribute under terminal_synthesis. Nor is a composed instance: only the one the plan factory bound is what the turn runs.

Source code in src/symfonic/platform/transport_conversation.py 
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def terminal_synthesis_enabled(agent: Any) -> bool:
    """Whether the agent's plan binds the ``TerminalSynthesisCapability`` it composed.

    The folded name alone is not proof: any capability may contribute under
    ``terminal_synthesis``. Nor is a composed instance: only the one the plan
    factory bound is what the turn runs.
    """
    names = getattr(agent, "capabilities", None)
    composed = getattr(agent, "_composed_capabilities", None)
    if not isinstance(names, tuple) or not isinstance(composed, tuple):
        return False
    policy = terminal_synthesis_for(composed)
    bound = getattr(getattr(agent, "_plans", None), "terminal_synthesis", None)
    return "terminal_synthesis" in names and policy is not None and bound is policy