Skip to content

symfonic.agent.fastapi.shared_turn

shared_turn

One selected-root streamed turn over the shared conversation authority.

Both selected streaming routes (/stream and /stream/typed) open their turn here, so ownership, attachment family, capability and budget refusals, session issuance, publication and empty-session discard are one code path. Each route only chooses how the admitted raw events are framed.

open_shared_turn refuses before the model is called. SharedTurn.events yields the admitted kernel events, publishes a usable final answer when the done event arrives (before the route can frame its terminal), drops the error event (the run's exception follows it) and discards an issued session nothing was published to. TurnResponse discards on every response exit, including a response that was never iterated.

SharedTurn 

SharedTurn(conversation: Any, principal: Any, invocation: Any, scope: Any = None)

One streamed turn: issue before headers, publish before the terminal.

Source code in src/symfonic/agent/fastapi/shared_turn.py 
112
113
114
115
116
117
118
119
def __init__(self, conversation: Any, principal: Any, invocation: Any, scope: Any = None):
    self._conversation = conversation
    self._principal = principal
    self._invocation = invocation
    self.scope = scope
    self._prepared: Any = None
    self.issued: str | None = None
    self._settled = False

revision property 

revision: int | None

The prepared revision, when the authority checks publication by one.

release async 

release() -> None

Discard an issued session no turn was published to; idempotent.

Source code in src/symfonic/agent/fastapi/shared_turn.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
async def release(self) -> None:
    """Discard an issued session no turn was published to; idempotent."""
    if self.issued is None or self._settled:
        return
    self._settled = True
    # A client disconnect cancels this task; only the cleanup is
    # shielded, so the empty session is still removed.
    with anyio.CancelScope(shield=True):
        try:
            await self._conversation.discard_empty(self._principal, self.issued)
        except Exception:
            logging.getLogger(__name__).warning(
                "could not discard empty conversation after unsuccessful first turn",
                exc_info=True,
            )

settle 

settle() -> None

Keep an issued session that a recorded pause now refers to.

Source code in src/symfonic/agent/fastapi/shared_turn.py 
138
139
140
def settle(self) -> None:
    """Keep an issued session that a recorded pause now refers to."""
    self._settled = True

TurnResponse 

TurnResponse(turn: SharedTurn, content: Any, **kwargs: Any)

Bases: EventSourceResponse

Release the turn when the response ends, iterated or not.

Source code in src/symfonic/agent/fastapi/shared_turn.py 
 98
 99
100
def __init__(self, turn: SharedTurn, content: Any, **kwargs: Any) -> None:
    super().__init__(content, **kwargs)
    self._turn = turn

open_shared_turn async 

open_shared_turn(host: Any, conversation: Any, request: Any, raw_request: Any, scope: Any, *, admit: Any = admit_native_stream) -> tuple[SharedTurn, Any, Any]

Admit one turn before the model; every refusal is mapped for streams.

Source code in src/symfonic/agent/fastapi/shared_turn.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def open_shared_turn(
    host: Any,
    conversation: Any,
    request: Any,
    raw_request: Any,
    scope: Any,
    *,
    admit: Any = admit_native_stream,
) -> tuple[SharedTurn, Any, Any]:
    """Admit one turn before the model; every refusal is mapped for streams."""
    await check_budget_for_tenant(raw_request, str(scope.tenant_id))
    invocation = _CODEC.decode_invocation(
        request,
        scope=scope,
        is_admin=request_is_admin(raw_request),
        principal_id=request_user_id(raw_request),
    )
    try:
        principal = principal_of(raw_request, scope)
        if invocation.attachments:
            from symfonic.platform.conversation_authority import ConversationFamilyRefused

            raise ConversationFamilyRefused(
                "shared local conversation history admits text-only chat"
            )
        prepared = None
        if invocation.session_id:
            prepared = await conversation.prepare(
                principal, invocation.session_id, invocation.query
            )
        agent = await host.agent_for(scope)
        admission = admit(agent, str(scope.tenant_id), invocation.query)
        turn = SharedTurn(conversation, principal, invocation, scope)
        await turn.open(prepared)
    except Exception as exc:
        from symfonic.platform.host import HostClosed

        if isinstance(exc, HostClosed):
            exc = SymfonicAgentError(
                "the host is shutting down and is composing no further agents",
                code="native_stream_unavailable",
            )
        raise raise_mapped(exc, STREAM_POLICY) from exc
    return turn, agent, admission