feat(providers): Codex CLI provider — OpenAI via ModelProvider.OPENAI (#659)

* feat(providers): Codex CLI provider — OpenAI via ModelProvider.OPENAI

Mirrors the grok blueprint end to end: CodexCliProvider (RO ~/.codex
mount, ANTHROPIC_* blanked), an orchestrator-side codex_auth.py
refresher (JWT-exp staleness, atomic rewrite, lock-serialized single-use
rotation, --check backstop; the CLI's own in-process refresh write
no-ops on the RO mount by design — margins keep the orchestrator ahead
of the CLI's 5-minute window), config.toml rendering with required=true
gateway MCP servers, execpolicy deny rules (forbidden-only), per-role
--sandbox (developer=workspace-write, review/doc roles read-only),
codex exec --json with pinned ROBOCO_CODEX_CLI_MODEL (gpt-5.3-codex),
usage summed from typed turn.completed events priced via the real
4-bucket split, dedicated image + entrypoint, registry/park/finalize/
compose/release wiring. V1 excludes interactive intake/secretary.

Per adversarial review: migration 083 seeds the openai provider row
enabled=True (without it every routing path 404'd — the whole feature
was operationally dead code; grok needed the same seed in 039), the
panel picker gained the OpenAI catalog group it silently lacked, and
exit classification is structural — only stderr and error.message
fields from error events are sniffed (word-boundaried patterns, exact
auth phrases, bare 'login' dropped), so the model echoing on-topic
words can never false-park the provider fleet-wide, proven by a
benign-transcript test. Known open risk flagged, not claimed: whether
codex's workspace-write OS sandbox excludes /app is unverified, and no
hook mechanism exists to port the bash-guard defense-in-depth.

* fix(providers): containment barrier on usage.json reads (code scanning)

CodeQL flagged the codex usage read as path injection — correctly:
os.path.basename does not neutralize '..', and the upstream segment
validator isn't in CodeQL's taint model. The grok/codex reads collapse
into one _read_usage_json_contained helper that resolves the built path
and refuses anything outside the resolved usage root — a hostile id can
never escape regardless of upstream drift. Traversal + containment
regression tests added; a stray noqa in the test file replaced with a
named constant per repo rule.

* fix(providers): use realpath+startswith containment CodeQL recognizes

The is_relative_to() guard was a real barrier but not in CodeQL's
py/path-injection sanitizer model, so the alert persisted. Switch to
the canonical os.path.realpath + startswith(root + os.sep) form, which
CodeQL recognizes as a path-traversal barrier; behavior is identical
(refuse any candidate resolving outside the usage root).

* fix(providers): regexp-allowlist the usage-id segment (CodeQL barrier)

Neither is_relative_to nor realpath+startswith was recognized by
CodeQL's py/path-injection sanitizer model across the str->Path->open
flow. Sanitize the tainted component at the source instead: the id must
fullmatch a strict slug token ([A-Za-z0-9][A-Za-z0-9._-]*, no
separators, no '..'), which CodeQL recognizes as a path-injection
barrier; the realpath+startswith containment stays as defense-in-depth.

* fix(providers): standalone regexp guard so CodeQL recognizes the barrier

The sanitizer was one disjunct of a compound 'or' condition, which
CodeQL's guard analysis does not trace as a barrier. Split the regexp
fullmatch into its own single-condition guard (the redundant '..' check
is dropped — the required alphanumeric first char already excludes it).

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

tests/integration/test_llm_routing.py

@@ -52,7 +52,16 @@ async def llm_setup(
         enabled=True,
         base_url="https://ollama.example.com",
     )
-    db_session.add_all([anthropic, grok, ollama])
+    # Mirrors migration 083_seed_openai_provider's contract: enabled=True at
+    # seed time (no apply_mode="codex" write path exists to flip it later —
+    # see that migration's docstring).
+    openai = ProviderConfigTable(
+        name="openai-test",
+        type=ModelProvider.OPENAI,
+        enabled=True,
+        base_url="https://api.openai.com/v1",
+    )
+    db_session.add_all([anthropic, grok, ollama, openai])
     await db_session.flush()
     yield {"svc": ModelRoutingService(db_session)}
 
@@ -196,6 +205,18 @@ async def test_derive_mode_grok_when_only_grok_global(llm_setup: dict) -> None:
     assert await svc.derive_mode() == "grok"
 
 
+@pytest.mark.asyncio
+async def test_derive_mode_codex_when_only_openai_global(llm_setup: dict) -> None:
+    """A pure-OPENAI global assignment reports "codex", not the catch-all
+    "mix" — the read-only branch derive_mode gained alongside the seed fix."""
+    svc = llm_setup["svc"]
+    codex_model = _first_model_for_type(ModelProvider.OPENAI)
+    await svc.upsert_assignment(
+        scope=AssignmentScope.GLOBAL, scope_value=None, model_name=codex_model
+    )
+    assert await svc.derive_mode() == "codex"
+
+
 @pytest.mark.asyncio
 async def test_derive_mode_mix_with_per_agent(llm_setup: dict) -> None:
     svc = llm_setup["svc"]
@@ -393,6 +414,32 @@ async def test_resolve_for_agent_uses_global_assignment(
     assert route.model_name == model
 
 
+@pytest.mark.asyncio
+async def test_upsert_and_resolve_openai_assignment_roundtrip(
+    llm_setup: dict,
+) -> None:
+    """gpt-5.3-codex through upsert_assignment -> resolve_for_agent, against
+    the seeded OPENAI row (migration 083). Before that seed existed,
+    upsert_assignment's `_get_seeded_provider(ModelProvider.OPENAI)` lookup
+    raised NotFoundError the moment anyone tried this — this is the round
+    trip that would have caught it."""
+    svc = llm_setup["svc"]
+    codex_model = _first_model_for_type(ModelProvider.OPENAI)
+    row = await svc.upsert_assignment(
+        scope=AssignmentScope.AGENT_SLUG,
+        scope_value="be-dev-1",
+        model_name=codex_model,
+    )
+    assert row.model_name == codex_model
+
+    route = await svc.resolve_for_agent("be-dev-1")
+    assert route.provider_type == ModelProvider.OPENAI
+    assert route.model_name == codex_model
+    # The seeded row carries no stored token — Codex authenticates via the
+    # mounted ~/.codex subscription dir, not a decrypted provider token.
+    assert route.auth_token is None
+
+
 @pytest.mark.asyncio
 async def test_resolve_for_agent_uses_provider_token(llm_setup: dict) -> None:
     """When provider has auth_token_encrypted, it's decrypted (lines 345-346)."""
@@ -672,6 +719,23 @@ async def test_get_seeded_provider_unknown_raises(
         await svc._get_seeded_provider(ModelProvider.ANTHROPIC)
 
 
+@pytest.mark.asyncio
+async def test_upsert_openai_assignment_without_seed_raises_not_found(
+    db_session: AsyncSession,
+) -> None:
+    """The exact pre-fix failure: assigning a catalog model whose provider
+    type has no seeded `provider_configs` row raises NotFoundError out of
+    `upsert_assignment`. This is what migration `083_seed_openai_provider`
+    fixes — a bare session (no `llm_setup` fixture, so no OPENAI row) proves
+    the seed is load-bearing, not incidental."""
+    svc = ModelRoutingService(db_session)
+    codex_model = _first_model_for_type(ModelProvider.OPENAI)
+    with pytest.raises(NotFoundError):
+        await svc.upsert_assignment(
+            scope=AssignmentScope.GLOBAL, scope_value=None, model_name=codex_model
+        )
+
+
 def test_get_model_routing_service_factory(db_session: AsyncSession) -> None:
     """Factory wraps ModelRoutingService with the given session (line 369)."""
 

tests/integration/test_migration_083_seed_openai_provider.py

@@ -0,0 +1,162 @@
+"""Migration 083 tests — seed_openai_provider.
+
+Verifies the post-upgrade state and exercises the downgrade SQL ordering,
+mirroring ``test_migration_028_seed_self_hosted.py`` and
+``039_seed_grok_provider``'s own shape.
+
+NOT a real alembic round-trip — the suite builds the test DB via
+Base.metadata.create_all (see conftest). Migration 083's upgrade()/downgrade()
+bodies are reviewed here; the tests guard the resulting DB-level contract —
+in particular ``enabled=True`` at seed time, the one detail that diverges
+from GROK's own seed (see the migration's docstring for why: there is no
+``apply_mode="codex"`` write path to flip it later).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+import pytest
+from roboco.db.tables import ModelAssignmentTable, ProviderConfigTable
+from roboco.models.base import AssignmentScope, ModelProvider
+from sqlalchemy import text
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+_INSERT_SQL = text(
+    """
+    INSERT INTO provider_configs
+        (id, name, type, base_url, auth_token_encrypted, enabled, created_at)
+    VALUES
+        (
+            gen_random_uuid(),
+            'Codex (OpenAI)',
+            'openai',
+            'https://api.openai.com/v1',
+            NULL,
+            true,
+            now()
+        )
+    ON CONFLICT (name) DO NOTHING
+    """
+)
+
+
+@pytest.mark.asyncio
+async def test_migration_083_upgrade_insert_contract(
+    db_session: AsyncSession,
+) -> None:
+    """The upgrade INSERT SQL seeds the Codex row ENABLED (unlike GROK's
+    seed, which starts disabled) and is idempotent."""
+    # --- First run: the row should be inserted.
+    await db_session.execute(_INSERT_SQL)
+    await db_session.flush()
+
+    result = await db_session.execute(
+        text(
+            "SELECT name, type, enabled, base_url "
+            "FROM provider_configs "
+            "WHERE name = 'Codex (OpenAI)'"
+        )
+    )
+    rows = list(result)
+    assert len(rows) == 1
+    name, ptype, enabled, base_url = rows[0]
+    assert name == "Codex (OpenAI)"
+    assert ptype == "openai"
+    # The load-bearing assertion: enabled=True at seed time. Seeding False
+    # (GROK's convention) would leave resolve_for_agent silently falling back
+    # to Anthropic forever, since no apply_mode="codex" write path exists to
+    # flip it — the exact "unreachable" failure this migration fixes.
+    assert enabled is True
+    assert base_url == "https://api.openai.com/v1"
+
+    # --- Second run: ON CONFLICT DO NOTHING must not create a duplicate.
+    await db_session.execute(_INSERT_SQL)
+    await db_session.flush()
+
+    result = await db_session.execute(
+        text("SELECT id FROM provider_configs WHERE name = 'Codex (OpenAI)'")
+    )
+    assert len(list(result)) == 1, (
+        "Expected exactly one 'Codex (OpenAI)' row after two INSERT "
+        "executions; ON CONFLICT DO NOTHING must prevent duplicates."
+    )
+
+
+@pytest.mark.asyncio
+async def test_migration_083_downgrade_deletes_assignments_before_config(
+    db_session: AsyncSession,
+) -> None:
+    """Downgrade SQL deletes model_assignments before provider_configs.
+
+    A FK RESTRICT constraint on model_assignments.provider_config_id means
+    deleting provider_configs first would raise an IntegrityError.
+    """
+    suffix = uuid4().hex[:8]
+    openai = ProviderConfigTable(
+        name=f"Codex (OpenAI)-test-{suffix}",
+        type=ModelProvider.OPENAI,
+        enabled=True,
+    )
+    db_session.add(openai)
+    await db_session.flush()
+
+    assignment = ModelAssignmentTable(
+        scope=AssignmentScope.AGENT_SLUG,
+        scope_value=f"test-agent-{suffix}",
+        provider_config_id=openai.id,
+        model_name="gpt-5.3-codex",
+    )
+    db_session.add(assignment)
+    await db_session.flush()
+
+    result = await db_session.execute(
+        text("SELECT id FROM provider_configs WHERE name = :name").bindparams(
+            name=openai.name
+        )
+    )
+    assert result.scalar_one_or_none() is not None
+
+    result = await db_session.execute(
+        text("SELECT id FROM model_assignments WHERE scope_value = :sv").bindparams(
+            sv=assignment.scope_value
+        )
+    )
+    assert result.scalar_one_or_none() is not None
+
+    # Step 1: delete referencing model_assignments first.
+    await db_session.execute(
+        text(
+            "DELETE FROM model_assignments "
+            "WHERE provider_config_id IN ("
+            "    SELECT id FROM provider_configs WHERE name = :name"
+            ")"
+        ).bindparams(name=openai.name)
+    )
+    # Step 2: now safe to delete the provider row.
+    await db_session.execute(
+        text("DELETE FROM provider_configs WHERE name = :name").bindparams(
+            name=openai.name
+        )
+    )
+
+    result = await db_session.execute(
+        text("SELECT id FROM provider_configs WHERE name = :name").bindparams(
+            name=openai.name
+        )
+    )
+    assert result.scalar_one_or_none() is None, (
+        "provider_configs row should be deleted by downgrade"
+    )
+
+    result = await db_session.execute(
+        text("SELECT id FROM model_assignments WHERE scope_value = :sv").bindparams(
+            sv=assignment.scope_value
+        )
+    )
+    assert result.scalar_one_or_none() is None, (
+        "model_assignments row should be deleted before provider_configs"
+    )

tests/unit/billing/test_pricing.py

@@ -61,6 +61,13 @@ _GROK_OUTPUT = 2.00
 _GROK_CACHE_READ = 0.20
 _GROK_CACHE_WRITE = 1.00
 
+# OpenAI Codex — priced non-Anthropic (ChatGPT-subscription CLI, priced here
+# for cost attribution)
+_CODEX_INPUT = 1.75
+_CODEX_OUTPUT = 14.00
+_CODEX_CACHE_READ = 0.175
+_CODEX_CACHE_WRITE = 1.75
+
 # Tolerance for floating-point comparisons
 _TOL = 1e-4
 
@@ -330,6 +337,55 @@ class TestGrokTier:
         assert calculate_cost("grok-build-0.1", tokens_input=_M, tokens_output=0) > 0.0
 
 
+# ---------------------------------------------------------------------------
+# Codex tier (OpenAI — priced non-Anthropic)
+# ---------------------------------------------------------------------------
+
+
+class TestCodexTier:
+    """gpt-5.3-codex pricing — a real input/output split, unlike grok's fold."""
+
+    def test_input_only(self) -> None:
+        cost = calculate_cost("gpt-5.3-codex", tokens_input=_M, tokens_output=0)
+        assert abs(cost - _CODEX_INPUT) < _TOL
+
+    def test_output_only(self) -> None:
+        cost = calculate_cost("gpt-5.3-codex", tokens_input=0, tokens_output=_M)
+        assert abs(cost - _CODEX_OUTPUT) < _TOL
+
+    def test_cached_input(self) -> None:
+        cost = calculate_cost(
+            "gpt-5.3-codex", tokens_input=0, tokens_output=0, tokens_cache_read=_M
+        )
+        assert abs(cost - _CODEX_CACHE_READ) < _TOL
+
+    def test_cache_write(self) -> None:
+        cost = calculate_cost(
+            "gpt-5.3-codex", tokens_input=0, tokens_output=0, tokens_cache_write=_M
+        )
+        assert abs(cost - _CODEX_CACHE_WRITE) < _TOL
+
+    def test_all_token_types(self) -> None:
+        cost = calculate_cost(
+            "gpt-5.3-codex",
+            tokens_input=_M,
+            tokens_output=_M,
+            tokens_cache_read=_M,
+            tokens_cache_write=_M,
+        )
+        expected = _CODEX_INPUT + _CODEX_OUTPUT + _CODEX_CACHE_READ + _CODEX_CACHE_WRITE
+        assert abs(cost - expected) < _TOL
+
+    def test_codex_is_not_treated_as_anthropic(self) -> None:
+        assert _is_anthropic_model("gpt-5.3-codex") is False
+        assert calculate_cost("gpt-5.3-codex", tokens_input=_M, tokens_output=0) > 0.0
+
+    def test_output_is_pricier_than_input(self) -> None:
+        # Codex's real split makes output 8x input — the property grok's
+        # single-total fold structurally cannot express.
+        assert _CODEX_OUTPUT > _CODEX_INPUT
+
+
 # ---------------------------------------------------------------------------
 # Unknown / edge cases — must return 0.0 without raising
 # ---------------------------------------------------------------------------
@@ -489,6 +545,14 @@ class TestCostResult:
         assert result.unpriced is False
         assert result.is_anthropic is False
 
+    def test_priced_non_anthropic_codex_is_not_unpriced(self) -> None:
+        result = calculate_cost_result(
+            "gpt-5.3-codex", tokens_input=_M, tokens_output=0
+        )
+        assert result.cost_usd > 0.0
+        assert result.unpriced is False
+        assert result.is_anthropic is False
+
     def test_calculate_cost_matches_structured_cost_usd(self) -> None:
         model = "claude-opus-4-6"
         assert (

tests/unit/llm/providers/test_codex_auth.py

@@ -0,0 +1,230 @@
+"""codex_auth — keep the Codex CLI credential live via the OAuth refresh grant.
+
+Unlike grok's bundle (keyed by ``<issuer>::<client_id>``, carrying its own
+``expires_at``), the Codex auth.json is flat — ``{tokens: {access_token,
+refresh_token, ...}}`` — and staleness is decided purely by decoding the
+access token's JWT ``exp`` claim.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import pathlib
+import threading
+import time
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING, Any
+
+from roboco.llm.providers import codex_auth as ca
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    import pytest
+
+
+def _jwt(exp_unix: int) -> str:
+    """Build a minimal JWT (header.payload.signature) carrying an ``exp`` claim."""
+    payload = (
+        base64.urlsafe_b64encode(json.dumps({"exp": exp_unix}).encode())
+        .rstrip(b"=")
+        .decode()
+    )
+    header = base64.urlsafe_b64encode(b'{"alg":"RS256"}').rstrip(b"=").decode()
+    return f"{header}.{payload}.sig"
+
+
+def _bundle(access_token: str, *, refresh_token: str = "rt") -> dict[str, Any]:
+    return {
+        "auth_mode": "chatgpt",
+        "tokens": {
+            "id_token": "id-tok",
+            "access_token": access_token,
+            "refresh_token": refresh_token,
+            "account_id": "acct-1",
+        },
+        "last_refresh": "2026-01-01T00:00:00Z",
+    }
+
+
+def _write(path: Path, bundle: dict[str, Any]) -> None:
+    path.write_text(json.dumps(bundle), encoding="utf-8")
+
+
+def _exp(delta: timedelta) -> int:
+    return int((datetime.now(UTC) + delta).timestamp())
+
+
+def test_seconds_until_expiry_and_is_valid(tmp_path: Path) -> None:
+    path = tmp_path / "auth.json"
+    token = _jwt(_exp(timedelta(minutes=54)))
+    _write(path, _bundle(token))
+    remaining = ca.seconds_until_expiry(path)
+    assert remaining is not None
+    assert 3100 < remaining < 3300  # noqa: PLR2004 — ~54 minutes
+    assert ca.is_valid(path)
+    assert not ca.is_valid(path, skew_seconds=3600)  # <1h left, 1h skew fails
+
+
+def test_seconds_until_expiry_none_for_missing_or_entryless(tmp_path: Path) -> None:
+    assert ca.seconds_until_expiry(tmp_path / "nope.json") is None
+    path = tmp_path / "auth.json"
+    _write(path, {"tokens": {"account_id": "x"}})  # no access_token
+    assert ca.seconds_until_expiry(path) is None
+
+
+def test_seconds_until_expiry_none_for_non_jwt_access_token(tmp_path: Path) -> None:
+    path = tmp_path / "auth.json"
+    _write(path, _bundle("api-key-not-a-jwt"))
+    assert ca.seconds_until_expiry(path) is None
+
+
+def test_refresh_skips_when_fresh(tmp_path: Path) -> None:
+    path = tmp_path / "auth.json"
+    _write(path, _bundle(_jwt(_exp(timedelta(hours=6)))))
+    calls: list[str] = []
+
+    def _post(url: str, _form: dict[str, str]) -> dict[str, Any]:
+        calls.append(url)
+        return {}
+
+    assert ca.refresh_if_stale(path, post=_post) == "fresh"
+    assert not calls  # no network call when the token is still valid
+
+
+def test_refresh_mints_new_token_when_stale(tmp_path: Path) -> None:
+    path = tmp_path / "auth.json"
+    _write(path, _bundle(_jwt(_exp(timedelta(minutes=-5)))))
+    new_access = _jwt(_exp(timedelta(hours=6)))
+
+    def _post(url: str, form: dict[str, str]) -> dict[str, Any]:
+        assert url == "https://auth.openai.com/oauth/token"
+        assert form["grant_type"] == "refresh_token"
+        assert form["refresh_token"] == "rt"
+        assert form["client_id"]
+        return {
+            "access_token": new_access,
+            "refresh_token": "new-rt",
+            "id_token": "new-id",
+        }
+
+    assert ca.refresh_if_stale(path, post=_post) == "refreshed"
+    bundle = json.loads(path.read_text())
+    assert bundle["tokens"]["access_token"] == new_access
+    assert bundle["tokens"]["refresh_token"] == "new-rt"  # rotated
+    assert bundle["tokens"]["id_token"] == "new-id"
+    assert bundle["last_refresh"] != "2026-01-01T00:00:00Z"
+    assert ca.is_valid(path)
+
+
+def test_refresh_keeps_old_refresh_token_when_response_omits_it(
+    tmp_path: Path,
+) -> None:
+    path = tmp_path / "auth.json"
+    _write(path, _bundle(_jwt(_exp(timedelta(minutes=-1)))))
+
+    def _post(_url: str, _form: dict[str, str]) -> dict[str, Any]:
+        return {"access_token": _jwt(_exp(timedelta(hours=6)))}  # no refresh_token
+
+    assert ca.refresh_if_stale(path, post=_post) == "refreshed"
+    assert json.loads(path.read_text())["tokens"]["refresh_token"] == "rt"
+
+
+def test_refresh_missing_file(tmp_path: Path) -> None:
+    assert ca.refresh_if_stale(tmp_path / "nope.json") == "missing"
+
+
+def test_refresh_no_refresh_token_api_key_mode(tmp_path: Path) -> None:
+    # auth_mode=apikey has no tokens/refresh_token — refresh is a graceful no-op.
+    path = tmp_path / "auth.json"
+    _write(path, {"auth_mode": "apikey", "OPENAI_API_KEY": "sk-x"})
+    assert ca.refresh_if_stale(path) == "no_refresh_token"
+
+
+def test_refresh_failed_on_post_error_leaves_file_untouched(tmp_path: Path) -> None:
+    path = tmp_path / "auth.json"
+    stale = _jwt(_exp(timedelta(minutes=-1)))
+    _write(path, _bundle(stale))
+
+    def _boom(_url: str, _form: dict[str, str]) -> dict[str, Any]:
+        raise RuntimeError("network down")
+
+    assert ca.refresh_if_stale(path, post=_boom) == "failed"
+    assert json.loads(path.read_text())["tokens"]["access_token"] == stale
+
+
+def test_refresh_failed_when_no_access_token(tmp_path: Path) -> None:
+    path = tmp_path / "auth.json"
+    _write(path, _bundle(_jwt(_exp(timedelta(minutes=-1)))))
+    assert ca.refresh_if_stale(path, post=lambda _u, _f: {}) == "failed"
+
+
+def test_refresh_persists_rotated_token_when_atomic_write_fails(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A rotated refresh_token is single-use; if the atomic write fails after
+    rotation, the direct-write fallback must still land it on disk."""
+    path = tmp_path / "auth.json"
+    _write(path, _bundle(_jwt(_exp(timedelta(minutes=-1)))))
+    new_access = _jwt(_exp(timedelta(hours=6)))
+
+    def _post(_url: str, _form: dict[str, str]) -> dict[str, Any]:
+        return {"access_token": new_access, "refresh_token": "rotated-rt"}
+
+    def _boom_replace(_self: pathlib.Path, _target: pathlib.Path) -> pathlib.Path:
+        raise OSError("replace failed (simulated)")
+
+    monkeypatch.setattr(pathlib.Path, "replace", _boom_replace)
+
+    assert ca.refresh_if_stale(path, post=_post) == "refreshed"
+    tokens = json.loads(path.read_text())["tokens"]
+    assert tokens["refresh_token"] == "rotated-rt"
+    assert tokens["access_token"] == new_access
+
+
+def test_main_check_exit_codes(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    home = tmp_path / ".codex"
+    home.mkdir()
+    monkeypatch.setenv("HOME", str(tmp_path))
+    _write(home / "auth.json", _bundle(_jwt(_exp(timedelta(hours=6)))))
+    assert ca.main(["--check"]) == 0
+    _write(home / "auth.json", _bundle(_jwt(_exp(timedelta(minutes=-1)))))
+    assert ca.main(["--check"]) == 1
+
+
+def test_concurrent_refresh_does_not_double_rotate_single_use_token(
+    tmp_path: Path,
+) -> None:
+    """Two near-simultaneous ``refresh_if_stale`` calls must POST the
+    refresh-token grant ONCE — a process-wide lock + re-load inside it makes
+    the loser see the winner's refreshed token and return "fresh" instead of
+    re-rotating (mirrors grok_auth's #94 fix)."""
+    path = tmp_path / "auth.json"
+    _write(path, _bundle(_jwt(_exp(timedelta(minutes=-1)))))
+    posted: list[str] = []
+    post_lock = threading.Lock()
+
+    def _post(_url: str, form: dict[str, str]) -> dict[str, Any]:
+        with post_lock:
+            posted.append(form["refresh_token"])
+        time.sleep(0.1)
+        return {
+            "access_token": _jwt(_exp(timedelta(hours=6))),
+            "refresh_token": "new-rt",
+        }
+
+    results: list[str] = []
+
+    def _run() -> None:
+        results.append(ca.refresh_if_stale(path, post=_post))
+
+    threads = [threading.Thread(target=_run) for _ in range(2)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert len(posted) == 1  # exactly one grant POST — no double rotation
+    assert all(r in ("refreshed", "fresh") for r in results)
+    assert "refreshed" in results

tests/unit/llm/providers/test_codex_cli_config.py

@@ -0,0 +1,120 @@
+"""codex_cli_config — mcp-config → config.toml + execpolicy rules + combined
+prompt + per-role sandbox flag."""
+
+from __future__ import annotations
+
+import tomllib
+from typing import TYPE_CHECKING
+
+from roboco.llm.providers import codex_cli_config as cc
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+_SAMPLE_MCP = {
+    "mcpServers": {
+        "roboco-flow": {
+            "command": "uv",
+            "args": ["run", "--no-sync", "python", "-m", "roboco.mcp.flow_server"],
+            "env": {"ROBOCO_AGENT_ID": "be-dev-1", "ROBOCO_AGENT_TOKEN": "tok-123"},
+        },
+        "roboco-do": {"command": "uv", "args": ["run", "x"]},
+        "roboco-optimal": {"command": "uv", "args": ["run", "y"]},
+    }
+}
+
+
+def test_render_config_toml_is_valid_toml_and_injects_env() -> None:
+    parsed = tomllib.loads(cc.render_config_toml(_SAMPLE_MCP))
+    flow = parsed["mcp_servers"]["roboco-flow"]
+    assert flow["command"] == "uv"
+    assert flow["args"][:2] == ["run", "--no-sync"]
+    assert flow["env"]["ROBOCO_AGENT_TOKEN"] == "tok-123"
+    assert "env" not in parsed["mcp_servers"]["roboco-do"]
+
+
+def test_render_config_toml_marks_gateway_pair_required() -> None:
+    parsed = tomllib.loads(cc.render_config_toml(_SAMPLE_MCP))
+    assert parsed["mcp_servers"]["roboco-flow"]["required"] is True
+    assert parsed["mcp_servers"]["roboco-do"]["required"] is True
+    # Every other server is best-effort — no `required` key at all.
+    assert "required" not in parsed["mcp_servers"]["roboco-optimal"]
+
+
+def test_render_config_toml_empty_when_no_servers() -> None:
+    assert cc.render_config_toml({}) == ""
+    assert cc.render_config_toml({"mcpServers": {}}) == ""
+
+
+def test_sandbox_level_developer_is_workspace_write() -> None:
+    assert cc.sandbox_level_for_role("developer") == "workspace-write"
+
+
+def test_sandbox_level_other_delivery_roles_are_read_only() -> None:
+    # Narrower than grok's per-role allows_write (documenter also writes there)
+    # — Codex V1 restricts local sandbox writes to developer only; documenter's
+    # real writes ride the roboco-docs MCP server, not a local file edit.
+    for role in ("qa", "documenter", "pr_reviewer", "cell_pm", "main_pm", ""):
+        assert cc.sandbox_level_for_role(role) == "read-only"
+
+
+def test_codex_cli_args_for_role_carries_sandbox_and_skip_git_check() -> None:
+    dev_args = cc.codex_cli_args_for_role("developer")
+    assert dev_args == ["--sandbox", "workspace-write", "--skip-git-repo-check"]
+    qa_args = cc.codex_cli_args_for_role("qa")
+    assert qa_args == ["--sandbox", "read-only", "--skip-git-repo-check"]
+
+
+def test_render_execpolicy_rules_covers_git_mutation_destructive_and_raw_pm() -> None:
+    rules = cc.render_execpolicy_rules()
+    assert 'prefix_rule(pattern = ["git", "push"], decision = "forbidden")' in rules
+    assert 'prefix_rule(pattern = ["git", "tag", "-d"], decision = "forbidden")' in (
+        rules
+    )
+    assert 'prefix_rule(pattern = ["rm", "-rf"], decision = "forbidden")' in rules
+    assert 'prefix_rule(pattern = ["uv", "run"], decision = "forbidden")' in rules
+    assert 'prefix_rule(pattern = ["pip", "install"], decision = "forbidden")' in rules
+    # Only allow/forbidden decisions — never `prompt` (blocks headless turns).
+    assert "prompt" not in rules
+
+
+def test_write_execpolicy_rules_writes_to_dest(tmp_path: Path) -> None:
+    dest = tmp_path / "rules" / "default.rules"
+    cc.write_execpolicy_rules(dest=dest)
+    assert dest.exists()
+    assert "git" in dest.read_text(encoding="utf-8")
+
+
+def test_render_combined_prompt_joins_system_and_task() -> None:
+    combined = cc.render_combined_prompt("You are the developer.", "Fix the bug.")
+    assert combined.startswith("You are the developer.")
+    assert combined.endswith("Fix the bug.")
+    assert "---" in combined
+
+
+def test_render_combined_prompt_degrades_gracefully() -> None:
+    assert cc.render_combined_prompt("", "task only") == "task only"
+    assert cc.render_combined_prompt("system only", "") == "system only"
+    assert cc.render_combined_prompt("", "") == ""
+
+
+def test_write_combined_prompt_reads_source_and_writes_dest(tmp_path: Path) -> None:
+    src = tmp_path / "system-prompt.md"
+    src.write_text("You are the RoboCo developer.", encoding="utf-8")
+    dest = tmp_path / "prompt.txt"
+    found = cc.write_combined_prompt(
+        task_prompt="Implement the feature.", source=src, dest=dest
+    )
+    assert found is True
+    text = dest.read_text(encoding="utf-8")
+    assert "You are the RoboCo developer." in text
+    assert "Implement the feature." in text
+
+
+def test_write_combined_prompt_degrades_when_source_absent(tmp_path: Path) -> None:
+    dest = tmp_path / "prompt.txt"
+    found = cc.write_combined_prompt(
+        task_prompt="Implement the feature.", source=tmp_path / "absent.md", dest=dest
+    )
+    assert found is False
+    assert dest.read_text(encoding="utf-8") == "Implement the feature."

tests/unit/llm/providers/test_codex_cli_sniff.py

@@ -0,0 +1,188 @@
+"""codex_cli_sniff — classify a Codex run from ONLY its machine-relevant text.
+
+The structural guarantee under test: the model's own on-topic prose (which
+can legitimately contain the words "quota-limited", "login page", or a "429"
+substring inside a commit hash / id) must NEVER reach the classifier, because
+extraction only pulls ``error.message`` fields off error-bearing JSONL events
+plus raw stderr — never ``turn.completed`` / ``item.*`` content.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+from roboco.llm.providers import codex_cli_sniff as sniff
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    import pytest
+
+
+def _write_jsonl(path: Path, lines: list[str]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def _turn_failed(message: str) -> str:
+    return json.dumps({"type": "turn.failed", "error": {"message": message}})
+
+
+# ---------------------------------------------------------------------------
+# extract_error_text — structural isolation
+# ---------------------------------------------------------------------------
+
+
+def test_extract_error_text_pulls_only_error_message(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(
+        log,
+        [
+            json.dumps(
+                {
+                    "type": "turn.completed",
+                    "usage": {"input_tokens": 1},
+                    "text": "the quota-limited rollout ships this sprint",
+                }
+            ),
+            _turn_failed("real error text"),
+        ],
+    )
+    assert sniff.extract_error_text(log) == "real error text"
+
+
+def test_extract_error_text_empty_for_missing_or_error_less_log(
+    tmp_path: Path,
+) -> None:
+    assert sniff.extract_error_text(tmp_path / "nope.jsonl") == ""
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [json.dumps({"type": "turn.completed", "usage": {}})])
+    assert sniff.extract_error_text(log) == ""
+
+
+# ---------------------------------------------------------------------------
+# The false-positive class this fix exists to kill
+# ---------------------------------------------------------------------------
+
+
+def test_benign_transcript_never_false_parks(tmp_path: Path) -> None:
+    """A transcript whose ONLY content is benign on-topic prose — mentioning
+    "quota-limited" work, a "login page" bug, and a commit hash containing
+    "429" — must classify as "" (no park), because none of it lives in an
+    error field the extractor even looks at."""
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(
+        log,
+        [
+            json.dumps(
+                {
+                    "type": "turn.completed",
+                    "usage": {"input_tokens": 10, "output_tokens": 5},
+                }
+            ),
+            json.dumps(
+                {
+                    "type": "item.completed",
+                    "item": {
+                        "type": "agent_message",
+                        "text": (
+                            "Fixed the quota-limited rollout gate and the "
+                            "login page redirect bug. Committed as abc4291f."
+                        ),
+                    },
+                }
+            ),
+        ],
+    )
+    err_log = tmp_path / "run.err"
+    err_log.write_text("", encoding="utf-8")
+    assert sniff.classify(log, err_log) == ""
+
+
+def test_word_boundary_prevents_429_substring_false_positive() -> None:
+    # "429" embedded inside a larger digit/word run must not match — grok's
+    # own \b429\b pattern, restored here after an initial cut dropped it.
+    assert not sniff.is_rate_limited("commit abc14293 deployed to prod")
+    assert not sniff.is_rate_limited("fix4297abc landed")
+    # "quota" alone legitimately matches wherever it appears (grok's own
+    # pattern, unchanged) — the false-positive class this fix kills is SCOPE
+    # (which text gets scanned, i.e. never turn.completed/item.* content),
+    # not the word "quota" itself. See test_benign_transcript_never_false_parks.
+
+
+def test_bare_login_word_does_not_classify_as_auth() -> None:
+    # "login" was dropped from the auth pattern — a mention of a login PAGE
+    # (this repo's own panel) must not false-park the provider.
+    assert not sniff.is_auth_failure("please visit the login page to continue")
+    assert not sniff.is_auth_failure("login required")
+
+
+# ---------------------------------------------------------------------------
+# True positives — real machine-extracted error text
+# ---------------------------------------------------------------------------
+
+
+def test_real_429_error_message_classifies_rate_limit(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_turn_failed("Rate limit exceeded: 429 Too Many Requests")])
+    assert sniff.classify(log) == "rate_limit"
+
+
+def test_insufficient_quota_error_message_classifies_rate_limit(
+    tmp_path: Path,
+) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_turn_failed("insufficient_quota: billing hard limit hit")])
+    assert sniff.classify(log) == "rate_limit"
+
+
+def test_exact_auth_phrase_refresh_token_expired_classifies_auth(
+    tmp_path: Path,
+) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(
+        log, [_turn_failed("Your refresh token has expired, please re-authenticate")]
+    )
+    assert sniff.classify(log) == "auth"
+
+
+def test_exact_auth_phrase_not_signed_in_classifies_auth(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_turn_failed("Error: not signed in")])
+    assert sniff.classify(log) == "auth"
+
+
+def test_classify_reads_stderr_too(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [json.dumps({"type": "turn.completed", "usage": {}})])
+    err_log = tmp_path / "run.err"
+    err_log.write_text("fatal: 429 too many requests\n", encoding="utf-8")
+    assert sniff.classify(log, err_log) == "rate_limit"
+
+
+def test_classify_missing_files_returns_empty(tmp_path: Path) -> None:
+    assert sniff.classify(tmp_path / "nope.jsonl", tmp_path / "nope.err") == ""
+
+
+def test_rate_limit_checked_before_auth_when_both_present(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(
+        log,
+        [_turn_failed("429 too many requests, and also not signed in downstream")],
+    )
+    assert sniff.classify(log) == "rate_limit"
+
+
+def test_main_cli_prints_classification(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_turn_failed("429 too many requests")])
+    assert sniff.main([str(log)]) == 0
+    assert capsys.readouterr().out.strip() == "rate_limit"
+
+
+def test_main_cli_no_args_prints_empty(capsys: pytest.CaptureFixture[str]) -> None:
+    assert sniff.main([]) == 0
+    assert capsys.readouterr().out.strip() == ""

tests/unit/llm/providers/test_codex_cli_usage.py

@@ -0,0 +1,159 @@
+"""codex_cli_usage — sum real input/output/cache usage across ``turn.completed``
+events in a captured ``codex exec --json`` JSONL log."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+from roboco.llm.providers import codex_cli_usage as cu
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    import pytest
+
+
+def _turn_completed(
+    *,
+    input_tokens: int,
+    cached_input_tokens: int = 0,
+    cache_write_input_tokens: int = 0,
+    output_tokens: int,
+    reasoning_output_tokens: int = 0,
+) -> str:
+    return json.dumps(
+        {
+            "type": "turn.completed",
+            "usage": {
+                "input_tokens": input_tokens,
+                "cached_input_tokens": cached_input_tokens,
+                "cache_write_input_tokens": cache_write_input_tokens,
+                "output_tokens": output_tokens,
+                "reasoning_output_tokens": reasoning_output_tokens,
+            },
+        }
+    )
+
+
+def _write_jsonl(path: Path, lines: list[str]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def test_aggregate_sums_across_multiple_turn_completed_events(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(
+        log,
+        [
+            json.dumps({"type": "thread.started"}),
+            json.dumps({"type": "turn.started"}),
+            _turn_completed(
+                input_tokens=1000, cached_input_tokens=200, output_tokens=100
+            ),
+            json.dumps({"type": "item.completed", "item": {"type": "command"}}),
+            _turn_completed(
+                input_tokens=500,
+                cached_input_tokens=100,
+                cache_write_input_tokens=50,
+                output_tokens=80,
+                reasoning_output_tokens=20,
+            ),
+        ],
+    )
+    agg = cu.aggregate_usage_from_jsonl(log)
+    assert agg["input_tokens"] == 1500  # noqa: PLR2004
+    assert agg["cached_input_tokens"] == 300  # noqa: PLR2004
+    assert agg["cache_write_input_tokens"] == 50  # noqa: PLR2004
+    assert agg["output_tokens"] == 180  # noqa: PLR2004
+    assert agg["reasoning_output_tokens"] == 20  # noqa: PLR2004
+    assert agg["turns"] == 2  # noqa: PLR2004
+
+
+def test_aggregate_ignores_turn_failed_and_bad_lines(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(
+        log,
+        [
+            "not json",
+            json.dumps({"type": "turn.failed", "error": {"message": "boom"}}),
+            _turn_completed(input_tokens=10, output_tokens=5),
+        ],
+    )
+    agg = cu.aggregate_usage_from_jsonl(log)
+    assert agg["input_tokens"] == 10  # noqa: PLR2004
+    assert agg["turns"] == 1
+
+
+def test_aggregate_zero_for_missing_or_empty_log(tmp_path: Path) -> None:
+    agg = cu.aggregate_usage_from_jsonl(tmp_path / "nope.jsonl")
+    assert agg["turns"] == 0
+    assert all(v == 0 for k, v in agg.items() if k != "turns")
+
+
+def test_usage_and_cost_treats_cached_as_subset_of_input() -> None:
+    # cached_input_tokens is a SUBSET of input_tokens (not additional) — the
+    # "fresh" input priced at the full rate is the difference.
+    agg = {
+        "input_tokens": 1000,
+        "cached_input_tokens": 300,
+        "cache_write_input_tokens": 0,
+        "output_tokens": 200,
+        "reasoning_output_tokens": 50,
+    }
+    tin, tout, cr, cw, cost = cu.usage_and_cost("gpt-5.3-codex", agg)
+    assert tin == 700  # 1000 - 300  # noqa: PLR2004
+    assert tout == 250  # output + reasoning folded in  # noqa: PLR2004
+    assert cr == 300  # noqa: PLR2004
+    assert cw == 0
+    assert cost > 0.0
+
+
+def test_usage_and_cost_never_goes_negative_when_cached_exceeds_input() -> None:
+    agg = {
+        "input_tokens": 10,
+        "cached_input_tokens": 50,  # malformed/inconsistent upstream data
+        "cache_write_input_tokens": 0,
+        "output_tokens": 0,
+        "reasoning_output_tokens": 0,
+    }
+    tin, *_rest = cu.usage_and_cost("gpt-5.3-codex", agg)
+    assert tin == 0
+
+
+def test_capture_run_usage_writes_usage_json(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_turn_completed(input_tokens=100, output_tokens=50)])
+    out = tmp_path / "usage.json"
+    tokens = cu.capture_run_usage(run_log=log, model="gpt-5.3-codex", out_path=out)
+    assert tokens == (100, 50, 0, 0)
+    data = json.loads(out.read_text())
+    assert data["model"] == "gpt-5.3-codex"
+    assert data["tokens_input"] == 100  # noqa: PLR2004
+    assert data["tokens_output"] == 50  # noqa: PLR2004
+    assert data["turns"] == 1
+    assert data["cost_usd"] > 0.0
+
+
+def test_main_writes_usage_file(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_turn_completed(input_tokens=200, output_tokens=100)])
+    out = tmp_path / "usage.json"
+    monkeypatch.setattr(cu, "USAGE_OUT_PATH", out)
+    monkeypatch.setenv("ROBOCO_CODEX_RUN_LOG", str(log))
+    monkeypatch.setenv("ROBOCO_AGENT_MODEL", "gpt-5.3-codex")
+    assert cu.main() == 0
+    data = json.loads(out.read_text())
+    assert data["tokens_input"] == 200  # noqa: PLR2004
+    assert data["tokens_output"] == 100  # noqa: PLR2004
+
+
+def test_main_warns_when_run_log_env_missing(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    monkeypatch.delenv("ROBOCO_CODEX_RUN_LOG", raising=False)
+    with caplog.at_level("WARNING", logger="roboco.llm.providers.codex_cli_usage"):
+        assert cu.main() == 0
+    assert any("ROBOCO_CODEX_RUN_LOG" in r.message for r in caplog.records)

tests/unit/llm/test_providers.py

@@ -19,6 +19,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 from roboco.llm.providers import (
     ClaudeCodeProvider,
+    CodexCliProvider,
     GrokCliProvider,
     ProviderError,
     ProviderNotRegisteredError,
@@ -37,6 +38,16 @@ def _isolate_grok_auth(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
     return tmp_path
 
 
+@pytest.fixture(autouse=True)
+def _isolate_codex_auth(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """Point CODEX_AUTH_HOST_PATH at a fresh tmp dir (parity with grok above)."""
+    codex_dir = tmp_path / "codex-auth"
+    monkeypatch.setattr(
+        "roboco.llm.providers.codex.CODEX_AUTH_HOST_PATH", str(codex_dir)
+    )
+    return codex_dir
+
+
 def _config(
     *,
     agent_id: str = "be-dev-1",
@@ -85,6 +96,9 @@ class _FakeHost:
     def _ensure_grok_usage_dir(self, agent_id: str) -> None:
         self.data_dirs_ensured.append(agent_id)
 
+    def _ensure_codex_usage_dir(self, agent_id: str) -> None:
+        self.data_dirs_ensured.append(agent_id)
+
     def _resolve_host_paths(
         self, config: OrchestratorAgentConfig, agent_settings_path: Path | None
     ) -> dict[str, str | None]:
@@ -94,6 +108,7 @@ class _FakeHost:
             else None,
             "settings": str(agent_settings_path) if agent_settings_path else None,
             "grok_usage": f"/host/data/grok-usage/{config.agent_id}",
+            "codex_usage": f"/host/data/codex-usage/{config.agent_id}",
         }
 
     def _build_mount_args(
@@ -312,6 +327,156 @@ async def test_grok_spawn_raises_on_docker_failure() -> None:
         await provider.spawn(_config())
 
 
+# ---------------------------------------------------------------------------
+# CodexCliProvider
+# ---------------------------------------------------------------------------
+
+
+def _codex_config(
+    *,
+    agent_id: str = "be-dev-1",
+    provider_base_url: str | None = "https://api.x.ai/v1",
+    provider_auth_token: str | None = "should-not-leak",
+    mcp_config_path: Path | None = Path("/host/mcp-configs/be-dev-1.json"),
+) -> OrchestratorAgentConfig:
+    return OrchestratorAgentConfig(
+        agent_id=agent_id,
+        blueprint_path=Path("/app/system-prompt.md"),
+        model="gpt-5.3-codex",
+        mcp_config_path=mcp_config_path,
+        claude_session_id="sess-1",
+        provider_type="openai",
+        provider_base_url=provider_base_url,
+        provider_auth_token=provider_auth_token,
+    )
+
+
+async def test_codex_spawn_requires_mcp_config() -> None:
+    provider = CodexCliProvider(_FakeHost())
+    with pytest.raises(ProviderError, match="MCP config"):
+        await provider.spawn(_codex_config(mcp_config_path=None))
+
+
+async def test_codex_spawn_does_not_require_api_key() -> None:
+    # Subscription auth (mounted ~/.codex) — a missing provider key is fine.
+    host = _FakeHost()
+    provider = CodexCliProvider(host)
+    with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())):
+        result = await provider.spawn(_codex_config(provider_auth_token=None))
+    assert result.instance_id == "roboco-agent-be-dev-1"
+
+
+async def test_codex_spawn_no_leaked_key_and_no_anthropic_leak() -> None:
+    host = _FakeHost()
+    provider = CodexCliProvider(host, image="roboco-agent-codex:test")
+    with patch(
+        "asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
+    ) as exec_mock:
+        await provider.spawn(_codex_config(), initial_prompt="do the work")
+    cmd = list(exec_mock.call_args.args)
+    assert not any(c.startswith("OPENAI_API_KEY=") for c in cmd)
+    # The provider endpoint must NOT be injected as an Anthropic var.
+    assert not any(c.startswith("ANTHROPIC_BASE_URL=") for c in cmd)
+    assert not any(c.startswith("ANTHROPIC_AUTH_TOKEN=") for c in cmd)
+    assert host.mount_config is not None
+    assert host.mount_config.provider_base_url is None
+    assert host.mount_config.provider_auth_token is None
+
+
+async def test_codex_spawn_wires_gateway_env_and_image_last() -> None:
+    host = _FakeHost()
+    provider = CodexCliProvider(host, image="roboco-agent-codex:test")
+    with patch(
+        "asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
+    ) as exec_mock:
+        result = await provider.spawn(_codex_config())
+    cmd = list(exec_mock.call_args.args)
+    assert "ROBOCO_MCP_CONFIG=/app/mcp-config.json" in cmd
+    assert "ROBOCO_AGENT_ID=be-dev-1" in cmd
+    assert "ROBOCO_AGENT_MODEL=gpt-5.3-codex" in cmd
+    # Usage capture: per-agent data dir mounted + the entrypoint's usage file.
+    assert host.data_dirs_ensured == ["be-dev-1"]
+    assert "/host/data/codex-usage/be-dev-1:/home/agent/.codex-usage" in cmd
+    assert "ROBOCO_CODEX_USAGE_FILE=/home/agent/.codex-usage/usage.json" in cmd
+    assert "ROBOCO_AGENT_TOKEN=hmac-be-dev-1" in cmd
+    assert cmd[-1] == "roboco-agent-codex:test"
+    assert host.removed == ["roboco-agent-be-dev-1"]
+    assert host.remove_stop_reasons == ["pre_spawn_stale_clear"]
+    assert result == SpawnResult(
+        instance_id="roboco-agent-be-dev-1",
+        extra={"container_id": "cid", "model": "gpt-5.3-codex"},
+    )
+
+
+async def test_codex_spawn_mounts_auth_when_present(
+    _isolate_codex_auth: Path,
+) -> None:
+    _isolate_codex_auth.mkdir(parents=True, exist_ok=True)
+    (_isolate_codex_auth / "auth.json").write_text("{}", encoding="utf-8")
+    host = _FakeHost()
+    provider = CodexCliProvider(host)
+    with patch(
+        "asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
+    ) as exec_mock:
+        await provider.spawn(_codex_config())
+    cmd = list(exec_mock.call_args.args)
+    # Mount the host ~/.codex DIRECTORY (ro), not the single auth.json file —
+    # a single-file bind mount pins the inode (same concern grok documents).
+    expected = f"{_isolate_codex_auth}:/home/agent/.codex-auth-ro:ro"
+    assert expected in cmd
+
+
+async def test_codex_spawn_omits_auth_mount_when_absent() -> None:
+    host = _FakeHost()
+    provider = CodexCliProvider(host)
+    with patch(
+        "asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
+    ) as exec_mock:
+        await provider.spawn(_codex_config())
+    cmd = list(exec_mock.call_args.args)
+    assert not any("/home/agent/.codex-auth-ro" in c for c in cmd)
+
+
+async def test_codex_spawn_warns_when_auth_absent(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    caplog.set_level("WARNING", logger="roboco.llm.providers.codex")
+    host = _FakeHost()
+    provider = CodexCliProvider(host)
+    with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())):
+        await provider.spawn(_codex_config())
+    warnings = [r for r in caplog.records if r.levelname == "WARNING"]
+    assert warnings, "expected a spawn-time WARNING for the missing host auth.json"
+    msg = warnings[0].getMessage()
+    assert "auth.json" in msg
+    assert "codex login" in msg
+
+
+async def test_codex_spawn_prompt_is_injection_safe() -> None:
+    host = _FakeHost()
+    provider = CodexCliProvider(host)
+    nasty = "--model evil --session-id pwned"
+    with patch(
+        "asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
+    ) as exec_mock:
+        await provider.spawn(_codex_config(), initial_prompt=nasty)
+    cmd = list(exec_mock.call_args.args)
+    assert f"ROBOCO_INITIAL_PROMPT={nasty}" in cmd
+    assert nasty not in cmd
+
+
+async def test_codex_spawn_raises_on_docker_failure() -> None:
+    provider = CodexCliProvider(_FakeHost())
+    with (
+        patch(
+            "asyncio.create_subprocess_exec",
+            AsyncMock(return_value=_proc(returncode=1, stderr=b"boom")),
+        ),
+        pytest.raises(ProviderError, match="boom"),
+    ):
+        await provider.spawn(_codex_config())
+
+
 # ---------------------------------------------------------------------------
 # ClaudeCodeProvider
 # ---------------------------------------------------------------------------

tests/unit/runtime/test_codex_rate_limit.py

@@ -0,0 +1,158 @@
+"""CODEX 429/auth parking: same exit-code convention as grok, scoped to
+ModelProvider.OPENAI so a numeric-code collision with another provider's crash
+can never mis-park (see ``_CODEX_RATE_LIMIT_EXIT_CODE`` / ``_CODEX_AUTH_EXIT_CODE``
+in ``roboco.runtime.orchestrator``).
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+from roboco.models.runtime import AgentInstance
+from roboco.runtime.orchestrator import (
+    _CODEX_AUTH_EXIT_CODE,
+    _CODEX_RATE_LIMIT_EXIT_CODE,
+    AgentOrchestrator,
+    AgentState,
+)
+
+
+def _codex_instance(provider_type: str = "openai") -> AgentInstance:
+    cfg = type("C", (), {"provider_type": provider_type, "model": "gpt-5.3-codex"})()
+    inst = AgentInstance(agent_id="be-dev-1", state=AgentState.ACTIVE, config=cfg)
+    inst.current_task_id = "task-1"
+    inst.container_id = "cid"
+    return inst
+
+
+class _FakeTracker:
+    def __init__(self) -> None:
+        self.activated_with: dict[str, object] | None = None
+
+    async def activate(
+        self,
+        *,
+        retry_after: float,
+        affected_agents: list[str],
+        kind: str = "rate_limited",
+    ) -> None:
+        self.activated_with = {
+            "retry_after": retry_after,
+            "affected_agents": affected_agents,
+            "kind": kind,
+        }
+
+
+def test_is_codex_rate_limit_exit() -> None:
+    inst = _codex_instance()
+    assert AgentOrchestrator._is_codex_rate_limit_exit(
+        inst, _CODEX_RATE_LIMIT_EXIT_CODE
+    )
+    assert not AgentOrchestrator._is_codex_rate_limit_exit(inst, 0)
+    assert not AgentOrchestrator._is_codex_rate_limit_exit(inst, 1)
+    # A grok exit at the SAME numeric code must NOT be classified as codex.
+    assert not AgentOrchestrator._is_codex_rate_limit_exit(
+        _codex_instance(provider_type="grok"), _CODEX_RATE_LIMIT_EXIT_CODE
+    )
+    assert not AgentOrchestrator._is_codex_rate_limit_exit(
+        _codex_instance(provider_type="anthropic"), _CODEX_RATE_LIMIT_EXIT_CODE
+    )
+
+
+def test_is_codex_auth_exit() -> None:
+    inst = _codex_instance()
+    assert AgentOrchestrator._is_codex_auth_exit(inst, _CODEX_AUTH_EXIT_CODE)
+    assert not AgentOrchestrator._is_codex_auth_exit(inst, 0)
+    assert not AgentOrchestrator._is_codex_auth_exit(inst, 1)
+    assert not AgentOrchestrator._is_codex_auth_exit(
+        _codex_instance(provider_type="grok"), _CODEX_AUTH_EXIT_CODE
+    )
+
+
+@pytest.mark.asyncio
+async def test_park_codex_rate_limited_activates_and_offlines(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    orch._waiting_records = {}
+    orch._rate_limit_ceo_notified = set()
+    inst = _codex_instance()
+    inst.error_count = 2  # pretend prior crashes — parking must NOT count one
+    tracker = _FakeTracker()
+    monkeypatch.setattr(orch, "_make_tracker", lambda _p: tracker)
+    finalize = AsyncMock()
+    monkeypatch.setattr(orch, "_finalize_spawn_session", finalize)
+    monkeypatch.setattr(orch, "_persist_waiting_record", AsyncMock())
+
+    await orch._park_codex_rate_limited("be-dev-1", inst)
+
+    finalize.assert_awaited_once()
+    assert inst.state == AgentState.OFFLINE
+    assert inst.container_id is None
+    assert inst.error_count == 0  # a 429 is not a crash
+    assert tracker.activated_with == {
+        "retry_after": pytest.approx(60.0),
+        "affected_agents": ["be-dev-1"],
+        "kind": "rate_limited",
+    }
+
+
+@pytest.mark.asyncio
+async def test_park_codex_auth_unavailable_activates_with_auth_missing_kind(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    orch._waiting_records = {}
+    orch._rate_limit_ceo_notified = set()
+    inst = _codex_instance()
+    inst.error_count = 2
+    tracker = _FakeTracker()
+    monkeypatch.setattr(orch, "_make_tracker", lambda _p: tracker)
+    monkeypatch.setattr(orch, "_finalize_spawn_session", AsyncMock())
+    monkeypatch.setattr(orch, "_persist_waiting_record", AsyncMock())
+
+    await orch._park_codex_auth_unavailable("be-dev-1", inst)
+
+    assert inst.state == AgentState.OFFLINE
+    assert inst.container_id is None
+    assert inst.error_count == 0
+    assert tracker.activated_with == {
+        "retry_after": pytest.approx(60.0),
+        "affected_agents": ["be-dev-1"],
+        "kind": "auth_missing",
+    }
+
+
+@pytest.mark.asyncio
+async def test_handle_stopped_container_parks_on_codex_429(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    inst = _codex_instance()
+    park = AsyncMock()
+    finalize = AsyncMock()
+    monkeypatch.setattr(orch, "_park_codex_rate_limited", park)
+    monkeypatch.setattr(orch, "_finalize_spawn_session", finalize)
+
+    await orch._handle_stopped_container("be-dev-1", inst, _CODEX_RATE_LIMIT_EXIT_CODE)
+
+    park.assert_awaited_once_with("be-dev-1", inst)
+    finalize.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_handle_stopped_container_parks_on_codex_auth_exit(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    inst = _codex_instance()
+    park = AsyncMock()
+    finalize = AsyncMock()
+    monkeypatch.setattr(orch, "_park_codex_auth_unavailable", park)
+    monkeypatch.setattr(orch, "_finalize_spawn_session", finalize)
+
+    await orch._handle_stopped_container("be-dev-1", inst, _CODEX_AUTH_EXIT_CODE)
+
+    park.assert_awaited_once_with("be-dev-1", inst)
+    finalize.assert_not_awaited()

tests/unit/runtime/test_codex_usage_finalize.py

@@ -0,0 +1,166 @@
+"""OPENAI (codex) agents capture real input/output/cache-split token usage
+from their captured ``usage.json`` — unlike grok's single cumulative total,
+codex's JSONL carries a genuine split (see ``codex_cli_usage``), so finalize
+must return the real 4-tuple instead of folding everything into output.
+"""
+
+from __future__ import annotations
+
+import json
+import tempfile
+from typing import TYPE_CHECKING
+
+import httpx
+import pytest
+from roboco.models.runtime import AgentInstance
+from roboco.runtime import orchestrator as orch_mod
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _write_usage(path: Path, **fields: object) -> None:
+    payload = {
+        "model": "gpt-5.3-codex",
+        "tokens_input": 0,
+        "tokens_output": 0,
+        "tokens_cache_read": 0,
+        "tokens_cache_write": 0,
+        "cost_usd": 0.0,
+        "turns": 1,
+        **fields,
+    }
+    path.write_text(
+        json.dumps(payload),
+        encoding="utf-8",
+    )
+
+
+def test_codex_usage_returns_real_split(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    usage = tmp_path / "usage.json"
+    _write_usage(
+        usage, tokens_input=700, tokens_output=250, tokens_cache_read=300, turns=2
+    )
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    monkeypatch.setattr(
+        orch, "_codex_usage_json", lambda _aid: json.loads(usage.read_text())
+    )
+
+    expected_turns = 2
+    assert orch._codex_usage_tokens("be-dev-1") == (700, 250, 300, 0)
+    assert orch._codex_usage_turns("be-dev-1") == expected_turns
+
+
+def test_read_usage_json_contained_refuses_escape(tmp_path: Path) -> None:
+    """A '..' id resolves outside the usage root and must be refused —
+    basename alone does not neutralize '..', the containment check does."""
+    (tmp_path / "usage.json").write_text('{"leak": 1}', encoding="utf-8")
+    base = tmp_path / "root"
+    base.mkdir()
+
+    assert AgentOrchestrator._read_usage_json_contained(base, "..") is None
+
+
+def test_read_usage_json_contained_reads_inside_root(tmp_path: Path) -> None:
+    agent_dir = tmp_path / "be-dev-1"
+    agent_dir.mkdir()
+    (agent_dir / "usage.json").write_text('{"total_tokens": 5}', encoding="utf-8")
+
+    data = AgentOrchestrator._read_usage_json_contained(tmp_path, "be-dev-1")
+    assert data == {"total_tokens": 5}
+
+
+def test_codex_usage_zero_when_store_missing(monkeypatch: pytest.MonkeyPatch) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    monkeypatch.setattr(orch, "_codex_usage_json", lambda _aid: None)
+    assert orch._codex_usage_tokens("be-dev-1") == (0, 0, 0, 0)
+    assert orch._codex_usage_turns("be-dev-1") == 0
+
+
+@pytest.mark.asyncio
+async def test_resolve_final_usage_routes_openai_to_usage_json(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    monkeypatch.setattr(
+        orch,
+        "_codex_usage_json",
+        lambda _aid: {
+            "tokens_input": 12,
+            "tokens_output": 34,
+            "tokens_cache_read": 5,
+            "tokens_cache_write": 1,
+        },
+    )
+    cfg = type("C", (), {"provider_type": "openai"})()
+    orch._instances = {"be-dev-1": AgentInstance(agent_id="be-dev-1", config=cfg)}
+
+    assert await orch._resolve_final_token_usage("be-dev-1") == (12, 34, 5, 1)
+
+
+@pytest.mark.asyncio
+async def test_resolve_final_turns_tools_routes_openai_to_usage_json(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    monkeypatch.setattr(orch, "_codex_usage_turns", lambda _aid: 3)
+    cfg = type("C", (), {"provider_type": "openai"})()
+    orch._instances = {"be-dev-1": AgentInstance(agent_id="be-dev-1", config=cfg)}
+
+    # Codex has no tool-call signal — tool_calls stays 0.
+    assert await orch._resolve_final_turns_tools("be-dev-1") == (3, 0)
+
+
+@pytest.mark.asyncio
+async def test_resolve_active_tokens_routes_openai_to_usage_json(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    monkeypatch.setattr(
+        orch,
+        "_codex_usage_json",
+        lambda _aid: {"tokens_input": 12, "tokens_output": 34},
+    )
+    cfg = type("C", (), {"provider_type": "openai"})()
+    orch._instances = {"be-dev-1": AgentInstance(agent_id="be-dev-1", config=cfg)}
+    async with httpx.AsyncClient() as client:
+        assert await orch._resolve_active_tokens(client, "be-dev-1") == (12, 34, 0, 0)
+
+
+def test_codex_usage_dir_branches_compose_vs_local(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(orch_mod, "PROJECT_HOST_PATH", "")
+    local = AgentOrchestrator._codex_usage_dir("be-dev-1")
+    assert "roboco-codex-usage" in str(local)
+    assert local.name == "be-dev-1"
+
+    monkeypatch.setattr(orch_mod, "PROJECT_HOST_PATH", "/volume1/roboco")
+    monkeypatch.setattr(orch_mod, "CODEX_USAGE_DATA_DIR", "/data/codex-usage")
+    assert str(AgentOrchestrator._codex_usage_dir("be-dev-1")) == (
+        "/data/codex-usage/be-dev-1"
+    )
+
+
+@pytest.mark.parametrize(
+    "bad",
+    ["..", ".", "../etc", "a/b", "a\\b", "", "be-dev-1/../x", "x\x00y"],
+)
+def test_codex_usage_dir_rejects_path_traversal(bad: str) -> None:
+    with pytest.raises(ValueError, match="unsafe agent id"):
+        AgentOrchestrator._codex_usage_dir(bad)
+
+
+def test_codex_usage_json_reads_the_real_local_dir(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(orch_mod, "PROJECT_HOST_PATH", "")
+    monkeypatch.setattr(tempfile, "gettempdir", lambda: str(tmp_path))
+    udir = tmp_path / "roboco-codex-usage" / "be-dev-1"
+    udir.mkdir(parents=True)
+    _write_usage(udir / "usage.json", tokens_input=55, tokens_output=10)
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    assert orch._codex_usage_tokens("be-dev-1") == (55, 10, 0, 0)