Skip to content

symfonic.agent.fastapi.native_stream_guard

native_stream_guard

Shared admission and observed-usage guard for native SSE projections.

RawUsageGuard 

RawUsageGuard(metered: bool)

Check each model round before a projected success terminal is yielded.

Source code in src/symfonic/agent/fastapi/native_stream_guard.py 
109
110
111
def __init__(self, metered: bool) -> None:
    self._metered = metered
    self._measured_rounds = 0

admit_native_stream 

admit_native_stream(agent: Any, tenant_id: str, query: str, *, human: Any = None) -> NativeStreamAdmission

Admit a stateless stream and pin a priced plan for budgeted turns.

human admits exactly that composed human-interaction capability as well; only the selected typed stream with guarded durable resume passes it.

Source code in src/symfonic/agent/fastapi/native_stream_guard.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def admit_native_stream(
    agent: Any, tenant_id: str, query: str, *, human: Any = None
) -> NativeStreamAdmission:
    """Admit a stateless stream and pin a priced plan for budgeted turns.

    ``human`` admits exactly that composed human-interaction capability as
    well; only the selected typed stream with guarded durable resume passes it.
    """
    if not isinstance(agent, Agent) or not _admitted_capabilities(agent, human):
        raise ConfigurationError(
            "native compatibility stream requires a stateless Agent "
            "without behavior capabilities"
        )
    metered = get_budget_tracker() is not None
    if "telemetry" in agent.capabilities or metered:
        from symfonic.platform.telemetry import budget_metering_is_bound

        # Without telemetry nothing bills the tenant, so a budgeted human-only
        # turn refuses here as any unmetered composition does.
        paired = human if "human" in agent.capabilities else None
        if not budget_metering_is_bound(agent, tenant_id, human=paired):
            raise SymfonicAgentError(
                "native stream cannot attest tenant-bound budget usage",
                code="native_stream_unavailable",
            )
    if metered and not agent.stream_usage_declared:
        raise SymfonicAgentError(
            "native stream cannot attest provider streaming usage",
            code="native_stream_unavailable",
        )
    prepared_plan = None
    if metered:
        from symfonic.core.observability.pricing import calculate_cost_with_meta

        resolved_config = agent._plans._resolve_config()
        model_name = resolved_config.model_name
        if not model_name or calculate_cost_with_meta(model_name, 0, 0)[1]:
            raise SymfonicAgentError(
                f"native stream has no price contract for model {model_name!r}; "
                "configure pricing_overrides before serving budgeted streams",
                code="native_stream_unavailable",
            )
        prepared_plan = agent._prepare(query, None, resolved_config=resolved_config)
        if prepared_plan.model.model_name != model_name:
            raise SymfonicAgentError(
                "native stream compiled a model different from its price contract",
                code="native_stream_unavailable",
            )
    return NativeStreamAdmission(metered, prepared_plan)

native_stream_source 

native_stream_source(agent: Agent, invocation: Any, session_id: str, prepared_plan: Any, history: tuple[Any, ...] = ()) -> Any

Use one admitted plan for execution; keep the regular unbudgeted path.

history must come from an authority that verified the session owner, never from the request body.

Source code in src/symfonic/agent/fastapi/native_stream_guard.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def native_stream_source(
    agent: Agent,
    invocation: Any,
    session_id: str,
    prepared_plan: Any,
    history: tuple[Any, ...] = (),
) -> Any:
    """Use one admitted plan for execution; keep the regular unbudgeted path.

    ``history`` must come from an authority that verified the session owner,
    never from the request body.
    """
    attachments = tuple(invocation.attachments or ())
    if prepared_plan is None:
        return agent.stream(
            invocation.query,
            attachments=attachments,
            session_id=session_id,
            **({"history": history} if history else {}),
        )
    from symfonic.agent.facade_turn import turn_for

    request = turn_for(invocation.query, attachments, history, session_id=session_id)
    return agent._kernel.stream(prepared_plan, request)