Codex Image

platform: openai
endpoint: https://api.openai.com/v1/responses
auth: chatgpt-oauth
models:
  - id: gpt-5.5
    role: top-level
    note: passed as the Responses API top-level "model" field.
  - id: gpt-image-2
    role: hosted-tool-default
    note: |
      Not sent explicitly; the Responses API picks the default image model
      when "tools[0].model" is absent. Mirrors codex's behaviour.
tools:
  - type: image_generation
client_id: eci-prd-pub-codex-123
scopes:
  - openid
  - profile
  - email
  - offline_access
  - api.connectors.read
  - api.connectors.invoke
fingerprint:
  originator: codex_cli_rs
  pretend_version: 0.45.0
  user_agent_template: "codex_cli_rs/{version} ({os_type} {os_ver}; {arch}) {terminal}"

[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "codex-image"
version = "0.1.0"
description = "Generate or edit images via OpenAI's Responses API using ChatGPT account auth reused from Codex CLI."
readme = "README.md"
requires-python = ">=3.10"
license = { text = "MIT" }
authors = [{ name = "codex-image contributors" }]
dependencies = [
    "requests>=2.31",
    "httpx[http2,socks]>=0.27",
    "pyjwt>=2.8",
    "pillow>=10",
]

[project.optional-dependencies]
dev = [
    "pytest>=8",
    "pytest-mock",
    "responses>=0.25",
]

[project.scripts]
codex-image = "codex_image:main"

[tool.setuptools]
package-dir = { "" = "scripts" }
py-modules = ["codex_image", "auth", "responses_client", "http_client", "output"]

[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["scripts"]

"""Authentication layer.

Resolves a ChatGPT access token, with three strategies in order:

1. Read ``$CODEX_HOME/auth.json``.
2. If close to expiry, ``refresh_tokens()``.
3. If still no luck, ``interactive_login()`` via a PKCE OAuth browser flow.

The on-disk file format matches the official ``codex`` CLI so the two
clients share credentials transparently.
"""

from __future__ import annotations

import base64
import hashlib
import http.server
import json
import os
import secrets
import socket
import threading
import time
import urllib.parse
import webbrowser
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import jwt as pyjwt
import requests

from http_client import OAUTH_CLIENT_ID, OAUTH_SCOPE, ORIGINATOR, make_session

# ---------------------------------------------------------------------------
# Constants.
# ---------------------------------------------------------------------------

AUTH_HOST = "https://auth.openai.com"
AUTHORIZE_URL = f"{AUTH_HOST}/oauth/authorize"
TOKEN_URL = f"{AUTH_HOST}/oauth/token"

REDIRECT_PORTS = (1455, 1457)
REDIRECT_PATH = "/auth/callback"

REFRESH_LEEWAY_SECONDS = 5 * 60  # refresh when access token has <5min left
OAUTH_TIMEOUT_SECONDS = 90

UNRECOVERABLE_REFRESH_ERRORS = {
    "refresh_token_expired",
    "refresh_token_reused",
    "refresh_token_invalidated",
    "invalid_grant",
}


# ---------------------------------------------------------------------------
# Data types.
# ---------------------------------------------------------------------------


@dataclass
class Credentials:
    access_token: str
    refresh_token: str | None
    id_token: str | None
    account_id: str | None
    last_refresh: str
    source: Path
    extra: dict[str, Any] = field(default_factory=dict)


class AuthError(RuntimeError):
    """Raised when no usable credential can be obtained."""


# ---------------------------------------------------------------------------
# auth.json paths.
# ---------------------------------------------------------------------------


def codex_home() -> Path:
    explicit = os.environ.get("CODEX_HOME")
    if explicit:
        return Path(explicit).expanduser()
    return Path.home() / ".codex"


def auth_path() -> Path:
    return codex_home() / "auth.json"


# ---------------------------------------------------------------------------
# Low-level helpers.
# ---------------------------------------------------------------------------


def _load_auth_json() -> dict | None:
    path = auth_path()
    try:
        raw = path.read_text(encoding="utf-8")
    except FileNotFoundError:
        return None
    except OSError as exc:
        raise AuthError(f"failed to read {path}: {exc}") from exc
    try:
        return json.loads(raw)
    except json.JSONDecodeError as exc:
        # Don't silently overwrite; back up the bad file so user can inspect.
        backup = path.with_suffix(path.suffix + f".broken-{int(time.time())}")
        try:
            path.rename(backup)
        except OSError:
            pass
        raise AuthError(
            f"{path} contained invalid JSON, backed up to {backup}: {exc}"
        ) from exc


def _write_auth_json(data: dict) -> None:
    path = auth_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(data, indent=2), encoding="utf-8")
    os.chmod(tmp, 0o600)
    os.replace(tmp, path)


def _decode_jwt_exp(token: str) -> int | None:
    """Return the ``exp`` claim of *token*, or ``None`` if not a JWT."""

    try:
        claims = pyjwt.decode(token, options={"verify_signature": False})
    except pyjwt.PyJWTError:
        return None
    exp = claims.get("exp")
    return int(exp) if isinstance(exp, (int, float)) else None


def _now() -> int:
    return int(time.time())


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def _credentials_from_data(data: dict) -> Credentials | None:
    tokens = data.get("tokens")
    if not tokens or not isinstance(tokens, dict):
        return None
    access = tokens.get("access_token")
    if not access:
        return None
    return Credentials(
        access_token=access,
        refresh_token=tokens.get("refresh_token"),
        id_token=tokens.get("id_token"),
        account_id=tokens.get("account_id"),
        last_refresh=data.get("last_refresh") or _now_iso(),
        source=auth_path(),
        extra={k: v for k, v in data.items() if k not in {"tokens", "last_refresh"}},
    )


def _data_from_credentials(creds: Credentials) -> dict:
    data: dict = {"auth_mode": "Chatgpt"}
    data.update(creds.extra)
    data["tokens"] = {
        "id_token": creds.id_token,
        "access_token": creds.access_token,
        "refresh_token": creds.refresh_token,
        "account_id": creds.account_id,
    }
    data["last_refresh"] = creds.last_refresh
    return data


# ---------------------------------------------------------------------------
# Public API.
# ---------------------------------------------------------------------------


def get_access_token() -> str:
    """Return a usable access token, refreshing or logging in as needed."""

    return get_access_credentials().access_token


def get_access_credentials() -> Credentials:
    """Return usable credentials including ``account_id``.

    Callers that need to attach ``chatgpt-account-id`` to outgoing requests
    (e.g. the ChatGPT-account ``responses`` endpoint) should prefer this
    helper over :func:`get_access_token` so the id stays in sync with the
    token that was actually selected/refreshed.
    """

    return _resolve_credentials()


def _resolve_credentials() -> Credentials:
    try:
        data = _load_auth_json()
    except AuthError:
        # Corrupted file already backed up — go straight to OAuth.
        return interactive_login()

    creds = _credentials_from_data(data) if data else None
    if creds is None:
        return interactive_login()

    exp = _decode_jwt_exp(creds.access_token)
    if exp is not None and exp - _now() > REFRESH_LEEWAY_SECONDS:
        return creds

    if creds.refresh_token:
        try:
            return refresh_tokens(creds.refresh_token, force=False)
        except AuthError:
            return interactive_login()

    return interactive_login()


def refresh_tokens(refresh_token: str, *, force: bool = False) -> Credentials:
    """Exchange *refresh_token* for a fresh access token.

    Parameters
    ----------
    force:
        If True, perform the refresh even when the current token still has
        plenty of validity. Used by the CLI's 401 recovery path.
    """

    payload = {
        "client_id": OAUTH_CLIENT_ID,
        "grant_type": "refresh_token",
        "refresh_token": refresh_token,
    }

    session = make_session()
    last_error: str | None = None
    for attempt in range(3):
        try:
            response = session.post(TOKEN_URL, json=payload, timeout=30)
        except requests.RequestException as exc:
            last_error = str(exc)
            time.sleep(2 ** (attempt + 1))
            continue

        if response.status_code == 200:
            return _store_refreshed_tokens(response.json(), refresh_token)

        # Hard failures — never retry, propagate.
        if response.status_code in (400, 401, 403):
            body: dict[str, Any] = {}
            try:
                body = response.json()
            except ValueError:
                pass
            error_raw = body.get("error") or ""
            error_code = error_raw if isinstance(error_raw, str) else error_raw.get("code", "")
            if error_code in UNRECOVERABLE_REFRESH_ERRORS:
                _clear_tokens_field()
            raise AuthError(
                f"refresh failed ({response.status_code}): {body or response.text!r}"
            )

        # 5xx — back off and retry.
        last_error = f"HTTP {response.status_code}: {response.text!r}"
        time.sleep(2 ** (attempt + 1))

    if force:
        raise AuthError(f"refresh failed after retries: {last_error}")
    raise AuthError(f"refresh failed after retries: {last_error}")


def _store_refreshed_tokens(payload: dict, prior_refresh: str) -> Credentials:
    existing = _load_auth_json() or {}
    extra = {k: v for k, v in existing.items() if k not in {"tokens", "last_refresh"}}
    tokens = existing.get("tokens") or {}

    access = payload.get("access_token") or tokens.get("access_token")
    if not access:
        raise AuthError(f"refresh response missing access_token: {payload!r}")
    refresh = payload.get("refresh_token") or prior_refresh
    id_token = payload.get("id_token") or tokens.get("id_token")
    account_id = payload.get("account_id") or tokens.get("account_id")

    creds = Credentials(
        access_token=access,
        refresh_token=refresh,
        id_token=id_token,
        account_id=account_id,
        last_refresh=_now_iso(),
        source=auth_path(),
        extra=extra if extra else {"auth_mode": "Chatgpt"},
    )
    _write_auth_json(_data_from_credentials(creds))
    return creds


def _clear_tokens_field() -> None:
    """Remove the ``tokens`` field from auth.json without deleting the file."""

    try:
        data = _load_auth_json()
    except AuthError:
        return
    if not data:
        return
    data.pop("tokens", None)
    data["auth_mode"] = data.get("auth_mode", "Chatgpt")
    data.setdefault("last_refresh", _now_iso())
    try:
        _write_auth_json(data)
    except OSError:
        pass


def logout() -> None:
    """Strip tokens from auth.json. File shape and other fields are kept."""

    _clear_tokens_field()


def status() -> dict:
    """Return a non-secret summary of the current credential, or raise."""

    data = _load_auth_json()
    if not data:
        raise AuthError("no credential found")
    creds = _credentials_from_data(data)
    if not creds:
        raise AuthError("auth.json present but no tokens stored")

    info: dict[str, Any] = {
        "source": str(creds.source),
        "account_id": creds.account_id,
        "last_refresh": creds.last_refresh,
    }
    exp = _decode_jwt_exp(creds.access_token)
    if exp is not None:
        info["expires_at"] = datetime.fromtimestamp(exp, timezone.utc).isoformat(
            timespec="seconds"
        )

    if creds.id_token:
        try:
            claims = pyjwt.decode(creds.id_token, options={"verify_signature": False})
        except pyjwt.PyJWTError:
            claims = {}
        info["email"] = claims.get("email")
        ns = claims.get("https://api.openai.com/auth") or {}
        if isinstance(ns, dict):
            info["plan"] = ns.get("chatgpt_plan_type") or ns.get("plan_type")

    return info


# ---------------------------------------------------------------------------
# PKCE OAuth flow.
# ---------------------------------------------------------------------------


def _make_pkce_pair() -> tuple[str, str]:
    verifier = base64.urlsafe_b64encode(secrets.token_bytes(43)).rstrip(b"=").decode()
    digest = hashlib.sha256(verifier.encode()).digest()
    challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
    return verifier, challenge


def _make_state() -> str:
    return base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b"=").decode()


