Implements LLMProvider by shelling out to ``codex exec``. Uses ``--json`` to get JSONL output on stdout, which includes ``item.completed`` events carrying the assistant response and ``turn.completed`` events with token usage.
| 14 | |
| 15 | |
| 16 | class CodexProvider: |
| 17 | """Implements LLMProvider by shelling out to ``codex exec``. |
| 18 | |
| 19 | Uses ``--json`` to get JSONL output on stdout, which includes |
| 20 | ``item.completed`` events carrying the assistant response and |
| 21 | ``turn.completed`` events with token usage. |
| 22 | """ |
| 23 | |
| 24 | def __init__( |
| 25 | self, |
| 26 | model: str, |
| 27 | *, |
| 28 | reasoning_effort: str | None = None, |
| 29 | codex_bin: str | Sequence[str] = "codex", |
| 30 | ) -> None: |
| 31 | # model format: "codex/<model>" (effort is passed separately) |
| 32 | parts = model.split("/", 1) |
| 33 | if len(parts) < 2 or not parts[1]: |
| 34 | raise ValueError( |
| 35 | f"codex provider requires a model name (e.g. codex/gpt-5.4-mini), got: {model!r}" |
| 36 | ) |
| 37 | self._model = model |
| 38 | self._underlying = parts[1] |
| 39 | self._reasoning_effort = reasoning_effort |
| 40 | self._codex_bin = [codex_bin] if isinstance(codex_bin, str) else list(codex_bin) |
| 41 | |
| 42 | def call(self, user_content: str) -> tuple[str, TokenUsage]: |
| 43 | prompt = SYSTEM_PROMPT + "\n\n" + user_content |
| 44 | |
| 45 | cmd = [ |
| 46 | *self._codex_bin, |
| 47 | "exec", |
| 48 | "--json", |
| 49 | "--sandbox", |
| 50 | "read-only", |
| 51 | ] |
| 52 | |
| 53 | cmd.extend(["--model", self._underlying]) |
| 54 | |
| 55 | if self._reasoning_effort: |
| 56 | cmd.extend(["-c", f'model_reasoning_effort="{self._reasoning_effort}"']) |
| 57 | |
| 58 | # Pass prompt on stdin (the "-" argument tells codex to read from stdin). |
| 59 | cmd.append("-") |
| 60 | |
| 61 | result = subprocess.run( |
| 62 | cmd, |
| 63 | input=prompt, |
| 64 | capture_output=True, |
| 65 | text=True, |
| 66 | timeout=CODEX_TIMEOUT_SECONDS, |
| 67 | ) |
| 68 | |
| 69 | content, usage = _parse_jsonl(result.stdout) |
| 70 | |
| 71 | if result.returncode != 0: |
| 72 | detail = result.stderr.strip() or result.stdout.strip() |
| 73 | if "usage limit" in detail.lower(): |
no outgoing calls