Skip to content

symfonic.core.codex_backend

codex_backend

Codex backend plumbing: credential names, error types, payload adaptation.

Split out of :mod:symfonic.core.codex_provider (378 lines against the 300-line budget). codex_provider is the ModelProvider implementation; this module is everything it needs to talk to the Codex backend and nothing to do with the provider protocol -- the environment variables the token is read from, the headers the backend attributes usage by, the two error types callers catch, and the ChatOpenAI subclass that rewrites system -> developer.

codex_provider re-exports CodexCredentialError, CodexProductionMisuseError and _token_fingerprint, which is how the existing tests and adopters import them.

CodexCredentialError

Bases: RuntimeError

Raised when a ChatGPT subscription token cannot be resolved.

CodexProductionMisuseError

Bases: RuntimeError

Raised when CodexOAuthProvider is used in a production-like env.

install_auth_failure_log_hook 

install_auth_failure_log_hook(kwargs: dict[str, Any], *, logger: Logger) -> None

Attach a log-only 401/403 response hook to sync + async transports.

Composes onto an adopter-supplied http_client / http_async_client if present; otherwise builds fresh httpx clients (adopting any resolved timeout). Purely observability — see _oauth_errors for the never-raise / body-untouched contract.

Source code in src/symfonic/core/codex_backend.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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
129
130
131
132
133
134
135
136
137
138
139
140
def install_auth_failure_log_hook(
    kwargs: dict[str, Any], *, logger: logging.Logger
) -> None:
    """Attach a log-only 401/403 response hook to sync + async transports.

    Composes onto an adopter-supplied ``http_client`` / ``http_async_client``
    if present; otherwise builds fresh httpx clients (adopting any resolved
    ``timeout``). Purely observability — see ``_oauth_errors`` for the
    never-raise / body-untouched contract.
    """
    import httpx

    from ._oauth_errors import (
        make_auth_failure_log_hook,
        make_auth_failure_log_hook_async,
    )

    remediation = (
        "Your ChatGPT/Codex subscription token is expired or invalid. Run "
        "`codex login` (or re-export OPENAI_CHATGPT_ACCESS_TOKEN), then "
        "retry."
    )
    resp_sync = make_auth_failure_log_hook(
        logger, provider="Codex", remediation=remediation
    )
    resp_async = make_auth_failure_log_hook_async(
        logger, provider="Codex", remediation=remediation
    )

    timeout_val = kwargs.get("timeout")
    httpx_timeout = (
        httpx.Timeout(float(timeout_val))
        if isinstance(timeout_val, (int, float))
        else httpx.Timeout(60.0)
    )

    existing_async = kwargs.get("http_async_client")
    if existing_async is not None:
        existing_async.event_hooks.setdefault("response", []).append(resp_async)
    else:
        kwargs["http_async_client"] = httpx.AsyncClient(
            timeout=httpx_timeout, event_hooks={"response": [resp_async]}
        )

    existing_sync = kwargs.get("http_client")
    if existing_sync is not None:
        existing_sync.event_hooks.setdefault("response", []).append(resp_sync)
    else:
        kwargs["http_client"] = httpx.Client(
            timeout=httpx_timeout, event_hooks={"response": [resp_sync]}
        )