class _CallbackHandler(http.server.BaseHTTPRequestHandler):
    server_version = "codex-image-callback/1.0"

    def do_GET(self) -> None:  # noqa: N802 — required name
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path != REDIRECT_PATH:
            self.send_response(404)
            self.end_headers()
            return
        query = urllib.parse.parse_qs(parsed.query)
        result: dict[str, str | None] = {
            "code": (query.get("code") or [None])[0],
            "state": (query.get("state") or [None])[0],
            "error": (query.get("error") or [None])[0],
            "error_description": (query.get("error_description") or [None])[0],
        }
        # Stash on the server instance so the main thread can pick it up.
        self.server.oauth_result = result  # type: ignore[attr-defined]
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.end_headers()
        if result["error"]:
            body = f"<h1>Login failed</h1><p>{result['error']}</p>"
        else:
            body = "<h1>codex-image login complete</h1><p>You can close this tab.</p>"
        self.wfile.write(body.encode("utf-8"))

    def log_message(self, *_args, **_kwargs) -> None:  # noqa: D401
        # Silence the default stderr logger.
        return


def _bind_callback_server() -> tuple[http.server.HTTPServer, int]:
    last_exc: OSError | None = None
    for port in REDIRECT_PORTS:
        try:
            server = http.server.HTTPServer(("127.0.0.1", port), _CallbackHandler)
        except OSError as exc:
            last_exc = exc
            continue
        server.oauth_result = None  # type: ignore[attr-defined]
        return server, port
    raise AuthError(
        f"could not bind any of {REDIRECT_PORTS!r} for OAuth callback: {last_exc}"
    )


