symfonic.agent.stream_usage_contract

stream_usage_contract

Conservative provider admission for budgeted native streams.

This is a declaration, not observed usage. The HTTP route still requires the tenant-bound telemetry sink, and default cutover still needs live evidence.

declares_stream_usage

declares_stream_usage(provider: Any) -> bool

Whether a trusted composition explicitly promises streamed usage.

The four built-in API-key adapters configure ChatOpenAI with stream_usage=True. Other built-ins are unproven here. A custom adapter may opt in with supports_metered_stream() is True; that is an operator contract to test against its actual API/model, not an inferred capability from supports_streaming() or a provider-family string.

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

def declares_stream_usage(provider: Any) -> bool:
    """Whether a trusted composition explicitly promises streamed usage.

    The four built-in API-key adapters configure ``ChatOpenAI`` with
    ``stream_usage=True``. Other built-ins are unproven here. A custom adapter
    may opt in with ``supports_metered_stream() is True``; that is an operator
    contract to test against its actual API/model, not an inferred capability
    from ``supports_streaming()`` or a provider-family string.
    """
    from symfonic.core.providers import (
        DeepSeekProvider,
        KimiProvider,
        OpenAIProvider,
        OpenRouterProvider,
    )

    if type(provider) in (OpenAIProvider, DeepSeekProvider, KimiProvider, OpenRouterProvider):
        return True
    declaration = getattr(provider, "supports_metered_stream", None)
    if not callable(declaration):
        return False
    try:
        return declaration() is True
    except Exception:  # noqa: BLE001 - an uncertain contract refuses admission
        return False