symfonic.core.provider_request_boundary

provider_request_boundary

Private invocation-local signals at exact provider request boundaries.

OpenAIPayloadMatch

Bases: StrEnum

Text-only agreement at SDK request handoff; never delivery evidence.

install_openai_request_observer

install_openai_request_observer(model: Any, chat_cls: type) -> Any

Observe calls on this model's SDK client during a bound attempt.

The SDK client is model-owned and shared by Pydantic's shallow model_copy; a supplied httpx client is never modified. Mocked constructors pass through. A Runnable's direct call on the same SDK client is also observed and compared with the bound transcript; neither result proves delivery.

Source code in src/symfonic/core/provider_request_boundary.py

def install_openai_request_observer(model: Any, chat_cls: type) -> Any:
    """Observe calls on this model's SDK client during a bound attempt.

    The SDK client is model-owned and shared by Pydantic's shallow model_copy;
    a supplied httpx client is never modified. Mocked constructors pass through.
    A Runnable's direct call on the same SDK client is also observed and
    compared with the bound transcript; neither result proves delivery.
    """
    if type(model) is not chat_cls:
        return model
    sdk = getattr(model, "root_async_client", None)
    request = getattr(sdk, "request", None)
    if not iscoroutinefunction(request):
        return model

    async def observed_request(*args: Any, **kwargs: Any) -> Any:
        options = args[1] if len(args) > 1 else kwargs.get("options")
        mark_request_boundary("openai", "sdk_request_handoff", options)
        return await request(*args, **kwargs)

    with suppress(AttributeError, TypeError):
        sdk.request = observed_request
    return model

mark_request_boundary

mark_request_boundary(provider: str, boundary: str, request_options: Any = None) -> None

Record only the boundary actually crossed by a supported adapter.

Source code in src/symfonic/core/provider_request_boundary.py

def mark_request_boundary(provider: str, boundary: str, request_options: Any = None) -> None:
    """Record only the boundary actually crossed by a supported adapter."""
    observer = _observer.get()
    if observer is not None and observer.active:
        match = OpenAIPayloadMatch.UNKNOWN
        if provider == "openai":
            actual = _sdk_text_messages(request_options)
            expected = observer.bound_messages
            if actual is not None and expected is not None:
                match = (
                    OpenAIPayloadMatch.MATCHED
                    if actual == expected
                    else OpenAIPayloadMatch.MISMATCHED
                )
        observer.callback(provider, boundary, match)

suspend_request_boundary

suspend_request_boundary() -> Iterator[None]

Keep a streaming consumer outside the provider's observation scope.

Source code in src/symfonic/core/provider_request_boundary.py

@contextmanager
def suspend_request_boundary() -> Iterator[None]:
    """Keep a streaming consumer outside the provider's observation scope."""
    token = _observer.set(None)
    try:
        yield
    finally:
        _observer.reset(token)