def interactive_login() -> Credentials:
    """Drive the PKCE OAuth flow end-to-end and persist the result."""

    verifier, challenge = _make_pkce_pair()
    state = _make_state()
    server, port = _bind_callback_server()
    redirect_uri = f"http://localhost:{port}{REDIRECT_PATH}"

    params = {
        "response_type": "code",
        "client_id": OAUTH_CLIENT_ID,
        "redirect_uri": redirect_uri,
        "scope": OAUTH_SCOPE,
        "code_challenge": challenge,
        "code_challenge_method": "S256",
        "state": state,
        "id_token_add_organizations": "true",
        "codex_cli_simplified_flow": "true",
        "originator": ORIGINATOR,
    }
    url = f"{AUTHORIZE_URL}?{urllib.parse.urlencode(params)}"

    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()

    print(f"Opening browser for OAuth login...\nIf nothing opens, visit:\n{url}")
    try:
        webbrowser.open(url)
    except Exception:  # noqa: BLE001 — best effort
        pass

    deadline = time.monotonic() + OAUTH_TIMEOUT_SECONDS
    try:
        while time.monotonic() < deadline:
            result = getattr(server, "oauth_result", None)
            if result is not None:
                break
            time.sleep(0.2)
        else:
            raise AuthError("OAuth login timed out waiting for callback")
    finally:
        server.shutdown()
        server.server_close()

    if result["error"]:
        raise AuthError(
            f"OAuth error: {result['error']} {result.get('error_description') or ''}"
        )
    if result["state"] != state:
        raise AuthError("OAuth state mismatch — possible CSRF, aborting")
    if not result["code"]:
        raise AuthError("OAuth callback missing code parameter")

    token_payload = {
        "client_id": OAUTH_CLIENT_ID,
        "grant_type": "authorization_code",
        "code": result["code"],
        "redirect_uri": redirect_uri,
        "code_verifier": verifier,
    }
    session = make_session()
    response = session.post(TOKEN_URL, json=token_payload, timeout=30)
    if response.status_code != 200:
        raise AuthError(
            f"token exchange failed ({response.status_code}): {response.text!r}"
        )
    payload = response.json()

    access = payload.get("access_token")
    if not access:
        raise AuthError(f"token response missing access_token: {payload!r}")

    creds = Credentials(
        access_token=access,
        refresh_token=payload.get("refresh_token"),
        id_token=payload.get("id_token"),
        account_id=payload.get("account_id"),
        last_refresh=_now_iso(),
        source=auth_path(),
        extra={"auth_mode": "Chatgpt"},
    )
    _write_auth_json(_data_from_credentials(creds))
    return creds


def _port_in_use(port: int) -> bool:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        return sock.connect_ex(("127.0.0.1", port)) == 0

"""``codex-image`` command-line entry point."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Iterable

import auth
import output
import responses_client as rc


# ---------------------------------------------------------------------------
# Exit codes.
# ---------------------------------------------------------------------------


EXIT_OK = 0
EXIT_USER_ERROR = 2
EXIT_FS_ERROR = 3
EXIT_AUTH_ERROR = 4
EXIT_NETWORK_ERROR = 5
EXIT_PARTIAL = 6
EXIT_API_ERROR = 7
EXIT_INTERRUPT = 130


# ---------------------------------------------------------------------------
# Argument parser.
# ---------------------------------------------------------------------------


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="codex-image",
        description=(
            "Generate or edit images via OpenAI's Responses API using "
            "ChatGPT-account auth reused from the codex CLI."
        ),
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    gen = subparsers.add_parser("generate", help="generate a new image")
    _add_common_image_flags(gen)
    gen.add_argument("prompt", help="prompt describing the desired image")

    edit = subparsers.add_parser("edit", help="edit reference image(s)")
    _add_common_image_flags(edit)
    edit.add_argument(
        "--input",
        dest="inputs",
        action="append",
        type=Path,
        required=True,
        help="path to a reference image (repeatable)",
    )
    edit.add_argument("prompt", help="prompt describing the change to apply")

    subparsers.add_parser("login", help="force PKCE OAuth browser flow")
    subparsers.add_parser("logout", help="clear tokens from auth.json")
    subparsers.add_parser("status", help="print current credential metadata")

    return parser


def _add_common_image_flags(sub: argparse.ArgumentParser) -> None:
    sub.add_argument(
        "--output-format",
        choices=("png", "webp", "jpeg"),
        default="png",
    )
    sub.add_argument("--out", type=Path, default=None, help="output path")
    sub.add_argument(
        "--force",
        action="store_true",
        help="overwrite existing file at --out",
    )
    sub.add_argument(
        "--image-model",
        default=None,
        help=(
            "advanced: set tools[0].model. Default unset to match codex "
            "byte-for-byte."
        ),
    )


# ---------------------------------------------------------------------------
# Dispatch.
# ---------------------------------------------------------------------------


def main(argv: list[str] | None = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)

    try:
        if args.command == "generate":
            return _cmd_generate(args)
        if args.command == "edit":
            return _cmd_edit(args)
        if args.command == "login":
            return _cmd_login()
        if args.command == "logout":
            return _cmd_logout()
        if args.command == "status":
            return _cmd_status()
    except KeyboardInterrupt:
        print("\nInterrupted.", file=sys.stderr)
        return EXIT_INTERRUPT

    parser.error(f"unknown command {args.command!r}")
    return EXIT_USER_ERROR


# ---------------------------------------------------------------------------
# Subcommand handlers.
# ---------------------------------------------------------------------------


def _cmd_generate(args: argparse.Namespace) -> int:
    return _run_image_call(
        prompt=args.prompt,
        inputs=None,
        output_format=args.output_format,
        out=args.out,
        force=args.force,
        image_model=args.image_model,
    )


def _cmd_edit(args: argparse.Namespace) -> int:
    inputs: list[Path] = list(args.inputs or [])
    for path in inputs:
        if not path.exists():
            print(f"error: reference image not found: {path}", file=sys.stderr)
            return EXIT_USER_ERROR
    return _run_image_call(
        prompt=args.prompt,
        inputs=inputs,
        output_format=args.output_format,
        out=args.out,
        force=args.force,
        image_model=args.image_model,
    )


def _run_image_call(
    *,
    prompt: str,
    inputs: Iterable[Path] | None,
    output_format: str,
    out: Path | None,
    force: bool,
    image_model: str | None,
) -> int:
    try:
        creds = auth.get_access_credentials()
    except auth.AuthError as exc:
        print(f"auth error: {exc}", file=sys.stderr)
        return EXIT_AUTH_ERROR

    try:
        result = _call_generate_with_refresh(
            token=creds.access_token,
            account_id=creds.account_id,
            prompt=prompt,
            inputs=inputs,
            output_format=output_format,
            image_model=image_model,
        )
    except rc.UnauthorizedError as exc:
        print(f"auth error: {exc}", file=sys.stderr)
        return EXIT_AUTH_ERROR
    except rc.ForbiddenError as exc:
        print(f"forbidden: {exc}", file=sys.stderr)
        return EXIT_AUTH_ERROR
    except rc.RateLimitedError as exc:
        print(f"rate limited: {exc}", file=sys.stderr)
        return EXIT_NETWORK_ERROR
    except rc.StreamBrokenError as exc:
        print(f"stream broken: {exc}", file=sys.stderr)
        return EXIT_PARTIAL
    except rc.ApiError as exc:
        print(f"api error: {exc}", file=sys.stderr)
        return EXIT_API_ERROR
    except ValueError as exc:
        print(f"input error: {exc}", file=sys.stderr)
        return EXIT_USER_ERROR

    try:
        saved = output.save(
            result.image_b64,
            out,
            force=force,
            slug_source=prompt,
            output_format=output_format,
        )
    except FileNotFoundError as exc:
        print(f"filesystem error: {exc}", file=sys.stderr)
        return EXIT_FS_ERROR
    except (PermissionError, OSError) as exc:
        print(f"filesystem error: {exc}", file=sys.stderr)
        return EXIT_FS_ERROR
    except ValueError as exc:
        print(f"api returned invalid data: {exc}", file=sys.stderr)
        return EXIT_API_ERROR

    print(f"Saved: {saved}")
    if result.revised_prompt:
        print(f"Revised prompt: {result.revised_prompt}")
    return EXIT_OK


def _call_generate_with_refresh(
    *,
    token: str,
    account_id: str | None,
    prompt: str,
    inputs: Iterable[Path] | None,
    output_format: str,
    image_model: str | None,
) -> rc.GenerateResult:
    """Run generate(); on 401 force-refresh once and retry."""

    try:
        return rc.generate(
            prompt,
            access_token=token,
            account_id=account_id,
            input_images=list(inputs) if inputs else None,
            output_format=output_format,
            image_model=image_model,
        )
    except rc.UnauthorizedError:
        refreshed = _force_refresh_credentials()
        return rc.generate(
            prompt,
            access_token=refreshed.access_token,
            account_id=refreshed.account_id,
            input_images=list(inputs) if inputs else None,
            output_format=output_format,
            image_model=image_model,
        )


def _force_refresh_credentials() -> auth.Credentials:
    data = auth._load_auth_json()  # noqa: SLF001 — intentional internal call
    refresh = (data or {}).get("tokens", {}).get("refresh_token") if data else None
    if refresh:
        try:
            return auth.refresh_tokens(refresh, force=True)
        except auth.AuthError:
            pass
    return auth.interactive_login()


def _cmd_login() -> int:
    try:
        creds = auth.interactive_login()
    except auth.AuthError as exc:
        print(f"login failed: {exc}", file=sys.stderr)
        return EXIT_AUTH_ERROR
    print(f"Logged in. Tokens written to {creds.source}.")
    return EXIT_OK


def _cmd_logout() -> int:
    auth.logout()
    print("Logged out (tokens cleared, auth.json shape preserved).")
    return EXIT_OK


def _cmd_status() -> int:
    try:
        info = auth.status()
    except auth.AuthError as exc:
        print(f"status error: {exc}", file=sys.stderr)
        return EXIT_AUTH_ERROR
    print(json.dumps(info, indent=2, sort_keys=True, default=str))
    return EXIT_OK


if __name__ == "__main__":
    raise SystemExit(main())

"""Shared HTTP session factory.

