From 6a0202b1b5260e442ebbdb35eeff22bbfa4d1009 Mon Sep 17 00:00:00 2001 From: iris <2+iris@noreply.localhost> Date: Mon, 7 Sep 2026 23:29:15 -0400 Subject: [PATCH] Add Codex JSON sessions and usage limits --- .agents/skills/ai-app-rigs | 1 + AGENTS.md | 6 +- PLAN.md | 45 +- .../src/main/kotlin/com/example/aiapp/Api.kt | 10 +- .../kotlin/com/example/aiapp/SpawnScreen.kt | 19 +- server/src/config.rs | 45 +- server/src/main.rs | 2 +- server/src/session/codex.rs | 563 ++++++++++++++++++ server/src/session/codex/translate.rs | 279 +++++++++ server/src/session/driver.rs | 6 +- server/src/session/mod.rs | 9 + server/src/session/transport.rs | 77 +++ server/src/setups.rs | 1 + server/src/usage.rs | 162 ++++- 14 files changed, 1173 insertions(+), 52 deletions(-) create mode 120000 .agents/skills/ai-app-rigs create mode 100644 server/src/session/codex.rs create mode 100644 server/src/session/codex/translate.rs diff --git a/.agents/skills/ai-app-rigs b/.agents/skills/ai-app-rigs new file mode 120000 index 0000000..9c1b62e --- /dev/null +++ b/.agents/skills/ai-app-rigs @@ -0,0 +1 @@ +../../.claude/skills/ai-app-rigs \ No newline at end of file diff --git a/AGENTS.md b/AGENTS.md index 0468eed..8debab4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,6 +1,6 @@ # ai-app -A phone interface to AI coding sessions (Claude Code and llama.cpp), +A phone interface to AI coding sessions (Codex, Claude Code and llama.cpp), replacing the Claude app for daily use. Rust/Axum backend on the desktop, Kotlin/Compose Android app, WireGuard + pinned self-signed TLS + bearer token between them. @@ -47,6 +47,9 @@ Module-by-module intent is in PLAN.md's "Backend layout". readiness poll watches the process as well as the port, since a model that will not load exits in a second and was being reported as "gave up after 300s". See PLAN.md's "Transport" and "llama-server management". + Codex is `codex exec --json`, one child process per turn; its driver keeps + the thread id for `exec resume`, persists messages queued behind a turn, and + reads subscription limits through the CLI's app-server protocol. - `app/` — the Compose app, package `com.example.aiapp`, label "AI Sessions". `AppRoot.kt` is the navigation `when`; `MainScreen.kt` the root's four tabs (sessions, import, models, setups); `Api.kt`/`EventStream.kt` the REST + SSE @@ -411,4 +414,3 @@ belongs in `~/.claude/TOOLCHAIN.md` or `~/.claude/MACHINE.md` instead. shape: the `markdownIn` scan that decides *what* to parse ran before the hop to `Dispatchers.Default`. The shape to watch for is a `withContext` that wraps the *fetch* and leaves the work done with the result outside it. - diff --git a/PLAN.md b/PLAN.md index 9ffb5e1..afed546 100644 --- a/PLAN.md +++ b/PLAN.md @@ -1,6 +1,6 @@ # ai-app — plan -A phone interface to AI coding sessions — Claude Code and llama.cpp — built +A phone interface to AI coding sessions — Codex, Claude Code and llama.cpp — built to replace the Claude app for day-to-day use. Two motivations: local models need a front end at all, and owning the client means fixing what the official app gets wrong (it won't deliver a typed message until the turn fully @@ -29,11 +29,12 @@ Android app (Compose) backend (Rust/Axum, desktop) ├─ SessionManager ── Session ── Driver (trait) │ ├─ ClaudeDriver (claude stream-json over stdio) + │ ├─ CodexDriver (codex exec --json, one process per turn) │ ├─ LlamaDriver (llama-server over HTTP) │ └─ EchoDriver (the test rig) │ each driver's process is spawned through a Transport, │ locally or as `ssh host …`, decided by the setup it names - ├─ usage.rs (Anthropic OAuth usage endpoint, per machine) + ├─ usage.rs (provider usage meters, per machine) ├─ models.rs (HuggingFace browsing and GGUF downloads) ├─ files.rs (the file explorer's half of the backend) └─ config.ron + per-session transcript files @@ -79,7 +80,7 @@ axum 0.8, axum-server + rustls, tokio, serde, clap, tracing. Rust edition - `config.rs` — the persisted schema. - `setups.rs` — machines and provider discovery. - `files.rs` — the file explorer (`EXPLORER.md`). -- `usage.rs` — Anthropic usage polling, per machine. +- `usage.rs` — provider usage polling, per machine. - `models.rs` — HuggingFace browsing and GGUF downloads. - `media.rs` — the image media-type/extension table, shared by the four places that must agree: storing an upload, serving it back, handing one to @@ -87,7 +88,8 @@ axum 0.8, axum-server + rustls, tokio, serde, clap, tracing. Rust edition - `session/mod.rs` — `SessionManager`, the live registry; every mutation funnels through it so in-memory and on-disk state cannot come apart. - `session/driver.rs` — the `Driver` trait and the common event model. -- `session/claude.rs`, `session/llama.rs`, `session/echo.rs` — the drivers. +- `session/claude.rs`, `session/codex.rs`, `session/llama.rs`, + `session/echo.rs` — the drivers. - `session/transcript.rs` — the append-only JSONL event log per session, with monotonically increasing sequence numbers (the phone's resume cursor). - `session/transport.rs`, `ssh.rs` — running a driver's command locally or @@ -146,9 +148,10 @@ server is going away and means to come back) and `stop` (the session is being deleted, so the process must not survive). Every driver owes exactly one of the two. -`send_user_message` during a run is the point of the whole app: the dialect -queues it for injection at the next tool boundary rather than the end of the -turn. +`send_user_message` during a run is the point of the whole app: a dialect with +a live input channel injects it at the next tool boundary. A turn-at-a-time +dialect persists it and starts the next turn as soon as the current process +ends. ### Claude driver specifics @@ -190,6 +193,34 @@ is the rule behind the import refusal, the single `ClaudeDriver::launch` entry point, and the `Exited` correction below; two CLIs on one session file duplicate the conversation into it and bill the second for re-reading it all. +### Codex driver specifics (2026-09-07) + +Codex uses `codex exec --json`, whose stdout is JSONL: `thread.started`, turn +boundaries, item start/completion records, and the final token usage. The +driver translates those records into the same events as every other session +and persists the reported thread id. Later turns run `codex exec resume +--json`; model, effort and attachments remain launch arguments owned by the +driver rather than branches in routes or screens. + +One `exec` process is one turn and exits normally at its end. The session +therefore owns a sequence of child processes rather than one permanently idle +child: a clean exit after `turn.completed` means `idle`, while an exit before a +turn boundary is an error. A process in flight still uses `process.json` plus +detached stdout/stderr logs, so it survives and is adopted across a backend +restart exactly like the long-lived CLI. Messages received during a turn are +persisted and start later turns in order. The JSON exec surface has no stdin +steering or interactive approval protocol, so it cannot inject a message at a +tool boundary or answer a question inside the same process; those limits are +reported rather than guessed around. + +Codex subscription limits come from the CLI's `account/rateLimits/read` +app-server request on the machine whose setup runs Codex. This keeps login and +token refresh inside the CLI. Its primary and secondary windows are normalized +into the existing usage snapshot shape, under provider name `codex`, so the +phone and auto-resume need no Codex branch. This is the CLI's local protocol +and is treated defensively for the same reason as Claude's undocumented usage +endpoint: missing fields or a refusal degrade to an unavailable snapshot. + ### The llama driver One `llama-server` per session, started through the same `Transport` as any diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt index edf020b..4d89a3b 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt @@ -150,7 +150,7 @@ data class SessionSummary( */ val effort: String?, /** - * Whether a thinking level does anything here -- a Claude CLI session, not a llama or echo one. + * Whether a thinking level does anything here -- a coding CLI session, not llama or echo. * * Asked of the server rather than worked out from the provider's name, because this is a * property of the driver's *kind* and the phone only has the name. @@ -1084,9 +1084,9 @@ fun setSessionModel(settings: ServerSettings, sessionId: String, model: String) } /** - * The permission modes the Claude CLI accepts, in the order they give up asking. "manual" asks for - * everything (each ask arrives on the phone as a question card); the others are the CLI's own - * escalating levels of autonomy. + * Common permission intents, in the order they give up asking. Each coding CLI maps these onto its + * own flags: "manual" asks, "acceptEdits" and "auto" allow ordinary work, "bypassPermissions" + * accepts everything, and "plan" makes no changes. * * One list for every screen that offers them -- spawn, import, and the session's own picker -- * because three copies had already drifted: the import screen was missing "plan". @@ -1115,7 +1115,7 @@ fun setDefaultEffort(settings: ServerSettings, level: String?) { } /** - * How hard the model thinks, as `claude --effort` takes them, cheapest first. + * How hard a coding model thinks, cheapest first. * * Not offered alongside the model and the permission mode on the session's own bar, because it does * not behave like them: the CLI has a control request for those two and none for this (checked diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SpawnScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SpawnScreen.kt index 2b67f83..04ae660 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SpawnScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SpawnScreen.kt @@ -137,10 +137,12 @@ fun SpawnScreen( .getOrDefault(emptyList()) } val current = setup?.providers?.firstOrNull { it.name == providerName } - // Only the Claude CLI has models, a working directory and permission modes; keying the - // extra fields on the kind rather than the provider name keeps a second Claude provider + // Coding CLIs take a working directory, model, permission mode and thinking level. Keying + // the extra fields on the kind rather than the provider name keeps a second installation // from needing anything here. val isClaude = current?.kind == "claude_cli" + val isCodex = current?.kind == "codex_cli" + val isCodingCli = isClaude || isCodex val isLlama = current?.kind == "llama_cpp" // The machine first, because it decides what can be run at all. @@ -237,7 +239,7 @@ fun SpawnScreen( Spacer(Modifier.height(16.dp)) } - if (isClaude) { + if (isCodingCli) { if (current.models.isNotEmpty()) { Spacer(Modifier.height(16.dp)) ChipGroup( @@ -307,7 +309,7 @@ fun SpawnScreen( // an intent about new sessions in general, so a spawn that then // fails must not also lose the choice. Non-fatal for the same // reason the fetch above is -- the session is what was asked for. - if (isClaude) { + if (isCodingCli) { runCatching { setDefaultEffort(settings, effort) } } spawnSession( @@ -319,10 +321,11 @@ fun SpawnScreen( provider = chosen.name, title = title.trim(), model = - if (isLlama) modelKey else model.trim().takeIf { isClaude }, - cwd = cwd.trim().takeIf { isClaude }, - permissionMode = permissionMode.takeIf { isClaude }, - effort = effort.takeIf { isClaude }, + if (isLlama) modelKey + else model.trim().takeIf { isCodingCli }, + cwd = cwd.trim().takeIf { isCodingCli }, + permissionMode = permissionMode.takeIf { isCodingCli }, + effort = effort.takeIf { isCodingCli }, // Sent only when set, so blank means "whatever llama.cpp does // by default" rather than a zero. params = diff --git a/server/src/config.rs b/server/src/config.rs index 7e5fb8f..1798839 100644 --- a/server/src/config.rs +++ b/server/src/config.rs @@ -156,6 +156,9 @@ pub enum DriverKind { /// The Claude Code CLI over stream-json. Named for the CLI specifically: /// bare "claude" would suggest the credit-billed API, which this is not. ClaudeCli, + /// The Codex CLI's `exec --json` JSONL stream. Each process is one turn; + /// the thread id it reports is resumed by the next process. + CodexCli, } impl DriverKind { @@ -176,7 +179,7 @@ impl DriverKind { pub fn max_image_edge(self) -> Option { match self { DriverKind::ClaudeCli => Some(1568), - DriverKind::Echo | DriverKind::LlamaCpp => None, + DriverKind::Echo | DriverKind::LlamaCpp | DriverKind::CodexCli => None, } } @@ -199,6 +202,7 @@ impl DriverKind { pub fn usage_provider(self) -> Option<&'static str> { match self { Self::ClaudeCli => Some(crate::usage::CLAUDE), + Self::CodexCli => Some(crate::usage::CODEX), Self::Echo => Some(crate::usage::ECHO), Self::LlamaCpp => None, } @@ -207,12 +211,13 @@ impl DriverKind { /// The executable a provider of this kind runs when it names none. /// /// Here rather than at each spawn site because it is not only the spawn - /// that runs it: `usage` runs the Claude CLI too, to have it refresh its - /// own OAuth token, and a default that disagreed with the driver's would - /// ask the wrong binary on a machine with two installs. + /// that runs it: `usage` runs provider CLIs too, so a default that + /// disagreed with the driver's would ask the wrong binary on a machine + /// with two installs. pub fn default_program(self) -> &'static str { match self { Self::ClaudeCli => "claude", + Self::CodexCli => "codex", Self::LlamaCpp => "llama-server", // Echo is translated in-process; nothing is spawned for it. Self::Echo => "echo", @@ -222,9 +227,8 @@ impl DriverKind { /// Whether the conversation exists outside this app, so that deleting the /// session here does not end it. /// - /// The Claude Code CLI owns its own transcript and is resumable from it - /// whatever started it, so a session this app spawned is every bit as - /// recoverable as one it imported. Echo has nothing to keep, and a llama + /// Claude Code and Codex own their transcripts and can resume them + /// independently of this app. Echo has nothing to keep, and a llama /// session's conversation is folded out of *this* app's transcript. /// /// Asked before warning somebody that a deletion cannot be undone, which is @@ -232,7 +236,7 @@ impl DriverKind { /// be brought back, it spends the credibility the warning needs. pub fn keeps_own_transcript(self) -> bool { match self { - Self::ClaudeCli => true, + Self::ClaudeCli | Self::CodexCli => true, Self::Echo | Self::LlamaCpp => false, } } @@ -242,15 +246,15 @@ impl DriverKind { /// /// Reported from here rather than decided on the phone, and asked of the /// *kind* rather than branched on: the alternative is the session-type - /// `if` this app does not have anywhere else. `--effort` is the Claude - /// CLI's; a llama session's sampling is `params`, and echo does not think. + /// `if` this app does not have anywhere else. Coding CLIs take an effort + /// setting; a llama session's sampling is `params`, and echo does not think. /// /// It matters more than a control that would simply do nothing, because /// choosing a level stops the process -- so on a session that cannot use /// one it is a button whose only effect is the cost. pub fn takes_effort(self) -> bool { match self { - Self::ClaudeCli => true, + Self::ClaudeCli | Self::CodexCli => true, Self::Echo | Self::LlamaCpp => false, } } @@ -283,20 +287,17 @@ pub struct SessionConfig { pub model: Option, #[serde(skip_serializing_if = "Option::is_none")] pub cwd: Option, - /// Claude permission mode chosen at spawn. Kept as a string because it is - /// passed straight to `--permission-mode` rather than interpreted here, so - /// the CLI stays the one authority on which modes exist. + /// Provider permission mode chosen at spawn. Kept as a string because the + /// driver maps it onto its CLI rather than shared code interpreting it. #[serde(skip_serializing_if = "Option::is_none")] pub permission_mode: Option, - /// How hard the model thinks, passed straight to `--effort`. A string for - /// the same reason `permission_mode` is: the CLI owns which levels exist. + /// How hard the model thinks. A string for the same reason + /// `permission_mode` is: the CLI owns which levels exist. /// - /// Unlike the model and the mode, there is no control request that changes - /// one -- checked against 2.1.258, whose only two are `set_model` and - /// `set_permission_mode` -- so this is settled at launch and `None` means - /// whatever the CLI's own default is. That is a state the phone has to be - /// able to *choose*, not just start in, which is why it is an option - /// rather than a level with a default written here. + /// This is settled at launch and `None` means whatever the CLI's own + /// default is. That is a state the phone has to be able to *choose*, not + /// just start in, which is why it is an option rather than a level with a + /// default written here. #[serde(skip_serializing_if = "Option::is_none")] pub effort: Option, /// Settings the driver interprets, chosen at spawn. diff --git a/server/src/main.rs b/server/src/main.rs index 3571b7b..a7ea498 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -42,7 +42,7 @@ use session::SessionManager; const DEFAULT_PORT: u16 = 8443; -/// Serves AI coding sessions (Claude Code, llama.cpp) to the phone app. +/// Serves AI coding sessions (Codex, Claude Code, llama.cpp) to the phone app. #[derive(Parser)] struct Args { /// TLS port for the whole API surface. diff --git a/server/src/session/codex.rs b/server/src/session/codex.rs new file mode 100644 index 0000000..69237cb --- /dev/null +++ b/server/src/session/codex.rs @@ -0,0 +1,563 @@ +//! Codex CLI sessions over `codex exec --json`. +//! +//! Unlike Claude's long-lived stream, one Codex `exec` process is one turn. It +//! reports a thread id, exits after `turn.completed`, and the next turn is +//! `codex exec resume --json …`. The process and its output files still +//! use the common adoption machinery, so a backend restart does not interrupt +//! a turn that is already running. + +mod translate; + +use std::collections::VecDeque; +use std::os::unix::fs::OpenOptionsExt; +use std::path::{Path, PathBuf}; +use std::process::Stdio; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; + +use super::driver::{AttachmentRef, Driver, Event, EventSink, SessionStatus, Unqueued}; +use super::process; +use super::transport::{Launch, Streams, Transport}; +use crate::config::{ProviderConfig, SessionConfig}; +use translate::Translator; + +const STDOUT_LOG: &str = "codex-stdout.log"; +const STDERR_LOG: &str = "codex-stderr.log"; +const THREAD_FILE: &str = "codex-thread.json"; +const QUEUE_FILE: &str = "codex-queue.json"; +const POLL: std::time::Duration = std::time::Duration::from_millis(50); + +#[derive(Clone, Serialize, Deserialize)] +struct Waiting { + id: String, + text: String, + attachments: Vec, +} + +#[derive(Default, Serialize, Deserialize)] +struct Queue { + waiting: VecDeque, + #[serde(skip)] + running: bool, + #[serde(skip)] + closed: bool, + #[serde(skip)] + interrupting: bool, +} + +struct Settings { + model: Option, + permission_mode: Option, + effort: Option, +} + +struct Inner { + sink: EventSink, + queue: Mutex, + settings: Mutex, + program: String, + cwd: Option, + transport: Transport, + session_dir: PathBuf, + reading: AtomicBool, +} + +pub struct CodexDriver { + inner: Arc, +} + +impl CodexDriver { + pub fn launch( + meta: &SessionConfig, + provider: &ProviderConfig, + transport: Transport, + session_dir: &Path, + sink: EventSink, + ) -> Result { + let mut queue = read_queue(session_dir); + let recorded = process::recorded(session_dir); + queue.running = matches!( + recorded, + Some((_, process::Liveness::Alive | process::Liveness::Unknown)) + ); + let inner = Arc::new(Inner { + sink, + queue: Mutex::new(queue), + settings: Mutex::new(Settings { + model: meta.model.clone(), + permission_mode: meta.permission_mode.clone(), + effort: meta.effort.clone(), + }), + program: provider.program().to_string(), + cwd: meta.cwd.clone(), + transport, + session_dir: session_dir.to_path_buf(), + reading: AtomicBool::new(true), + }); + + match recorded { + Some((record, process::Liveness::Alive | process::Liveness::Unknown)) => { + tracing::info!( + "session {} reattaching to the Codex turn it left running (pid {})", + meta.id, + record.pid + ); + spawn_follower(Arc::clone(&inner), record); + } + Some((_, process::Liveness::Dead)) | None => { + process::clear(session_dir); + let next = inner.queue.lock().unwrap().waiting.pop_front(); + if let Some(next) = next { + save_queue(&inner); + take_and_start(&inner, next); + } else { + let _ = inner.sink.send(Event::Status { + state: SessionStatus::Idle, + }); + } + } + } + Ok(Self { inner }) + } +} + +impl Driver for CodexDriver { + fn send_user_message(&self, text: String, attachments: Vec) { + let mut queue = self.inner.queue.lock().unwrap(); + if queue.closed { + drop(queue); + let _ = self.inner.sink.send(Event::Error { + message: "this Codex session has been stopped".to_string(), + }); + return; + } + if queue.running { + let waiting = Waiting { + id: super::random_hex(), + text, + attachments, + }; + queue.waiting.push_back(waiting.clone()); + drop(queue); + save_queue(&self.inner); + let _ = self.inner.sink.send(Event::MessageQueued { + id: waiting.id, + text: waiting.text, + attachments: waiting.attachments, + }); + return; + } + queue.running = true; + drop(queue); + let waiting = Waiting { + id: String::new(), + text, + attachments, + }; + take_and_start(&self.inner, waiting); + } + + fn unqueue(&self, id: &str) -> Unqueued { + let mut queue = self.inner.queue.lock().unwrap(); + let Some(at) = queue.waiting.iter().position(|message| message.id == id) else { + return Unqueued::Unknown; + }; + queue.waiting.remove(at); + drop(queue); + save_queue(&self.inner); + let _ = self + .inner + .sink + .send(Event::MessageDropped { id: id.to_string() }); + Unqueued::Dropped + } + + fn answer_question(&self, _id: &str, _answers: &[String]) { + let _ = self.inner.sink.send(Event::Error { + message: "Codex's JSON exec stream cannot continue an interactive question".to_string(), + }); + } + + fn interrupt(&self) { + let mut queue = self.inner.queue.lock().unwrap(); + queue.interrupting = true; + drop(queue); + if let Some(record) = process::live(&self.inner.session_dir) { + process::stop(&record, process::STOP_GRACE); + } + } + + fn set_model(&self, model: &str) { + self.inner.settings.lock().unwrap().model = Some(model.to_string()); + let _ = self.inner.sink.send(Event::Settings { + model: Some(model.to_string()), + permission_mode: None, + }); + } + + fn set_permission_mode(&self, mode: &str) { + self.inner.settings.lock().unwrap().permission_mode = Some(mode.to_string()); + let _ = self.inner.sink.send(Event::Settings { + model: None, + permission_mode: Some(mode.to_string()), + }); + } + + fn set_title(&self, _title: &str) {} + + fn run_command(&self, text: &str) { + let _ = self.inner.sink.send(Event::Error { + message: format!("Codex's JSON exec stream has no `{text}` command channel"), + }); + } + + fn compact(&self) { + let _ = self.inner.sink.send(Event::Error { + message: "Codex manages compaction itself in JSON exec sessions".to_string(), + }); + } + + fn clear(&self) { + let path = self.inner.session_dir.join(THREAD_FILE); + if let Err(err) = std::fs::remove_file(&path) + && err.kind() != std::io::ErrorKind::NotFound + { + let _ = self.inner.sink.send(Event::Error { + message: format!("couldn't clear the Codex thread id: {err}"), + }); + return; + } + let _ = self.inner.sink.send(Event::Cleared); + } + + fn between_turns(&self) -> bool { + !self.inner.queue.lock().unwrap().running + } + + fn detach(&self) { + self.inner.reading.store(false, Ordering::SeqCst); + } + + fn stop(&self) { + self.inner.reading.store(false, Ordering::SeqCst); + let dropped = { + let mut queue = self.inner.queue.lock().unwrap(); + queue.closed = true; + queue + .waiting + .drain(..) + .map(|message| message.id) + .collect::>() + }; + save_queue(&self.inner); + for id in dropped { + let _ = self.inner.sink.send(Event::MessageDropped { id }); + } + if let Some(record) = process::live(&self.inner.session_dir) { + process::stop(&record, process::STOP_GRACE); + } + process::clear(&self.inner.session_dir); + } +} + +fn take_and_start(inner: &Arc, message: Waiting) { + { + inner.queue.lock().unwrap().running = true; + } + match start_process(inner, &message.text, &message.attachments) { + Ok(record) => { + let _ = inner.sink.send(Event::MessageTaken { + id: (!message.id.is_empty()).then_some(message.id), + text: message.text, + attachments: message.attachments, + }); + let _ = inner.sink.send(Event::Status { + state: SessionStatus::Running, + }); + spawn_follower(Arc::clone(inner), record); + } + Err(err) => { + inner.queue.lock().unwrap().running = false; + if !message.id.is_empty() { + let _ = inner.sink.send(Event::MessageDropped { id: message.id }); + } + let _ = inner.sink.send(Event::Error { + message: format!("couldn't start Codex: {err:#}"), + }); + let _ = inner.sink.send(Event::Status { + state: SessionStatus::Idle, + }); + } + } +} + +fn start_process( + inner: &Inner, + text: &str, + attachments: &[AttachmentRef], +) -> Result { + let settings = inner.settings.lock().unwrap(); + let mut args = Vec::new(); + match settings.permission_mode.as_deref() { + Some("bypassPermissions") => { + args.push("--dangerously-bypass-approvals-and-sandbox".to_string()); + } + Some("manual") => { + args.extend(["--ask-for-approval".to_string(), "on-request".to_string()]); + } + Some("auto" | "acceptEdits") => args.push("--approve-for-me".to_string()), + Some("plan") => args.extend([ + "--ask-for-approval".to_string(), + "never".to_string(), + "--sandbox".to_string(), + "read-only".to_string(), + ]), + _ => {} + } + args.push("exec".to_string()); + if let Some(thread) = read_thread(&inner.session_dir) { + args.extend(["resume".to_string(), thread]); + } + args.push("--json".to_string()); + args.push("--skip-git-repo-check".to_string()); + if let Some(model) = &settings.model { + args.extend(["--model".to_string(), model.clone()]); + } + if let Some(effort) = &settings.effort { + args.extend([ + "--config".to_string(), + format!("model_reasoning_effort={}", json!(effort)), + ]); + } + drop(settings); + + let mut body = text.to_string(); + for attachment in attachments { + let path = attachment_path(&inner.session_dir, attachment)?; + if crate::media::media_type_for(attachment).is_some() + && matches!(inner.transport, Transport::Here) + { + args.extend(["--image".to_string(), path.display().to_string()]); + } else { + if !body.is_empty() { + body.push_str("\n\n"); + } + body.push_str(&format!("Attached file: {}", path.display())); + } + } + args.push(body); + + let stdout = create_log(&inner.session_dir.join(STDOUT_LOG))?; + let stderr = create_log(&inner.session_dir.join(STDERR_LOG))?; + let launch = Launch::new(&inner.program, args, inner.cwd.as_deref()); + let mut child = inner.transport.spawn( + &launch, + Streams::Detached { + stdin: Stdio::null(), + stdout: stdout.into(), + stderr: stderr.into(), + }, + )?; + let pid = child + .id() + .context("Codex exited before it could be recorded")?; + tokio::spawn(async move { + let _ = child.wait().await; + }); + let record = process::Record::of(pid, process::Detail::Stdio { stdout_read: 0 }) + .context("Codex was gone before its start time could be read")?; + process::write(&inner.session_dir, &record); + Ok(record) +} + +fn spawn_follower(inner: Arc, record: process::Record) { + let offset = match record.detail { + process::Detail::Stdio { stdout_read } => stdout_read, + _ => 0, + }; + tokio::spawn(follow(inner, record, offset)); +} + +async fn follow(inner: Arc, mut record: process::Record, mut offset: u64) { + let stdout = inner.session_dir.join(STDOUT_LOG); + let stderr = inner.session_dir.join(STDERR_LOG); + let mut translator = Translator::default(); + while inner.reading.load(Ordering::SeqCst) { + let (bytes, _) = match process::read_from(&stdout, offset) { + Ok(read) => read, + Err(err) => { + let _ = inner.sink.send(Event::Error { + message: format!("couldn't read Codex output: {err:#}"), + }); + return; + } + }; + let complete = bytes + .iter() + .rposition(|byte| *byte == b'\n') + .map(|at| at + 1) + .unwrap_or(0); + for line in String::from_utf8_lossy(&bytes[..complete]).lines() { + let Ok(value) = serde_json::from_str::(line) else { + tracing::warn!( + "unparseable Codex JSONL line: {}", + line.chars().take(200).collect::() + ); + continue; + }; + let before = translator.thread_id.clone(); + for event in translator.translate(&value) { + let _ = inner.sink.send(event); + } + if translator.thread_id != before + && let Some(thread) = &translator.thread_id + { + write_thread(&inner.session_dir, thread); + } + } + if complete > 0 { + offset += complete as u64; + record.detail = process::Detail::Stdio { + stdout_read: offset, + }; + process::write(&inner.session_dir, &record); + } + match record.liveness() { + process::Liveness::Alive => {} + process::Liveness::Unknown => {} + process::Liveness::Dead if complete > 0 => {} + process::Liveness::Dead => { + process::clear(&inner.session_dir); + let mut queue = inner.queue.lock().unwrap(); + let interrupted = std::mem::take(&mut queue.interrupting); + queue.running = false; + let next = queue.waiting.pop_front(); + drop(queue); + save_queue(&inner); + if !translator.completed() && !translator.limited() && !interrupted { + let detail = stderr_tail(&stderr); + let _ = inner.sink.send(Event::Error { + message: if detail.is_empty() { + "Codex exited before completing the turn".to_string() + } else { + format!("Codex exited:\n{detail}") + }, + }); + } + if !translator.completed() { + let _ = inner.sink.send(Event::Status { + state: SessionStatus::Idle, + }); + } + if let Some(next) = next { + take_and_start(&inner, next); + } + return; + } + } + tokio::time::sleep(POLL).await; + } +} + +fn create_log(path: &Path) -> Result { + std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .mode(0o600) + .open(path) + .with_context(|| format!("creating {}", path.display())) +} + +fn attachment_path(session_dir: &Path, id: &str) -> Result { + if id.contains("..") + || !id + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_')) + { + anyhow::bail!("invalid attachment id"); + } + Ok(session_dir.join("attachments").join(id)) +} + +fn read_thread(session_dir: &Path) -> Option { + serde_json::from_str::(&std::fs::read_to_string(session_dir.join(THREAD_FILE)).ok()?) + .ok()? + .get("threadId")? + .as_str() + .map(str::to_string) +} + +fn write_thread(session_dir: &Path, thread: &str) { + let path = session_dir.join(THREAD_FILE); + let written = std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .mode(0o600) + .open(&path) + .and_then(|mut file| { + use std::io::Write; + file.write_all(json!({"threadId": thread}).to_string().as_bytes()) + }); + if let Err(err) = written { + tracing::error!( + "couldn't persist Codex thread id to {}: {err}", + path.display() + ); + } +} + +fn read_queue(session_dir: &Path) -> Queue { + std::fs::read_to_string(session_dir.join(QUEUE_FILE)) + .ok() + .and_then(|text| serde_json::from_str(&text).ok()) + .unwrap_or_default() +} + +fn save_queue(inner: &Inner) { + let path = inner.session_dir.join(QUEUE_FILE); + let text = match serde_json::to_string(&*inner.queue.lock().unwrap()) { + Ok(text) => text, + Err(err) => { + tracing::error!("couldn't serialize the Codex queue: {err}"); + return; + } + }; + let temporary = path.with_extension("json.new"); + let written = std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .mode(0o600) + .open(&temporary) + .and_then(|mut file| { + use std::io::Write; + file.write_all(text.as_bytes()) + }) + .and_then(|()| std::fs::rename(&temporary, &path)); + if let Err(err) = written { + tracing::error!( + "couldn't persist the Codex queue to {}: {err}", + path.display() + ); + let _ = std::fs::remove_file(temporary); + } +} + +fn stderr_tail(path: &Path) -> String { + std::fs::read_to_string(path) + .unwrap_or_default() + .lines() + .rev() + .take(20) + .collect::>() + .into_iter() + .rev() + .collect::>() + .join("\n") +} diff --git a/server/src/session/codex/translate.rs b/server/src/session/codex/translate.rs new file mode 100644 index 0000000..c8440b8 --- /dev/null +++ b/server/src/session/codex/translate.rs @@ -0,0 +1,279 @@ +//! `codex exec --json` lines into the common event model. +//! +//! The CLI promises JSONL but deliberately leaves room for new item kinds. We +//! therefore match only the records that have a useful common equivalent and +//! ignore the rest; an added Codex item must not make a live session go deaf. + +use serde_json::{Value, json}; + +use super::super::driver::{Event, SessionStatus}; + +#[derive(Default)] +pub(super) struct Translator { + pub(super) thread_id: Option, + completed: bool, + limited: bool, +} + +impl Translator { + pub(super) fn translate(&mut self, line: &Value) -> Vec { + match line.get("type").and_then(Value::as_str) { + Some("thread.started") => { + self.thread_id = line + .get("thread_id") + .and_then(Value::as_str) + .map(str::to_string); + Vec::new() + } + Some("turn.started") => vec![Event::Status { + state: SessionStatus::Running, + }], + Some("item.started") => start_item(&line["item"]), + Some("item.updated") => update_item(&line["item"]), + Some("item.completed") => complete_item(&line["item"]), + Some("turn.completed") => { + self.completed = true; + let mut events = Vec::new(); + if let Some(usage) = line.get("usage") { + let input = number(usage, "input_tokens"); + let output = number(usage, "output_tokens"); + if input.is_some() || output.is_some() { + events.push(Event::UsageDelta { + tokens: input.unwrap_or(0) + output.unwrap_or(0), + // `exec` reports the sum across every model call in + // a turn, not the final call's context. + context: None, + }); + } + } + events.push(Event::Status { + state: SessionStatus::Idle, + }); + events + } + Some("turn.failed") | Some("error") => self.failure(line), + _ => Vec::new(), + } + } + + pub(super) fn completed(&self) -> bool { + self.completed + } + + pub(super) fn limited(&self) -> bool { + self.limited + } + + fn failure(&mut self, line: &Value) -> Vec { + let detail = error_text(line); + if is_limit(&detail) { + if self.limited { + return Vec::new(); + } + self.limited = true; + vec![Event::LimitReached { + resets_at: find_reset(line), + }] + } else { + vec![Event::Error { message: detail }] + } + } +} + +fn start_item(item: &Value) -> Vec { + let Some((id, tool, input)) = tool(item) else { + return Vec::new(); + }; + vec![Event::ToolStart { id, tool, input }] +} + +fn update_item(item: &Value) -> Vec { + let Some(id) = item.get("id").and_then(Value::as_str) else { + return Vec::new(); + }; + let output = item + .get("aggregated_output") + .or_else(|| item.get("output")) + .and_then(value_text); + output + .filter(|text| !text.is_empty()) + .map(|output| { + vec![Event::ToolUpdate { + id: id.to_string(), + output, + }] + }) + .unwrap_or_default() +} + +fn complete_item(item: &Value) -> Vec { + match item.get("type").and_then(Value::as_str) { + Some("agent_message") => item + .get("text") + .and_then(Value::as_str) + .filter(|text| !text.is_empty()) + .map(|delta| { + vec![Event::AssistantText { + delta: delta.to_string(), + }] + }) + .unwrap_or_default(), + Some("reasoning") => Vec::new(), + _ => { + let Some((id, _, _)) = tool(item) else { + return Vec::new(); + }; + let output = tool_output(item); + vec![Event::ToolEnd { id, output }] + } + } +} + +fn tool(item: &Value) -> Option<(String, String, Value)> { + let id = item.get("id")?.as_str()?.to_string(); + let kind = item.get("type")?.as_str()?; + let (name, input) = match kind { + "command_execution" => ( + "exec_command".to_string(), + json!({"command": item.get("command").cloned().unwrap_or(Value::Null)}), + ), + "file_change" => ( + "apply_patch".to_string(), + item.get("changes").cloned().unwrap_or(Value::Null), + ), + "mcp_tool_call" => ( + item.get("tool") + .or_else(|| item.get("name")) + .and_then(Value::as_str) + .map(|tool| format!("mcp:{tool}")) + .unwrap_or_else(|| "mcp".to_string()), + item.get("arguments").cloned().unwrap_or(Value::Null), + ), + "web_search" => ( + "web_search".to_string(), + json!({"query": item.get("query").cloned().unwrap_or(Value::Null)}), + ), + "todo_list" => ("update_plan".to_string(), item.clone()), + _ => return None, + }; + Some((id, name, input)) +} + +fn tool_output(item: &Value) -> String { + for key in ["aggregated_output", "output", "result", "error"] { + if let Some(text) = item.get(key).and_then(value_text) + && !text.is_empty() + { + return text; + } + } + match item.get("status").and_then(Value::as_str) { + Some(status) => status.to_string(), + None => String::new(), + } +} + +fn value_text(value: &Value) -> Option { + value + .as_str() + .map(str::to_string) + .or_else(|| (!value.is_null()).then(|| value.to_string())) +} + +fn number(value: &Value, key: &str) -> Option { + value.get(key).and_then(Value::as_u64) +} + +fn error_text(line: &Value) -> String { + line.get("message") + .and_then(Value::as_str) + .or_else(|| line.pointer("/error/message").and_then(Value::as_str)) + .or_else(|| line.get("error").and_then(Value::as_str)) + .unwrap_or("Codex ended the turn with an unknown error") + .to_string() +} + +fn is_limit(detail: &str) -> bool { + let lower = detail.to_ascii_lowercase(); + lower.contains("usage limit") + || lower.contains("rate limit") + || lower.contains("quota exceeded") + || lower.contains("credits depleted") +} + +fn find_reset(value: &Value) -> Option { + for key in ["resets_at", "resetsAt", "reset_at", "resetAt"] { + if let Some(at) = value.get(key).and_then(Value::as_f64) { + return Some(at); + } + } + value.as_object()?.values().find_map(find_reset) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn line(text: &str) -> Value { + serde_json::from_str(text).expect("fixture") + } + + #[test] + fn translates_the_observed_minimal_stream() { + let mut translator = Translator::default(); + assert!( + translator + .translate(&line(r#"{"type":"thread.started","thread_id":"thread-1"}"#)) + .is_empty() + ); + assert_eq!(translator.thread_id.as_deref(), Some("thread-1")); + assert_eq!( + translator.translate(&line( + r#"{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"hello"}}"# + )), + vec![Event::AssistantText { + delta: "hello".to_string() + }] + ); + let events = translator.translate(&line( + r#"{"type":"turn.completed","usage":{"input_tokens":13,"cached_input_tokens":8,"output_tokens":5}}"#, + )); + assert_eq!( + events[0], + Event::UsageDelta { + tokens: 18, + context: None + } + ); + assert!(translator.completed()); + } + + #[test] + fn translates_tools_and_limits_without_matching_whole_records() { + let mut translator = Translator::default(); + let started = translator.translate(&line( + r#"{"type":"item.started","item":{"id":"item_1","type":"command_execution","command":"pwd","status":"in_progress"}}"#, + )); + assert!(matches!(&started[0], Event::ToolStart { tool, .. } if tool == "exec_command")); + let ended = translator.translate(&line( + r#"{"type":"item.completed","item":{"id":"item_1","type":"command_execution","command":"pwd","aggregated_output":"/tmp\n","exit_code":0,"status":"completed"}}"#, + )); + assert_eq!( + ended, + vec![Event::ToolEnd { + id: "item_1".to_string(), + output: "/tmp\n".to_string() + }] + ); + let limit = translator.translate(&line( + r#"{"type":"turn.failed","error":{"message":"usage limit reached","resetsAt":1234}}"#, + )); + assert_eq!( + limit, + vec![Event::LimitReached { + resets_at: Some(1234.0) + }] + ); + assert!(translator.limited()); + } +} diff --git a/server/src/session/driver.rs b/server/src/session/driver.rs index d67d34a..eb1cac3 100644 --- a/server/src/session/driver.rs +++ b/server/src/session/driver.rs @@ -498,9 +498,9 @@ pub type EventSink = mpsc::UnboundedSender; /// The inbound half of a session. Deliberately small; see PLAN.md for the /// per-driver mapping of each method onto its dialect. /// -/// `send_user_message` during a run is the point of the whole app: both -/// real dialects queue it for injection at the next tool boundary rather -/// than the end of the turn. +/// `send_user_message` during a run is the point of the whole app: a dialect +/// with live input injects it at the next tool boundary, while a turn-at-a-time +/// dialect queues it for the next child process. pub trait Driver: Send + Sync { /// Takes a message, now or once the session is free for it. /// diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index 060c7e5..5714cfa 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -9,6 +9,7 @@ //! fall behind or reconnect catch up from the file by cursor. pub mod claude; +pub mod codex; pub mod driver; pub mod echo; pub mod import; @@ -33,6 +34,7 @@ use crate::config::{ SetupConfig, SshConfig, TokenEntry, }; use claude::ClaudeDriver; +use codex::CodexDriver; use driver::{ AttachmentRef, Driver, Event, EventSink, SessionCommand, SessionStatus, Unqueued, context_after, }; @@ -2471,6 +2473,13 @@ fn make_driver( sink.clone(), Arc::clone(subagents), )?), + DriverKind::CodexCli => Arc::new(CodexDriver::launch( + meta, + provider, + Transport::for_setup(setup), + dir, + sink.clone(), + )?), }) } diff --git a/server/src/session/transport.rs b/server/src/session/transport.rs index 5e20ff4..0054340 100644 --- a/server/src/session/transport.rs +++ b/server/src/session/transport.rs @@ -101,6 +101,83 @@ pub enum Transport { } impl Transport { + /// Exchanges newline-delimited JSON requests with a short-lived stdio + /// server. `initial` is written first; after its response arrives, + /// `requests` is written and the response bearing `wanted_id` is returned. + /// + /// This is the shape Codex's app-server requires for a usage read: an + /// initialize round trip must finish before the initialized notification + /// and account request are accepted. + pub fn request_json_blocking( + &self, + launch: &Launch, + initial: &serde_json::Value, + requests: &[serde_json::Value], + wanted_id: u64, + ) -> Result { + use std::io::{BufRead, BufReader, Write}; + + let host = match self { + Self::Here => None, + Self::Ssh { ssh, .. } => Some(ssh), + }; + let mut command = crate::ssh::command( + host, + &launch.program, + &launch.args, + launch.cwd.as_deref(), + launch.forward, + ); + command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let mut child = command + .spawn() + .with_context(|| format!("couldn't run \"{}\" {}", launch.program, self.describe()))?; + let mut stdin = child.stdin.take().context("the JSON server has no stdin")?; + let stdout = child + .stdout + .take() + .context("the JSON server has no stdout")?; + writeln!(stdin, "{initial}")?; + stdin.flush()?; + let mut reader = BufReader::new(stdout); + let mut line = String::new(); + loop { + line.clear(); + if reader.read_line(&mut line)? == 0 { + anyhow::bail!("the JSON server exited before initialization completed"); + } + let Ok(value) = serde_json::from_str::(&line) else { + continue; + }; + if value.get("id").and_then(serde_json::Value::as_u64) + == initial.get("id").and_then(serde_json::Value::as_u64) + { + break; + } + } + for request in requests { + writeln!(stdin, "{request}")?; + } + stdin.flush()?; + loop { + line.clear(); + if reader.read_line(&mut line)? == 0 { + anyhow::bail!("the JSON server exited before answering request {wanted_id}"); + } + let Ok(value) = serde_json::from_str::(&line) else { + continue; + }; + if value.get("id").and_then(serde_json::Value::as_u64) == Some(wanted_id) { + let _ = child.kill(); + let _ = child.wait(); + return Ok(value); + } + } + } + /// The transport a setup describes; a setup with no `ssh` is here. pub fn for_setup(setup: &crate::config::SetupConfig) -> Self { match &setup.ssh { diff --git a/server/src/setups.rs b/server/src/setups.rs index 1b2a853..9faaedd 100644 --- a/server/src/setups.rs +++ b/server/src/setups.rs @@ -25,6 +25,7 @@ use crate::session::transport::{Launch, Transport}; /// session stores. const PROBES: &[(&str, &str, DriverKind)] = &[ ("claude-cli", "claude", DriverKind::ClaudeCli), + ("codex-cli", "codex", DriverKind::CodexCli), // Named for the program rather than for where it runs: it runs // wherever the setup is, and "local" was true only while a llama // session could not be spawned on another machine. diff --git a/server/src/usage.rs b/server/src/usage.rs index 1f14a9e..daa59e2 100644 --- a/server/src/usage.rs +++ b/server/src/usage.rs @@ -1,4 +1,4 @@ -//! Usage-limit reporting -- the same numbers as Claude Code's `/usage`. +//! Usage-limit reporting -- the same numbers the provider CLIs show. //! //! Polls `https://api.anthropic.com/api/oauth/usage` with the OAuth access //! token from Claude Code's local credential store. The endpoint is @@ -12,8 +12,8 @@ //! below enforces the latter across any number of phone refreshes; there is no //! background poll at all. //! -//! One [`UsageProvider`] per paid service, so a second service later is a new -//! impl behind the same snapshot shape, not a parallel screen. +//! One [`UsageProvider`] per paid service keeps each provider's wire format +//! behind the same snapshot shape. //! //! **Asked of the machine that spends the tokens, not of this one.** A session //! runs wherever its setup says, so the account being billed is that machine's. @@ -21,7 +21,7 @@ //! no `claude` CLI, and the CLI machine is a remote -- so the one set of numbers //! the screen could show would be an account with no sessions. Credentials are //! read through the session `Transport`, one snapshot per setup that offers -//! Claude. +//! that provider. //! //! The token is read *to* the backend and the HTTP call is made from here, so //! the far machine needs nothing beyond a shell and the wire format stays in @@ -112,6 +112,8 @@ pub struct UsageSnapshot { /// labels a row with, and [`crate::config::DriverKind::usage_provider`], /// which is how a session says which of those rows is about it. pub const CLAUDE: &str = "claude"; +/// ChatGPT-backed Codex CLI subscription usage. +pub const CODEX: &str = "codex"; /// The invented one, for testing the screens that draw these -- see /// [`Fixture`]. pub const ECHO: &str = "echo"; @@ -216,6 +218,129 @@ impl UsageProvider for ClaudeUsage { } } +/// Reads the same snapshot as Codex's status display through the CLI's local +/// app-server protocol. The CLI owns authentication and token refresh; this +/// process never opens or copies its credentials. +pub struct CodexUsage { + pub setup: String, + pub setup_name: String, + pub transport: Transport, + pub program: String, +} + +impl CodexUsage { + fn snapshot(&self, state: UsageState, windows: Vec) -> UsageSnapshot { + UsageSnapshot { + provider: CODEX.to_string(), + setup: self.setup.clone(), + setup_name: self.setup_name.clone(), + state, + windows, + fetched_at: crate::session::now(), + } + } +} + +impl UsageProvider for CodexUsage { + fn name(&self) -> &'static str { + CODEX + } + + fn fetch(&self) -> UsageSnapshot { + let launch = Launch::new( + &self.program, + vec!["app-server".to_string(), "--stdio".to_string()], + None, + ); + let initialized = serde_json::json!({ + "id": 1, + "method": "initialize", + "params": {"clientInfo": {"name": "ai-app", "title": "AI Sessions", "version": env!("CARGO_PKG_VERSION")}} + }); + let requests = [ + serde_json::json!({"method": "initialized"}), + serde_json::json!({"id": 2, "method": "account/rateLimits/read"}), + ]; + let answer = match self + .transport + .request_json_blocking(&launch, &initialized, &requests, 2) + { + Ok(answer) => answer, + Err(err) => { + return self.snapshot( + UsageState::Unreachable { + detail: format!("couldn't ask Codex on {}: {err:#}", self.setup_name), + }, + Vec::new(), + ); + } + }; + if let Some(error) = answer.pointer("/error/message").and_then(Value::as_str) { + let state = if error.to_ascii_lowercase().contains("login") + || error.to_ascii_lowercase().contains("authentication") + { + UsageState::NotLoggedIn + } else { + UsageState::Failed { + detail: error.to_string(), + } + }; + return self.snapshot(state, Vec::new()); + } + let Some(limits) = answer.pointer("/result/rateLimits") else { + return self.snapshot( + UsageState::Failed { + detail: "Codex returned no rate-limit snapshot".to_string(), + }, + Vec::new(), + ); + }; + self.snapshot(UsageState::Ok, parse_codex_windows(limits)) + } +} + +fn parse_codex_windows(limits: &Value) -> Vec { + [("primary", true), ("secondary", false)] + .into_iter() + .filter_map(|(kind, primary)| { + let window = limits.get(kind)?; + if window.is_null() { + return None; + } + let minutes = window.get("windowDurationMins").and_then(Value::as_u64); + let label = match minutes { + Some(300) => "5-hour window".to_string(), + Some(10_080) => "Weekly".to_string(), + Some(minutes) if minutes % 1_440 == 0 => { + format!("{}-day window", minutes / 1_440) + } + Some(minutes) if minutes % 60 == 0 => { + format!("{}-hour window", minutes / 60) + } + Some(minutes) => format!("{minutes}-minute window"), + None if primary => "Primary window".to_string(), + None => "Secondary window".to_string(), + }; + Some(UsageWindow { + // Common semantic names: the phone's compact bar asks for + // `session`, and auto-resume treats all windows alike. + kind: if primary { "session" } else { "weekly_all" }.to_string(), + label, + percent: window.get("usedPercent")?.as_f64()?, + resets_at: window + .get("resetsAt") + .and_then(Value::as_i64) + .and_then(|seconds| time::OffsetDateTime::from_unix_timestamp(seconds).ok()) + .and_then(|at| { + at.format(&time::format_description::well_known::Rfc3339) + .ok() + }), + active: primary, + }) + }) + .collect() +} + /// Why one call to the usage endpoint did not produce numbers. /// /// 401 is apart from the rest because it is the only one with a way out: the @@ -621,6 +746,12 @@ fn providers_for(setup: &SetupConfig, fixture: &Fixture) -> Vec found.push(Box::new(CodexUsage { + setup: setup.id.clone(), + setup_name: setup.name.clone(), + transport: Transport::for_setup(setup), + program: provider.program().to_string(), + })), // Nothing at all until a test has asked for something: an // echo session costs nothing, so the honest answer is no row // rather than a row saying zero. @@ -763,6 +894,7 @@ mod tests { .expect("json"); let windows = parse_windows(&body); assert_eq!(windows.len(), 4); + assert_eq!(windows[0].kind, "session"); assert_eq!(windows[0].label, "5-hour window"); assert_eq!(windows[0].percent, 70.0); assert!(windows[0].active); @@ -873,6 +1005,7 @@ mod tests { assert_eq!(found[0].name(), ECHO); assert_eq!(DriverKind::Echo.usage_provider(), Some(ECHO)); assert_eq!(DriverKind::ClaudeCli.usage_provider(), Some(CLAUDE)); + assert_eq!(DriverKind::CodexCli.usage_provider(), Some(CODEX)); // A local model costs nothing to run, so it meters nothing. assert_eq!(DriverKind::LlamaCpp.usage_provider(), None); } @@ -920,4 +1053,25 @@ mod tests { assert!(parse_windows(&serde_json::json!({})).is_empty()); assert!(parse_windows(&serde_json::json!({"limits": "what"})).is_empty()); } + + #[test] + fn parses_codex_primary_and_secondary_windows() { + let limits = serde_json::json!({ + "primary": {"usedPercent": 10, "windowDurationMins": 300, "resetsAt": 1788853003_i64}, + "secondary": {"usedPercent": 2, "windowDurationMins": 10080, "resetsAt": 1789439803_i64} + }); + let windows = parse_codex_windows(&limits); + assert_eq!(windows.len(), 2); + assert_eq!(windows[0].label, "5-hour window"); + assert_eq!(windows[0].percent, 10.0); + assert!(windows[0].active); + assert_eq!(windows[1].kind, "weekly_all"); + assert_eq!(windows[1].label, "Weekly"); + assert!( + windows[1] + .resets_at + .as_deref() + .is_some_and(|at| at.ends_with('Z')) + ); + } }