Centralises all "fingerprint" headers so :mod:`auth` and
:mod:`responses_client` send byte-identical metadata.

The codex CLI emits a small, deterministic set of headers; this module
constructs the same set and explicitly removes the headers that
``requests`` adds by default but ``codex`` does not.
"""

from __future__ import annotations

import os
import platform
import uuid
from pathlib import Path

import requests

# ---------------------------------------------------------------------------
# Fingerprint constants — keep in sync with references/fingerprint.md.
# ---------------------------------------------------------------------------

ORIGINATOR = "codex_cli_rs"
CODEX_PRETEND_VERSION = "0.130.0"
# OAuth client_id used by the official codex CLI (codex-rs/login/src/auth/manager.rs).
OAUTH_CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"
OAUTH_SCOPE = (
    "openid profile email offline_access "
    "api.connectors.read api.connectors.invoke"
)
USER_AGENT_TEMPLATE = "codex_cli_rs/{version} ({os_type} {os_ver}; {arch}) {terminal}"
# Required by chatgpt.com/backend-api/codex/responses for ChatGPT-account tokens.
OPENAI_BETA_HEADER_VALUE = "responses=experimental"


def _detect_terminal() -> str:
    return os.environ.get("TERM_PROGRAM") or "unknown"


def build_user_agent() -> str:
    return USER_AGENT_TEMPLATE.format(
        version=CODEX_PRETEND_VERSION,
        os_type=platform.system(),
        os_ver=platform.release(),
        arch=platform.machine(),
        terminal=_detect_terminal(),
    )


# ---------------------------------------------------------------------------
# Installation id — UUIDv4 persisted on disk so successive invocations look
# like the same physical client. Stored separately from auth.json so logout
# does not invalidate it.
# ---------------------------------------------------------------------------


def _installation_id_path() -> Path:
    explicit = os.environ.get("CODEX_IMAGE_INSTALLATION_ID_FILE")
    if explicit:
        return Path(explicit).expanduser()
    cache_root = os.environ.get("XDG_CACHE_HOME")
    if cache_root:
        return Path(cache_root).expanduser() / "codex-image" / "installation_id"
    return Path.home() / ".codex-image" / "installation_id"


def get_installation_id() -> str:
    """Return a stable installation UUIDv4, creating it on first use."""

    path = _installation_id_path()
    try:
        existing = path.read_text(encoding="utf-8").strip()
        if existing:
            return existing
    except FileNotFoundError:
        pass
    except OSError:
        # Disk error reading the file — fall through and try to recreate it.
        pass

    new_id = str(uuid.uuid4())
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(new_id, encoding="utf-8")
    except OSError:
        # Best effort. If the disk is unwritable we still return a UUID for
        # this invocation — the fingerprint then only differs across runs.
        pass
    return new_id


# Session-id is per process, not per disk.
_SESSION_ID = str(uuid.uuid4())


def get_session_id() -> str:
    return _SESSION_ID


# ---------------------------------------------------------------------------
# Session factory.
# ---------------------------------------------------------------------------


def make_session(
    *,
    with_auth_token: str | None = None,
    account_id: str | None = None,
    include_chatgpt_beta: bool = False,
) -> requests.Session:
    """Return a :class:`requests.Session` carrying the codex fingerprint.

    Parameters
    ----------
    with_auth_token:
        Optional bearer token to inject as ``Authorization`` header. The
        OAuth token endpoint must not receive a bearer header, so callers
        for that path pass ``None``.
    account_id:
        Optional ChatGPT account id. When provided, attaches the
        ``chatgpt-account-id`` header expected by
        ``chatgpt.com/backend-api/codex/responses``.
    include_chatgpt_beta:
        When True, attaches ``OpenAI-Beta: responses=experimental`` which
        the ChatGPT-account responses endpoint requires.
    """

    session = requests.Session()

    # Drop requests defaults that codex never sends.
    for header in ("User-Agent", "Accept", "Accept-Encoding"):
        session.headers.pop(header, None)

    session.headers.update(
        {
            "User-Agent": build_user_agent(),
            "originator": ORIGINATOR,
            "session-id": get_session_id(),
            "x-codex-installation-id": get_installation_id(),
        }
    )

    if with_auth_token is not None:
        session.headers["Authorization"] = f"Bearer {with_auth_token}"
    if account_id:
        session.headers["chatgpt-account-id"] = account_id
    if include_chatgpt_beta:
        session.headers["OpenAI-Beta"] = OPENAI_BETA_HEADER_VALUE

    return session

"""Filesystem output helpers."""

from __future__ import annotations

import base64
import os
import re
from datetime import datetime, timezone
from pathlib import Path

from auth import codex_home

EXTENSION_BY_FORMAT = {
    "png": ".png",
    "webp": ".webp",
    "jpeg": ".jpg",
    "jpg": ".jpg",
}


def default_output_dir() -> Path:
    return codex_home() / "generated_images" / "codex-image"


def _slugify(text: str, *, max_length: int = 48) -> str:
    cleaned = re.sub(r"[^A-Za-z0-9]+", "-", text.strip().lower()).strip("-")
    if not cleaned:
        cleaned = "image"
    return cleaned[:max_length].rstrip("-") or "image"


def default_path(slug_source: str, output_format: str) -> Path:
    ext = EXTENSION_BY_FORMAT.get(output_format.lower())
    if ext is None:
        raise ValueError(f"unsupported output format {output_format!r}")
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    slug = _slugify(slug_source)
    return default_output_dir() / f"{timestamp}-{slug}{ext}"


def _avoid_collision(path: Path) -> Path:
    if not path.exists():
        return path
    parent = path.parent
    stem = path.stem
    suffix = path.suffix
    n = 2
    while True:
        candidate = parent / f"{stem}-v{n}{suffix}"
        if not candidate.exists():
            return candidate
        n += 1


def save(
    image_b64: str,
    out_path: Path | None,
    *,
    force: bool = False,
    slug_source: str | None = None,
    output_format: str = "png",
) -> Path:
    """Decode *image_b64* and write it to disk. Returns the absolute path."""

    try:
        data = base64.b64decode(image_b64, validate=True)
    except (ValueError, TypeError) as exc:
        raise ValueError(f"invalid base64 payload: {exc}") from exc

    if out_path is None:
        target = default_path(slug_source or "image", output_format)
        target.parent.mkdir(parents=True, exist_ok=True)
    else:
        target = out_path.expanduser().resolve()
        if not target.parent.exists():
            raise FileNotFoundError(
                f"output directory does not exist: {target.parent}"
            )
        if target.exists() and not force:
            target = _avoid_collision(target)

    tmp = target.with_suffix(target.suffix + ".tmp")
    tmp.write_bytes(data)
    os.replace(tmp, target)
    return target.resolve()

"""Streaming client for the ChatGPT-account ``responses`` endpoint.

Speaks Server-Sent Events, mirrors the codex CLI's request shape exactly,
and extracts the final base64 image from the ``image_generation_call``
output item.

ChatGPT-account tokens (issued by the official ``codex`` CLI's OAuth flow)
can only reach the ``/v1/responses`` surface via
``https://chatgpt.com/backend-api/codex/responses``. The traditional
``api.openai.com/v1/responses`` endpoint rejects them with HTTP 401
``Missing scopes: api.responses.write`` because the token's ``scp`` only
contains ``api.connectors.*``.
"""

from __future__ import annotations

import base64
import io
import json
import mimetypes
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Iterable

import httpx

from http_client import (
    ORIGINATOR,
    OPENAI_BETA_HEADER_VALUE,
    build_user_agent,
    get_installation_id,
    get_session_id,
)

# ChatGPT-account responses endpoint (codex-rs/model-provider-info/src/lib.rs
# CHATGPT_CODEX_BASE_URL = "https://chatgpt.com/backend-api/codex").
RESPONSES_URL = "https://chatgpt.com/backend-api/codex/responses"
# Default model accepted by Codex-with-ChatGPT-account; gpt-5.4 and gpt-5.2
# pass model gating, others return "model is not supported".
TOP_LEVEL_MODEL = "gpt-5.4"
DEFAULT_INSTRUCTIONS = (
    "You are an image-generation assistant. When the user asks for an image, "
    "immediately call the image_generation tool with the user's full "
    "description as the prompt. Do not ask follow-up questions and do not "
    "emit any text output beyond what the tool returns."
)

MAX_INPUT_BYTES_BEFORE_RESIZE = 5 * 1024 * 1024
MAX_INPUT_BYTES_HARD = 10 * 1024 * 1024
MAX_INPUT_LONG_EDGE = 2048

RETRY_BACKOFF_SECONDS = (2.0, 4.0, 8.0)
# How many times to retry when the SSE stream is cut mid-generation (e.g.,
# Cloudflare / local proxy idle-killing the connection before the image is
# emitted). Each attempt re-POSTs from scratch since ``store=false`` means
# the previous response cannot be resumed by id.
MAX_STREAM_RESTARTS = 3


class ResponsesError(Exception):
    """Base error for the Responses API client."""


class UnauthorizedError(ResponsesError):
    """HTTP 401 — caller should refresh the token and retry once."""


class ForbiddenError(ResponsesError):
    """HTTP 403 — usually a plan/permission issue."""


class RateLimitedError(ResponsesError):
    """HTTP 429 retries exhausted."""


class ApiError(ResponsesError):
    """Other HTTP / business error returned by the backend."""


class StreamBrokenError(ResponsesError):
    """SSE stream ended before ``response.completed``."""


# ---------------------------------------------------------------------------
# Result type.
# ---------------------------------------------------------------------------


@dataclass
class GenerateResult:
    image_b64: str
    revised_prompt: str | None
    call_id: str
    model: str
    raw_events: list[dict] = field(default_factory=list)


# ---------------------------------------------------------------------------
# Request body builders.
# ---------------------------------------------------------------------------


def _tool_spec(output_format: str, image_model: str | None) -> dict:
    spec: dict[str, Any] = {"type": "image_generation", "output_format": output_format}
    if image_model:
        spec["model"] = image_model
    return spec


def _build_generate_body(
    prompt: str, *, output_format: str, image_model: str | None
) -> dict:
    return {
        "model": TOP_LEVEL_MODEL,
        "store": False,
        "stream": True,
        "instructions": DEFAULT_INSTRUCTIONS,
        "input": [
            {
                "role": "user",
                "content": [{"type": "input_text", "text": prompt}],
            }
        ],
        "tools": [_tool_spec(output_format, image_model)],
    }


def _build_edit_body(
    prompt: str,
    input_images: Iterable[Path],
    *,
    output_format: str,
    image_model: str | None,
) -> dict:
    content: list[dict[str, Any]] = [{"type": "input_text", "text": prompt}]
    for path in input_images:
        content.append(
            {"type": "input_image", "image_url": _image_to_data_uri(path)}
        )
    return {
        "model": TOP_LEVEL_MODEL,
        "store": False,
        "stream": True,
        "instructions": DEFAULT_INSTRUCTIONS,
        "input": [{"role": "user", "content": content}],
        "tools": [_tool_spec(output_format, image_model)],
    }


def _image_to_data_uri(path: Path) -> str:
    raw = path.read_bytes()
    mime, _ = mimetypes.guess_type(path.name)
    if mime is None or not mime.startswith("image/"):
        mime = "image/png"

    if len(raw) > MAX_INPUT_BYTES_BEFORE_RESIZE:
        raw, mime = _downscale(raw, mime)

    if len(raw) > MAX_INPUT_BYTES_HARD:
        raise ValueError(
            f"reference image {path} is {len(raw)} bytes after downscaling, "
            f"exceeds hard limit of {MAX_INPUT_BYTES_HARD} bytes"
        )

    encoded = base64.b64encode(raw).decode("ascii")
    return f"data:{mime};base64,{encoded}"


def _downscale(raw: bytes, mime: str) -> tuple[bytes, str]:
    from PIL import Image  # imported lazily so tests without Pillow still work

    img = Image.open(io.BytesIO(raw))
    img.load()
    longest = max(img.size)
    if longest <= MAX_INPUT_LONG_EDGE:
        return raw, mime

    scale = MAX_INPUT_LONG_EDGE / float(longest)
    new_size = (max(1, int(img.width * scale)), max(1, int(img.height * scale)))
    img = img.resize(new_size, Image.LANCZOS)

    buf = io.BytesIO()
    fmt = "PNG" if mime == "image/png" else "JPEG"
    if fmt == "JPEG" and img.mode in ("RGBA", "P"):
        img = img.convert("RGB")
    img.save(buf, format=fmt)
    new_mime = "image/png" if fmt == "PNG" else "image/jpeg"
    return buf.getvalue(), new_mime


# ---------------------------------------------------------------------------
# Public entrypoints.
# ---------------------------------------------------------------------------


def generate(
    prompt: str,
    *,
    access_token: str,
    account_id: str | None,
    input_images: Iterable[Path] | None = None,
    output_format: str = "png",
    image_model: str | None = None,
) -> GenerateResult:
    """Call the ChatGPT-account ``responses`` endpoint and return the image.

    Parameters
    ----------
    prompt:
        Natural-language description of the image to generate, or — when
        editing — the change to apply to the reference image(s).
    access_token:
        Bearer token resolved by :mod:`auth`.
    account_id:
        ChatGPT account id (``tokens.account_id`` in ``auth.json``). The
        endpoint requires it as the ``chatgpt-account-id`` header; missing
        or stale ids surface as HTTP 401.
    input_images:
        Optional iterable of reference image paths. When non-empty the call
        is treated as an *edit*.
    output_format:
        ``png`` / ``webp`` / ``jpeg``.
    image_model:
        Advanced override of the ``image_generation`` tool's ``model``
        field. ``None`` (default) lets the server pick.
    """

    if input_images:
        body = _build_edit_body(
            prompt,
            list(input_images),
            output_format=output_format,
            image_model=image_model,
        )
    else:
        body = _build_generate_body(
            prompt, output_format=output_format, image_model=image_model
        )

    headers = _build_headers(access_token, account_id)
    return _stream_with_restarts(body, headers)


def _build_headers(access_token: str, account_id: str | None) -> dict[str, str]:
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "Accept": "text/event-stream",
        "User-Agent": build_user_agent(),
        "originator": ORIGINATOR,
        "session-id": get_session_id(),
        "x-codex-installation-id": get_installation_id(),
        "OpenAI-Beta": OPENAI_BETA_HEADER_VALUE,
    }
    if account_id:
        headers["chatgpt-account-id"] = account_id
    return headers


def _stream_with_restarts(body: dict, headers: dict[str, str]) -> GenerateResult:
    """POST + consume SSE, restarting from scratch when the stream is cut.

    ``store=false`` means a broken stream cannot be resumed by response id,
    so the only recovery is to re-POST. We do this up to
    ``MAX_STREAM_RESTARTS`` times before giving up.
    """

    timeout = httpx.Timeout(connect=30.0, read=900.0, write=60.0, pool=30.0)
    last_partial_error: Exception | None = None

    with httpx.Client(
        http2=True,
        timeout=timeout,
        trust_env=True,
        follow_redirects=False,
    ) as client:
        for restart in range(MAX_STREAM_RESTARTS + 1):
            try:
                return _post_once(client, body, headers)
            except StreamBrokenError as exc:
                last_partial_error = exc
                if restart >= MAX_STREAM_RESTARTS:
                    raise
                delay = RETRY_BACKOFF_SECONDS[
                    min(restart, len(RETRY_BACKOFF_SECONDS) - 1)
                ]
                time.sleep(delay)
                continue

    # Loop only exits via return or raise; this is defensive.
    raise ApiError(f"stream restarts exhausted: {last_partial_error}")


def _post_once(
    client: httpx.Client, body: dict, headers: dict[str, str]
) -> GenerateResult:
    """Single POST attempt with HTTP-status retries (excluding mid-stream cuts)."""

    last_error: Exception | None = None
    for attempt in range(len(RETRY_BACKOFF_SECONDS) + 1):
        try:
            with client.stream(
                "POST", RESPONSES_URL, json=body, headers=headers
            ) as response:
                status = response.status_code

                if status == 200:
                    return _consume_sse(response)

                if status == 401:
                    raise UnauthorizedError(
                        f"access token rejected: {_safe_read(response)}"
                    )
                if status == 403:
                    raise ForbiddenError(f"forbidden: {_safe_read(response)}")

                if status == 429:
                    if attempt >= len(RETRY_BACKOFF_SECONDS):
                        raise RateLimitedError("rate-limit retries exhausted")
                    delay = _retry_after(response, attempt)
                    time.sleep(delay)
                    continue

                if 500 <= status < 600:
                    text = _safe_read(response)
                    if attempt >= len(RETRY_BACKOFF_SECONDS):
                        raise ApiError(f"server error {status}: {text}")
                    time.sleep(RETRY_BACKOFF_SECONDS[attempt])
                    continue

                raise ApiError(f"unexpected status {status}: {_safe_read(response)}")

        except (httpx.ConnectError, httpx.ConnectTimeout, httpx.ReadTimeout) as exc:
            last_error = exc
            if attempt < len(RETRY_BACKOFF_SECONDS):
                time.sleep(RETRY_BACKOFF_SECONDS[attempt])
                continue
            raise ApiError(f"network error: {exc}") from exc

    raise ApiError(f"retries exhausted: {last_error}")


def _retry_after(response: httpx.Response, attempt: int) -> float:
    raw = response.headers.get("Retry-After")
    if not raw:
        return RETRY_BACKOFF_SECONDS[attempt]
    try:
        return float(raw)
    except ValueError:
        return RETRY_BACKOFF_SECONDS[attempt]


def _safe_read(response: httpx.Response) -> str:
    try:
        return response.read().decode("utf-8", errors="replace")[:500]
    except Exception:  # noqa: BLE001
        return "<unreadable body>"


# ---------------------------------------------------------------------------
# SSE consumer.
# ---------------------------------------------------------------------------


def _iter_sse_events(response: httpx.Response) -> Iterable[dict]:
    """Yield decoded JSON payloads from an SSE response.

    Re-raises transport-level errors as :class:`StreamBrokenError` so the
    caller can decide whether to restart the request.
    """

    event_name: str | None = None
    data_chunks: list[str] = []

    line_source = response.iter_lines()
    while True:
        try:
            line = next(line_source)
        except StopIteration:
            break
        except (
            httpx.RemoteProtocolError,
            httpx.ReadError,
            httpx.ReadTimeout,
            httpx.ProtocolError,
        ) as exc:
            raise StreamBrokenError(f"transport error mid-stream: {exc}") from exc

        if line is None:
            continue
        if line == "":
            if data_chunks:
                raw = "\n".join(data_chunks)
                data_chunks.clear()
                try:
                    payload = json.loads(raw)
                except json.JSONDecodeError:
                    payload = {"_raw": raw}
                if event_name and "type" not in payload:
                    payload["type"] = event_name
                event_name = None
                yield payload
            else:
                event_name = None
            continue
        if line.startswith(":"):
            # Comment / keepalive — ignore but the heartbeat keeps the
            # HTTP/2 stream open.
            continue
        if line.startswith("event:"):
            event_name = line[len("event:") :].strip()
            continue
        if line.startswith("data:"):
            data_chunks.append(line[len("data:") :].lstrip())
            continue


def _consume_sse(response: httpx.Response) -> GenerateResult:
    raw_events: list[dict] = []
    final_image: str | None = None
    revised_prompt: str | None = None
    call_id: str | None = None
    failure: dict | None = None

    for event in _iter_sse_events(response):
        raw_events.append(event)
        etype = event.get("type")

        if etype == "response.output_item.done":
            item = event.get("item") or {}
            if item.get("type") == "image_generation_call":
                final_image = item.get("result") or final_image
                revised_prompt = item.get("revised_prompt") or revised_prompt
                call_id = item.get("id") or call_id

        elif etype == "response.completed":
            response_obj = event.get("response") or {}
            for item in response_obj.get("output") or []:
                if item.get("type") == "image_generation_call":
                    final_image = item.get("result") or final_image
                    revised_prompt = item.get("revised_prompt") or revised_prompt
                    call_id = item.get("id") or call_id

        elif etype == "response.failed":
            failure = event.get("response") or event

        elif etype == "error":
            failure = event

    if failure:
        raise ApiError(f"response failed: {failure}")

    if not final_image:
        raise StreamBrokenError("stream ended without producing an image")

    return GenerateResult(
        image_b64=final_image,
        revised_prompt=revised_prompt,
        call_id=call_id or "",
        model=TOP_LEVEL_MODEL,
        raw_events=raw_events,
    )

"""Shared pytest fixtures."""

from __future__ import annotations

import sys
from pathlib import Path

# Make ``scripts/`` importable as top-level modules.
SCRIPTS_DIR = Path(__file__).resolve().parent.parent / "scripts"
if str(SCRIPTS_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPTS_DIR))

"""Unit tests for :mod:`auth`."""

from __future__ import annotations

import base64
import json
import time
from pathlib import Path

import jwt as pyjwt
import pytest
import responses

import auth


# ---------------------------------------------------------------------------
# Helpers.
# ---------------------------------------------------------------------------


def _make_jwt(exp_offset_seconds: int, extra: dict | None = None) -> str:
    claims = {"exp": int(time.time()) + exp_offset_seconds}
    if extra:
        claims.update(extra)
    return pyjwt.encode(claims, "secret", algorithm="HS256")


@pytest.fixture
def codex_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    home = tmp_path / "codex"
    monkeypatch.setenv("CODEX_HOME", str(home))
    return home


def _write_auth(codex_dir: Path, payload: dict) -> Path:
    codex_dir.mkdir(parents=True, exist_ok=True)
    path = codex_dir / "auth.json"
    path.write_text(json.dumps(payload), encoding="utf-8")
    return path


# ---------------------------------------------------------------------------
# _load_auth_json
# ---------------------------------------------------------------------------


def test_load_auth_json_missing(codex_dir: Path) -> None:
    assert auth._load_auth_json() is None


def test_load_auth_json_corrupt_backs_up(codex_dir: Path) -> None:
    codex_dir.mkdir(parents=True, exist_ok=True)
    bad = codex_dir / "auth.json"
    bad.write_text("not json{", encoding="utf-8")
    with pytest.raises(auth.AuthError):
        auth._load_auth_json()
    # Original should have been renamed.
    assert not bad.exists()
    backups = list(codex_dir.glob("auth.json.broken-*"))
    assert backups, "broken file should be backed up"


# ---------------------------------------------------------------------------
# _decode_jwt_exp
# ---------------------------------------------------------------------------


def test_decode_jwt_exp_returns_int() -> None:
    token = _make_jwt(60)
    assert auth._decode_jwt_exp(token) is not None


def test_decode_jwt_exp_invalid_returns_none() -> None:
    assert auth._decode_jwt_exp("not-a-jwt") is None


# ---------------------------------------------------------------------------
# logout
# ---------------------------------------------------------------------------


def test_logout_strips_tokens_but_keeps_file(codex_dir: Path) -> None:
    _write_auth(
        codex_dir,
        {
            "auth_mode": "Chatgpt",
            "tokens": {"access_token": "x", "refresh_token": "y"},
            "last_refresh": "2025-01-01T00:00:00+00:00",
        },
    )
    auth.logout()
    data = json.loads((codex_dir / "auth.json").read_text())
    assert "tokens" not in data
    assert data["auth_mode"] == "Chatgpt"
    assert data["last_refresh"] == "2025-01-01T00:00:00+00:00"


def test_logout_when_no_file_is_noop(codex_dir: Path) -> None:
    auth.logout()  # should not raise
    assert not (codex_dir / "auth.json").exists()


# ---------------------------------------------------------------------------
# status
# ---------------------------------------------------------------------------


def test_status_missing_raises(codex_dir: Path) -> None:
    with pytest.raises(auth.AuthError):
        auth.status()


def test_status_reports_email_plan_and_expiry(codex_dir: Path) -> None:
    id_token = _make_jwt(
        3600,
        {
            "email": "user@example.com",
            "https://api.openai.com/auth": {"chatgpt_plan_type": "plus"},
        },
    )
    access_token = _make_jwt(3600)
    _write_auth(
        codex_dir,
        {
            "auth_mode": "Chatgpt",
            "tokens": {
                "id_token": id_token,
                "access_token": access_token,
                "refresh_token": "r",
                "account_id": "acct_1",
            },
            "last_refresh": "2025-01-01T00:00:00+00:00",
        },
    )
    info = auth.status()
    assert info["email"] == "user@example.com"
    assert info["plan"] == "plus"
    assert info["account_id"] == "acct_1"
    assert info["expires_at"].startswith("20")


# ---------------------------------------------------------------------------
# refresh_tokens — happy path
# ---------------------------------------------------------------------------


@responses.activate
def test_refresh_tokens_success_persists(codex_dir: Path) -> None:
    _write_auth(
        codex_dir,
        {
            "auth_mode": "Chatgpt",
            "tokens": {
                "id_token": _make_jwt(60),
                "access_token": _make_jwt(60),
                "refresh_token": "old-refresh",
                "account_id": "acct_1",
            },
            "last_refresh": "2025-01-01T00:00:00+00:00",
        },
    )

    new_access = _make_jwt(3600)
    responses.add(
        responses.POST,
        auth.TOKEN_URL,
        json={
            "access_token": new_access,
            "refresh_token": "new-refresh",
            "id_token": _make_jwt(3600),
        },
        status=200,
    )

    creds = auth.refresh_tokens("old-refresh", force=True)
    assert creds.access_token == new_access
    assert creds.refresh_token == "new-refresh"

    on_disk = json.loads((codex_dir / "auth.json").read_text())
    assert on_disk["tokens"]["access_token"] == new_access
    assert on_disk["tokens"]["refresh_token"] == "new-refresh"


@responses.activate
def test_refresh_tokens_expired_clears_tokens(codex_dir: Path) -> None:
    _write_auth(
        codex_dir,
        {
            "auth_mode": "Chatgpt",
            "tokens": {
                "id_token": "x",
                "access_token": "y",
                "refresh_token": "old",
                "account_id": "z",
            },
            "last_refresh": "2025-01-01T00:00:00+00:00",
        },
    )

    responses.add(
        responses.POST,
        auth.TOKEN_URL,
        json={"error": "refresh_token_expired"},
        status=400,
    )

    with pytest.raises(auth.AuthError):
        auth.refresh_tokens("old", force=True)

    on_disk = json.loads((codex_dir / "auth.json").read_text())
    assert "tokens" not in on_disk
    assert on_disk["auth_mode"] == "Chatgpt"


# ---------------------------------------------------------------------------
# _resolve_credentials — token still fresh
# ---------------------------------------------------------------------------


def test_resolve_credentials_returns_fresh_token(
    codex_dir: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    fresh = _make_jwt(3600)
    _write_auth(
        codex_dir,
        {
            "auth_mode": "Chatgpt",
            "tokens": {
                "id_token": _make_jwt(3600),
                "access_token": fresh,
                "refresh_token": "r",
                "account_id": "acct",
            },
            "last_refresh": "2025-01-01T00:00:00+00:00",
        },
    )

    def _boom(*_args, **_kwargs):
        raise AssertionError("should not refresh when token is fresh")

    monkeypatch.setattr(auth, "refresh_tokens", _boom)
    monkeypatch.setattr(auth, "interactive_login", _boom)

    assert auth.get_access_token() == fresh


# ---------------------------------------------------------------------------
# _resolve_credentials — token near expiry triggers refresh
# ---------------------------------------------------------------------------


def test_resolve_credentials_triggers_refresh_when_near_expiry(
    codex_dir: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    soon = _make_jwt(60)  # within 5min leeway → refresh
    _write_auth(
        codex_dir,
        {
            "auth_mode": "Chatgpt",
            "tokens": {
                "id_token": _make_jwt(60),
                "access_token": soon,
                "refresh_token": "rrr",
                "account_id": "acct",
            },
            "last_refresh": "2025-01-01T00:00:00+00:00",
        },
    )

    refreshed_creds = auth.Credentials(
        access_token="REFRESHED",
        refresh_token="rrr2",
        id_token=None,
        account_id="acct",
        last_refresh=auth._now_iso(),
        source=codex_dir / "auth.json",
    )

    called = {"n": 0}

    def fake_refresh(refresh_token: str, *, force: bool = False):
        called["n"] += 1
        assert refresh_token == "rrr"
        return refreshed_creds

    monkeypatch.setattr(auth, "refresh_tokens", fake_refresh)
    monkeypatch.setattr(
        auth,
        "interactive_login",
        lambda: pytest.fail("should not fall back to OAuth"),
    )

    assert auth.get_access_token() == "REFRESHED"
    assert called["n"] == 1


# ---------------------------------------------------------------------------
# PKCE helpers — light sanity check
# ---------------------------------------------------------------------------


def test_pkce_pair_uses_s256() -> None:
    import hashlib

    verifier, challenge = auth._make_pkce_pair()
    digest = hashlib.sha256(verifier.encode()).digest()
    expected = (
        base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
    )
    assert challenge == expected