From 8c88a7e99149063a7ca0a513dca523f20b30fda7 Mon Sep 17 00:00:00 2001 From: iris <2+iris@noreply.localhost> Date: Wed, 9 Sep 2026 12:19:11 -0400 Subject: [PATCH] Use native Codex steering and transcript deletion --- .claude/skills/ai-app-rigs/SKILL.md | 2 +- AGENTS.md | 7 +- PLAN.md | 42 +- .../src/main/kotlin/com/example/aiapp/Api.kt | 11 +- .../com/example/aiapp/SessionListScreen.kt | 12 +- server/src/config.rs | 16 +- server/src/routes.rs | 23 +- server/src/session/claude.rs | 45 +- server/src/session/codex.rs | 957 +++++++++++++----- server/src/session/codex/translate.rs | 161 ++- server/src/session/mod.rs | 85 +- server/src/session/process.rs | 33 + 12 files changed, 1004 insertions(+), 390 deletions(-) diff --git a/.claude/skills/ai-app-rigs/SKILL.md b/.claude/skills/ai-app-rigs/SKILL.md index 0c7ecbd..d77a686 100644 --- a/.claude/skills/ai-app-rigs/SKILL.md +++ b/.claude/skills/ai-app-rigs/SKILL.md @@ -200,7 +200,7 @@ never be able to close the app, whatever produced it. **Deleting a session offers to take the machine's own transcript with it** — `DELETE /sessions/{id}?deleteForeign=true`, behind a switch in the confirmation, and only where the driver keeps a record of its own -(`keepsOwnTranscript`, which today means Claude Code). Off by default, +(`keepsOwnTranscript`, currently Claude Code or Codex). Off by default, because leaving that copy is what makes an ordinary delete recoverable — and the dialog's paragraph is rewritten when it is on rather than appended to, since the sentence promising the conversation "should still be there to diff --git a/AGENTS.md b/AGENTS.md index 8debab4..b28aa86 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -47,9 +47,10 @@ 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. + Codex is one persistent `codex app-server --stdio` process per session; its + driver uses native turn steering and interruption, persists the protocol + state and thread id, and reads subscription limits through the same CLI + 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 diff --git a/PLAN.md b/PLAN.md index 8285d26..8289166 100644 --- a/PLAN.md +++ b/PLAN.md @@ -29,7 +29,7 @@ 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) + │ ├─ CodexDriver (persistent codex app-server JSONL) │ ├─ LlamaDriver (llama-server over HTTP) │ └─ EchoDriver (the test rig) │ each driver's process is spawned through a Transport, @@ -193,25 +193,24 @@ 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 driver specifics (2026-09-09) -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. +Codex uses one persistent `codex app-server --stdio` per session. The original +2026-09-07 implementation used one `codex exec --json` process per turn, but +that surface cannot steer: a message typed during work was held until the turn +ended, and Pause killed the whole process before starting another resume. The +app-server protocol provides the operations the interface actually promises: +`turn/steer` injects a message into the active turn and `turn/interrupt` stops +that turn while leaving the conversation process alive. -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. +The process's stdin is a fifo and its output is a detached log, with protocol +state persisted beside the thread id. It therefore survives and is adopted +across a backend restart like the Claude CLI. A steer is sent immediately and +is announced where Codex emits its user-message item; if Codex says the active +turn is not steerable, the message remains queued and starts the next turn +instead of being lost. An interrupt requested while a turn is still starting +is applied once Codex supplies that turn's id, so it cannot leak forward and +hide a later failure. 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 @@ -810,6 +809,13 @@ directory under `$XDG_DATA_HOME/ai-app/sessions/` (transcript, attachments, produced images, process record), owner-only. Deleting a session is the complete path out of everything spawning one created. +Claude Code and Codex also keep their own durable transcript. The delete +dialog names that owner and can remove its copy too: Claude files are resolved +under `~/.claude/projects`, while a Codex thread id resolves only the matching +rollout under `~/.codex/sessions`. The provider-owned copy is deleted first, so +a remote-machine failure leaves the app session intact rather than reporting a +half-delete as success. + **Every request body refuses fields it does not know** (`serde(deny_unknown_fields)`). A caller that misspells `permissionMode` got a 200 and a session in the default mode, which is indistinguishable from 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 f70bb30..e33ef65 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt @@ -140,6 +140,8 @@ data class SessionSummary( * still there, and re-importing is not a restore. */ val keepsOwnTranscript: Boolean, + /** Product whose durable transcript survives an ordinary app deletion. */ + val ownTranscriptName: String?, /** How much the session asks before acting; null when it was never set. */ val permissionMode: String?, /** @@ -238,6 +240,7 @@ private fun parseSession(session: JSONObject) = id = session.getString("id"), setup = session.getString("setup"), keepsOwnTranscript = session.optBoolean("keepsOwnTranscript", false), + ownTranscriptName = session.optString("ownTranscriptName").ifEmpty { null }, setupName = session.getString("setupName"), provider = session.getString("provider"), title = session.getString("title"), @@ -1230,10 +1233,10 @@ fun compactSession(settings: ServerSettings, sessionId: String) { /** * Removes a session, and optionally the machine's own transcript of the same conversation. * - * [deleteForeign] is the delete this app cannot otherwise reach: Claude Code keeps its own record - * under `~/.claude/projects`, and leaving it is what makes an ordinary delete recoverable. The - * server does both halves, and does the unrecoverable one first, so a machine it cannot reach - * leaves the session exactly where it was rather than half-deleted. + * [deleteForeign] is the delete this app cannot otherwise reach: coding CLIs keep their own durable + * record, and leaving it is what makes an ordinary delete recoverable. The server does both halves, + * and does the unrecoverable one first, so a machine it cannot reach leaves the session exactly + * where it was rather than half-deleted. */ fun deleteSession(settings: ServerSettings, sessionId: String, deleteForeign: Boolean = false) { val query = if (deleteForeign) "?deleteForeign=true" else "" diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt index ef2164b..ce60878 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt @@ -318,17 +318,19 @@ fun SessionListScreen( // Reset per session, so a toggle turned on for one conversation is not still on for the // next. Off to begin with: see [deleteSession]. var alsoDeleteForeign by remember(session.id) { mutableStateOf(false) } + // Old servers reported only the capability, when Claude Code was its sole owner. + val transcriptOwner = session.ownTranscriptName ?: "Claude Code" AlertDialog( onDismissRequest = { confirmingDelete = null }, title = { Text("Delete \"${session.title}\"?") }, text = { // Two different acts behind one button, so it says which one this is. What // separates them is whether the *driver* keeps its own record of the conversation - // -- the Claude Code CLI does, whether this app spawned the session or imported it; + // -- the coding CLIs do, whether this app spawned the session or imported it; // echo and llama.cpp do not. // // This used to branch on `imported`, above a comment asserting that "a session - // started here has no copy anywhere". That was false for every claude-cli session + // started here has no copy anywhere". That was false for every coding-CLI session // this app spawned, and getting it wrong in that direction is the expensive one: // "this can't be undone", said of something that can, spends the credibility the // sentence needs. @@ -349,12 +351,12 @@ fun SessionListScreen( // the reassurance being read at the moment it stops being true. alsoDeleteForeign -> "Kills the process and deletes both copies of the conversation: " + - "this app's, and Claude Code's own transcript on the " + + "this app's, and $transcriptOwner's own transcript on the " + "machine. Nothing keeps another, so this can't be undone." else -> "Stops the process and deletes this app's copy of the " + "conversation, including any images, peer messages and " + - "commands recorded only here. Claude Code keeps its own " + + "commands recorded only here. $transcriptOwner keeps its own " + "transcript on the machine, so the conversation itself " + "should still be there to import again." } @@ -369,7 +371,7 @@ fun SessionListScreen( // line of text and re-centres whatever shares a row with it. Row(verticalAlignment = Alignment.CenterVertically) { Text( - "Delete Claude Code's transcript too", + "Delete $transcriptOwner's transcript too", style = MaterialTheme.typography.bodyMedium, modifier = Modifier.weight(1f), ) diff --git a/server/src/config.rs b/server/src/config.rs index c63d667..6936858 100644 --- a/server/src/config.rs +++ b/server/src/config.rs @@ -156,8 +156,7 @@ 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. + /// The Codex CLI's persistent app-server JSONL protocol. CodexCli, } @@ -235,9 +234,18 @@ impl DriverKind { /// the one sentence that has to be true: said of a session that can in fact /// be brought back, it spends the credibility the warning needs. pub fn keeps_own_transcript(self) -> bool { + self.own_transcript_name().is_some() + } + + /// The product whose durable transcript survives an ordinary app delete. + /// Reported to the phone because a provider's configured name is not the + /// name of its storage, and calling Codex's rollout a Claude transcript is + /// especially misleading on an irreversible switch. + pub fn own_transcript_name(self) -> Option<&'static str> { match self { - Self::ClaudeCli | Self::CodexCli => true, - Self::Echo | Self::LlamaCpp => false, + Self::ClaudeCli => Some("Claude Code"), + Self::CodexCli => Some("Codex"), + Self::Echo | Self::LlamaCpp => None, } } diff --git a/server/src/routes.rs b/server/src/routes.rs index b5f5477..89fd7cc 100644 --- a/server/src/routes.rs +++ b/server/src/routes.rs @@ -1130,8 +1130,8 @@ async fn spawn(manager: &Arc, body: SpawnRequest) -> Result String { tail_of(&kept) } -/// Creates the stdin fifo if it is not already there, and opens it read-write -/// for the process to inherit. -/// -/// Read-write is the whole trick: a fifo opened read-only delivers EOF as soon -/// as the last writer closes, so the process would exit the moment this server -/// did -- exactly what leaving it running has to prevent. Holding it open for -/// writing means the process is its own last writer. -fn make_fifo(path: &Path) -> Result { - if !path.exists() { - let c_path = std::ffi::CString::new(path.as_os_str().as_encoded_bytes()) - .with_context(|| format!("{} is not a usable path", path.display()))?; - // SAFETY: a nul-terminated path this call only reads, and a mode with - // no bits the kernel can object to. Owner-only, like everything else in - // a session directory: this carries what the person typed. - let made = unsafe { libc::mkfifo(c_path.as_ptr(), 0o600) }; - if made != 0 { - return Err(std::io::Error::last_os_error()) - .with_context(|| format!("creating the fifo {}", path.display())); - } - } - std::fs::OpenOptions::new() - .read(true) - .write(true) - .open(path) - .with_context(|| format!("opening the fifo {}", path.display())) -} - -/// A fresh, empty, owner-only log for one of the process's output streams. -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())) -} - pub(super) fn read_resume_token(session_dir: &Path) -> Option { let text = std::fs::read_to_string(session_dir.join(RESUME_FILE)).ok()?; serde_json::from_str::(&text) diff --git a/server/src/session/codex.rs b/server/src/session/codex.rs index 86feb02..7572397 100644 --- a/server/src/session/codex.rs +++ b/server/src/session/codex.rs @@ -1,23 +1,22 @@ -//! Codex CLI sessions over `codex exec --json`. +//! Codex CLI sessions over its persistent app-server protocol. //! -//! 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. +//! One app-server owns one conversation. Unlike `codex exec --json`, this +//! surface can interrupt a turn without killing the connection and can steer +//! an active turn through `turn/steer`. Its stdio is a fifo plus logs so both +//! the process and an in-flight turn survive an ai-server restart. 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 tokio::io::AsyncWriteExt; +use tokio::sync::mpsc; use super::driver::{AttachmentRef, Driver, Event, EventSink, SessionStatus, Unqueued}; use super::process; @@ -25,30 +24,63 @@ use super::transport::{Launch, Streams, Transport}; use crate::config::{ProviderConfig, SessionConfig}; use translate::Translator; +const STDIN_FIFO: &str = "codex-stdin.fifo"; 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 STATE_FILE: &str = "codex-state.json"; const POLL: std::time::Duration = std::time::Duration::from_millis(50); -#[derive(Clone, Serialize, Deserialize)] +#[derive(Clone, Default, Serialize, Deserialize)] struct Waiting { + /// The phone's queued-bubble id. Empty for the message that starts a turn. + #[serde(default)] id: String, + /// The id Codex echoes on its user-message item. + #[serde(default)] + client_id: String, text: String, + #[serde(default)] attachments: Vec, } +#[derive(Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +enum RequestKind { + Start, + Steer, +} + +#[derive(Clone, Serialize, Deserialize)] +struct PendingRequest { + id: String, + client_id: String, + kind: RequestKind, +} + #[derive(Default, Serialize, Deserialize)] -struct Queue { +struct ProtocolState { + #[serde(default)] waiting: VecDeque, + #[serde(default)] + sent: VecDeque, + #[serde(default)] + pending: Vec, + #[serde(default)] + initialize_request: Option, + #[serde(default)] + thread_request: Option, + #[serde(default)] + active_turn: Option, + #[serde(default)] + interrupt_when_started: bool, #[serde(skip)] running: bool, #[serde(skip)] closed: bool, - #[serde(skip)] - interrupting: bool, } +#[derive(Clone)] struct Settings { model: Option, permission_mode: Option, @@ -57,10 +89,9 @@ struct Settings { struct Inner { sink: EventSink, - queue: Mutex, + state: Mutex, settings: Mutex, - program: String, - cwd: Option, + to_child: mpsc::UnboundedSender, transport: Transport, session_dir: PathBuf, reading: AtomicBool, @@ -78,117 +109,157 @@ impl CodexDriver { session_dir: &Path, sink: EventSink, ) -> Result { - let mut queue = read_queue(session_dir); + let mut state = read_state(session_dir); let recorded = process::recorded(session_dir); - queue.running = matches!( - recorded, - Some((_, process::Liveness::Alive | process::Liveness::Unknown)) - ); + let (record, started_here) = match recorded { + Some((record, process::Liveness::Alive | process::Liveness::Unknown)) => { + state.running = state.active_turn.is_some() + || state + .pending + .iter() + .any(|request| request.kind == RequestKind::Start); + (record, false) + } + Some((_, process::Liveness::Dead)) | None => { + process::clear(session_dir); + // Requests written to a dead process have no recipient. Put their messages back + // in front of the unsent queue so restarting cannot silently lose them. + while let Some(message) = state.sent.pop_back() { + state.waiting.push_front(message); + } + state.pending.clear(); + state.initialize_request = None; + state.thread_request = None; + state.active_turn = None; + state.interrupt_when_started = false; + state.running = !state.waiting.is_empty(); + ( + start_process(meta, provider, &transport, session_dir)?, + true, + ) + } + }; + + let stdin = std::fs::OpenOptions::new() + .write(true) + .open(session_dir.join(STDIN_FIFO)) + .with_context(|| format!("opening {STDIN_FIFO} for session {}", meta.id))?; + let (to_child, mut from_driver) = mpsc::unbounded_channel::(); + tokio::spawn(async move { + let mut stdin = tokio::fs::File::from_std(stdin); + while let Some(line) = from_driver.recv().await { + if stdin.write_all(line.as_bytes()).await.is_err() + || stdin.write_all(b"\n").await.is_err() + || stdin.flush().await.is_err() + { + break; + } + } + }); + let inner = Arc::new(Inner { sink, - queue: Mutex::new(queue), + state: Mutex::new(state), 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(), + to_child, 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, - }); - } - } + if started_here { + begin_initialization(&inner); + let _ = inner.sink.send(Event::Status { + state: SessionStatus::Idle, + }); } + save_state(&inner); + spawn_follower(Arc::clone(&inner), record); 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 mut state = self.inner.state.lock().unwrap(); + if state.closed { + drop(state); 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(), + let id = state.running.then(super::random_hex).unwrap_or_default(); + state.running = true; + state.waiting.push_back(Waiting { + id: id.clone(), + client_id: format!("ai-app-{}", super::random_hex()), + text: text.clone(), + attachments: attachments.clone(), + }); + drop(state); + save_state(&self.inner); + if !id.is_empty() { + let _ = self.inner.sink.send(Event::MessageQueued { + id, 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); + dispatch_waiting(&self.inner); } 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 + let mut state = self.inner.state.lock().unwrap(); + if let Some(at) = state.waiting.iter().position(|message| message.id == id) { + state.waiting.remove(at); + drop(state); + save_state(&self.inner); + let _ = self + .inner + .sink + .send(Event::MessageDropped { id: id.to_string() }); + return Unqueued::Dropped; + } + if state.sent.iter().any(|message| message.id == id) { + Unqueued::AlreadySent + } else { + Unqueued::Unknown + } } 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(), + message: "Codex question answering is not available in this app yet".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); - } + let mut state = self.inner.state.lock().unwrap(); + let Some(thread_id) = read_thread(&self.inner.session_dir) else { + state.interrupt_when_started = state.running; + drop(state); + save_state(&self.inner); + return; + }; + let Some(turn_id) = state.active_turn.clone() else { + state.interrupt_when_started = state.running; + drop(state); + save_state(&self.inner); + return; + }; + state.interrupt_when_started = false; + drop(state); + save_state(&self.inner); + send_request( + &self.inner, + "turn/interrupt", + json!({"threadId": thread_id, "turnId": turn_id}), + ); } fn set_model(&self, model: &str) { @@ -207,18 +278,22 @@ impl Driver for CodexDriver { }); } - fn set_title(&self, _title: &str) {} + fn set_title(&self, title: &str) { + if let Some(thread_id) = read_thread(&self.inner.session_dir) { + send_request( + &self.inner, + "thread/name/set", + json!({"threadId": thread_id, "name": title}), + ); + } + } 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"), - }); + self.send_user_message(text.to_string(), Vec::new()); } fn compact(&self) { - let _ = self.inner.sink.send(Event::Error { - message: "Codex manages compaction itself in JSON exec sessions".to_string(), - }); + self.send_user_message("/compact".to_string(), Vec::new()); } fn clear(&self) { @@ -231,11 +306,19 @@ impl Driver for CodexDriver { }); return; } + let request = request_id(); + self.inner.state.lock().unwrap().thread_request = Some(request.clone()); + save_state(&self.inner); + send_json( + &self.inner, + json!({"id": request, "method": "thread/start", "params": thread_params(&self.inner)}), + ); let _ = self.inner.sink.send(Event::Cleared); } fn between_turns(&self) -> bool { - !self.inner.queue.lock().unwrap().running + let state = self.inner.state.lock().unwrap(); + !state.running && !state.closed } fn detach(&self) { @@ -245,15 +328,16 @@ impl Driver for CodexDriver { 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) + let mut state = self.inner.state.lock().unwrap(); + state.closed = true; + let mut messages: Vec<_> = state.waiting.drain(..).collect(); + messages.extend(state.sent.drain(..)); + messages + .into_iter() + .filter_map(|message| (!message.id.is_empty()).then_some(message.id)) .collect::>() }; - save_queue(&self.inner); + save_state(&self.inner); for id in dropped { let _ = self.inner.sink.send(Event::MessageDropped { id }); } @@ -264,118 +348,206 @@ impl Driver for CodexDriver { } } -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], + meta: &SessionConfig, + provider: &ProviderConfig, + transport: &Transport, + session_dir: &Path, ) -> Result { - let settings = inner.settings.lock().unwrap(); - let mut args = Vec::new(); - match settings.permission_mode.as_deref() { - Some("bypassPermissions" | "danger-full-access") => { - 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" | "workspace-write") => { - args.push("--approve-for-me".to_string()); - } - Some("plan" | "read-only") => 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( + let stdin = process::make_fifo(&session_dir.join(STDIN_FIFO))?; + let stdout = process::create_log(&session_dir.join(STDOUT_LOG))?; + let stderr = process::create_log(&session_dir.join(STDERR_LOG))?; + let launch = Launch::new( + provider.program(), + vec!["app-server".to_string(), "--stdio".to_string()], + meta.cwd.as_deref(), + ); + let mut child = transport.spawn( &launch, Streams::Detached { - stdin: Stdio::null(), + stdin: stdin.into(), stdout: stdout.into(), stderr: stderr.into(), }, )?; let pid = child .id() - .context("Codex exited before it could be recorded")?; + .context("Codex app-server 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); + .context("Codex app-server was gone before its start time could be read")?; + process::write(session_dir, &record); Ok(record) } +fn begin_initialization(inner: &Arc) { + let id = request_id(); + inner.state.lock().unwrap().initialize_request = Some(id.clone()); + save_state(inner); + send_json( + inner, + json!({ + "id": id, + "method": "initialize", + "params": {"clientInfo": { + "name": "ai-app", + "title": "AI Sessions", + "version": env!("CARGO_PKG_VERSION") + }} + }), + ); +} + +fn send_json(inner: &Inner, value: Value) { + let _ = inner.to_child.send(value.to_string()); +} + +fn send_request(inner: &Inner, method: &str, params: Value) -> String { + let id = request_id(); + send_json(inner, json!({"id": id, "method": method, "params": params})); + id +} + +fn request_id() -> String { + format!("ai-app-{}", super::random_hex()) +} + +fn thread_params(inner: &Inner) -> Value { + let settings = inner.settings.lock().unwrap().clone(); + let mut params = json!({"approvalPolicy": "never"}); + if let Some(model) = settings.model { + params["model"] = Value::String(model); + } + if let Some(mode) = settings.permission_mode { + params["sandbox"] = Value::String(mode); + } + params +} + +fn turn_params(inner: &Inner, thread_id: &str, message: &Waiting) -> Result { + let settings = inner.settings.lock().unwrap().clone(); + let mut params = json!({ + "threadId": thread_id, + "input": input_for(inner, message)?, + "clientUserMessageId": message.client_id, + }); + if let Some(model) = settings.model { + params["model"] = Value::String(model); + } + if let Some(effort) = settings.effort { + params["effort"] = Value::String(effort); + } + if let Some(mode) = settings.permission_mode { + params["sandboxPolicy"] = sandbox_policy(&mode); + params["approvalPolicy"] = Value::String("never".to_string()); + } + Ok(params) +} + +fn sandbox_policy(mode: &str) -> Value { + match mode { + "danger-full-access" => json!({"type": "dangerFullAccess"}), + "read-only" => json!({"type": "readOnly"}), + _ => json!({"type": "workspaceWrite"}), + } +} + +fn input_for(inner: &Inner, message: &Waiting) -> Result> { + let mut text = message.text.clone(); + let mut input = Vec::new(); + for attachment in &message.attachments { + let path = attachment_path(&inner.session_dir, attachment)?; + if crate::media::media_type_for(attachment).is_some() + && matches!(inner.transport, Transport::Here) + { + input.push(json!({"type": "localImage", "path": path})); + } else { + if !text.is_empty() { + text.push_str("\n\n"); + } + text.push_str(&format!("Attached file: {}", path.display())); + } + } + if !text.is_empty() { + input.insert(0, json!({"type": "text", "text": text})); + } + Ok(input) +} + +fn dispatch_waiting(inner: &Arc) { + let Some(thread_id) = read_thread(&inner.session_dir) else { + return; + }; + loop { + let (message, kind, active_turn) = { + let mut state = inner.state.lock().unwrap(); + if state.closed || state.waiting.is_empty() { + return; + } + if state.active_turn.is_none() + && state + .pending + .iter() + .any(|request| request.kind == RequestKind::Start) + { + return; + } + let kind = if state.active_turn.is_some() { + RequestKind::Steer + } else { + RequestKind::Start + }; + let mut message = state.waiting.pop_front().unwrap(); + // Queues written by the old per-turn driver predate app-server's + // client id. Give an adopted message one before Codex sees it so + // its eventual user item can still resolve the queued bubble. + if message.client_id.is_empty() { + message.client_id = format!("ai-app-{}", super::random_hex()); + } + let active_turn = state.active_turn.clone(); + (message, kind, active_turn) + }; + let mut params = match turn_params(inner, &thread_id, &message) { + Ok(params) => params, + Err(err) => { + if !message.id.is_empty() { + let _ = inner.sink.send(Event::MessageDropped { + id: message.id.clone(), + }); + } + let _ = inner.sink.send(Event::Error { + message: format!("couldn't send an attachment to Codex: {err:#}"), + }); + continue; + } + }; + let method = match kind { + RequestKind::Start => "turn/start", + RequestKind::Steer => { + params["expectedTurnId"] = Value::String(active_turn.unwrap()); + "turn/steer" + } + }; + let id = request_id(); + { + let mut state = inner.state.lock().unwrap(); + state.sent.push_back(message.clone()); + state.pending.push(PendingRequest { + id: id.clone(), + client_id: message.client_id.clone(), + kind, + }); + } + save_state(inner); + send_json(inner, json!({"id": id, "method": method, "params": params})); + if kind == RequestKind::Start { + return; + } + } +} + fn spawn_follower(inner: Arc, record: process::Record) { let offset = match record.detail { process::Detail::Stdio { stdout_read } => stdout_read, @@ -411,15 +583,7 @@ async fn follow(inner: Arc, mut record: process::Record, mut offset: u64) ); 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); - } + handle_line(&inner, &mut translator, &value); } if complete > 0 { offset += complete as u64; @@ -429,35 +593,34 @@ async fn follow(inner: Arc, mut record: process::Record, mut offset: u64) process::write(&inner.session_dir, &record); } match record.liveness() { - process::Liveness::Alive => {} - process::Liveness::Unknown => {} + 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 dropped = { + let mut state = inner.state.lock().unwrap(); + state.closed = true; + state.running = false; + let mut messages: Vec<_> = state.waiting.drain(..).collect(); + messages.extend(state.sent.drain(..)); + messages + .into_iter() + .filter_map(|message| (!message.id.is_empty()).then_some(message.id)) + .collect::>() + }; + save_state(&inner); + for id in dropped { + let _ = inner.sink.send(Event::MessageDropped { id }); + } + let detail = stderr_tail(&stderr); + if !detail.is_empty() { 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}") - }, + message: 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); - } + let _ = inner.sink.send(Event::Status { + state: SessionStatus::Exited, + }); return; } } @@ -465,14 +628,203 @@ async fn follow(inner: Arc, mut record: process::Record, mut offset: u64) } } -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 handle_line(inner: &Arc, translator: &mut Translator, line: &Value) { + if line.get("id").is_some() { + handle_response(inner, line); + } + let method = line.get("method").and_then(Value::as_str); + let params = line.get("params").unwrap_or(&Value::Null); + match method { + Some("thread/started") => { + if let Some(thread) = params.pointer("/thread/id").and_then(Value::as_str) { + write_thread(&inner.session_dir, thread); + } + } + Some("turn/started") => { + if let Some(turn) = params.pointer("/turn/id").and_then(Value::as_str) { + let interrupt = { + let mut state = inner.state.lock().unwrap(); + state.active_turn = Some(turn.to_string()); + state.running = true; + std::mem::take(&mut state.interrupt_when_started) + }; + save_state(inner); + if interrupt { + if let Some(thread) = read_thread(&inner.session_dir) { + send_request( + inner, + "turn/interrupt", + json!({"threadId": thread, "turnId": turn}), + ); + } + } else { + dispatch_waiting(inner); + } + } + } + Some("item/completed") => { + let item = ¶ms["item"]; + if item.get("type").and_then(Value::as_str) == Some("userMessage") { + announce_user(inner, item); + } + } + Some("turn/completed") => { + let mut state = inner.state.lock().unwrap(); + state.active_turn = None; + state.running = !state.waiting.is_empty(); + state.interrupt_when_started = false; + drop(state); + save_state(inner); + } + _ => {} + } + for event in translator.translate(line) { + let _ = inner.sink.send(event); + } + if method == Some("turn/completed") { + dispatch_waiting(inner); + } +} + +fn handle_response(inner: &Arc, line: &Value) { + let Some(id) = line.get("id").and_then(Value::as_str) else { + return; + }; + let mut state = inner.state.lock().unwrap(); + if state.initialize_request.as_deref() == Some(id) { + state.initialize_request = None; + if let Some(error) = response_error(line) { + drop(state); + protocol_error(inner, error); + return; + } + let request = request_id(); + state.thread_request = Some(request.clone()); + drop(state); + save_state(inner); + send_json(inner, json!({"method": "initialized"})); + let mut params = thread_params(inner); + let method = match read_thread(&inner.session_dir) { + Some(thread) => { + params["threadId"] = Value::String(thread); + "thread/resume" + } + None => "thread/start", + }; + send_json( + inner, + json!({"id": request, "method": method, "params": params}), + ); + return; + } + if state.thread_request.as_deref() == Some(id) { + state.thread_request = None; + if let Some(error) = response_error(line) { + drop(state); + protocol_error(inner, error); + return; + } + let thread = line.pointer("/result/thread/id").and_then(Value::as_str); + drop(state); + if let Some(thread) = thread { + write_thread(&inner.session_dir, thread); + } + save_state(inner); + dispatch_waiting(inner); + return; + } + let Some(at) = state.pending.iter().position(|request| request.id == id) else { + return; + }; + let request = state.pending.remove(at); + if let Some(error) = response_error(line) { + let message = state + .sent + .iter() + .position(|message| message.client_id == request.client_id) + .and_then(|at| state.sent.remove(at)); + if request.kind == RequestKind::Steer && active_turn_not_steerable(line) { + if let Some(message) = message { + state.waiting.push_back(message); + } + let between_turns = state.active_turn.is_none(); + drop(state); + save_state(inner); + if between_turns { + dispatch_waiting(inner); + } + return; + } + state.running = state.active_turn.is_some() + || state + .pending + .iter() + .any(|pending| pending.kind == RequestKind::Start) + || !state.waiting.is_empty(); + let idle = !state.running; + drop(state); + save_state(inner); + if let Some(message) = message + && !message.id.is_empty() + { + let _ = inner.sink.send(Event::MessageDropped { id: message.id }); + } + protocol_error(inner, error); + if idle { + let _ = inner.sink.send(Event::Status { + state: SessionStatus::Idle, + }); + } + return; + } + if request.kind == RequestKind::Start + && let Some(turn) = line.pointer("/result/turn/id").and_then(Value::as_str) + { + state.active_turn = Some(turn.to_string()); + } + drop(state); + save_state(inner); +} + +fn announce_user(inner: &Inner, item: &Value) { + let Some(client_id) = item.get("clientId").and_then(Value::as_str) else { + return; + }; + let message = { + let mut state = inner.state.lock().unwrap(); + state + .sent + .iter() + .position(|message| message.client_id == client_id) + .and_then(|at| state.sent.remove(at)) + }; + let Some(message) = message else { + return; + }; + save_state(inner); + let _ = inner.sink.send(Event::MessageTaken { + id: (!message.id.is_empty()).then_some(message.id), + text: message.text, + attachments: message.attachments, + }); +} + +fn response_error(line: &Value) -> Option { + line.pointer("/error/message") + .and_then(Value::as_str) + .map(str::to_string) +} + +fn active_turn_not_steerable(line: &Value) -> bool { + line.pointer("/error/data/codexErrorInfo/activeTurnNotSteerable") + .is_some() + || line.to_string().contains("activeTurnNotSteerable") +} + +fn protocol_error(inner: &Inner, message: String) { + let _ = inner.sink.send(Event::Error { + message: format!("Codex refused a request: {message}"), + }); } fn attachment_path(session_dir: &Path, id: &str) -> Result { @@ -486,7 +838,7 @@ fn attachment_path(session_dir: &Path, id: &str) -> Result { Ok(session_dir.join("attachments").join(id)) } -fn read_thread(session_dir: &Path) -> Option { +pub(super) 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")? @@ -496,17 +848,7 @@ fn read_thread(session_dir: &Path) -> Option { 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 { + if let Err(err) = std::fs::write(&path, json!({"threadId": thread}).to_string()) { tracing::error!( "couldn't persist Codex thread id to {}: {err}", path.display() @@ -514,37 +856,40 @@ fn write_thread(session_dir: &Path, thread: &str) { } } -fn read_queue(session_dir: &Path) -> Queue { - std::fs::read_to_string(session_dir.join(QUEUE_FILE)) +fn read_state(session_dir: &Path) -> ProtocolState { + if let Some(state) = std::fs::read_to_string(session_dir.join(STATE_FILE)) .ok() .and_then(|text| serde_json::from_str(&text).ok()) + { + return state; + } + // Upgrade messages left by the old per-turn exec driver. + std::fs::read_to_string(session_dir.join("codex-queue.json")) + .ok() + .and_then(|text| serde_json::from_str::(&text).ok()) + .and_then(|value| serde_json::from_value(value["waiting"].clone()).ok()) + .map(|waiting| ProtocolState { + waiting, + ..ProtocolState::default() + }) .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()) { +fn save_state(inner: &Inner) { + let path = inner.session_dir.join(STATE_FILE); + let text = match serde_json::to_string(&*inner.state.lock().unwrap()) { Ok(text) => text, Err(err) => { - tracing::error!("couldn't serialize the Codex queue: {err}"); + tracing::error!("couldn't serialize the Codex protocol state: {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)); + let written = + std::fs::write(&temporary, text).and_then(|()| std::fs::rename(&temporary, &path)); if let Err(err) = written { tracing::error!( - "couldn't persist the Codex queue to {}: {err}", + "couldn't persist Codex protocol state to {}: {err}", path.display() ); let _ = std::fs::remove_file(temporary); @@ -563,3 +908,81 @@ fn stderr_tail(path: &Path) -> String { .collect::>() .join("\n") } + +const DELETE_TRANSCRIPT_SCRIPT: &str = r#" +state=missing +for f in "$HOME"/.codex/sessions/*/*/*/rollout-*-${1}.jsonl; do + [ -f "$f" ] || continue + if rm -f "$f"; then state=deleted; else state=failed; fi +done +printf '%s\n' "$state" +"#; + +/// Removes the rollout whose suffix is this thread id. +pub async fn delete_transcript(transport: &Transport, id: &str) -> Result<()> { + if !valid_thread_id(id) { + anyhow::bail!("not a Codex thread id: {id}"); + } + let launch = Launch::new( + "sh", + vec![ + "-c".to_string(), + DELETE_TRANSCRIPT_SCRIPT.to_string(), + "sh".to_string(), + id.to_string(), + ], + None, + ); + match transport.capture(&launch).await?.trim() { + "deleted" => Ok(()), + "missing" => anyhow::bail!("no Codex thread {id} on that machine"), + _ => anyhow::bail!("couldn't remove Codex thread {id}"), + } +} + +fn valid_thread_id(id: &str) -> bool { + !id.is_empty() + && id + .chars() + .all(|character| character.is_ascii_hexdigit() || character == '-') +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn old_exec_queues_are_adopted() { + let dir = tempfile::tempdir().expect("tempdir"); + std::fs::write( + dir.path().join("codex-queue.json"), + r#"{"waiting":[{"id":"q1","text":"next","attachments":[]}]}"#, + ) + .expect("queue"); + let state = read_state(dir.path()); + assert_eq!(state.waiting.len(), 1); + assert_eq!(state.waiting[0].id, "q1"); + } + + #[test] + fn transcript_delete_resolves_only_the_named_codex_rollout() { + use std::process::Command; + + let home = tempfile::tempdir().expect("home"); + let sessions = home.path().join(".codex/sessions/2026/09/09"); + std::fs::create_dir_all(&sessions).expect("sessions"); + let wanted = sessions.join("rollout-2026-09-09T12-00-00-abcd-1234.jsonl"); + let other = sessions.join("rollout-2026-09-09T12-00-01-abcd-5678.jsonl"); + std::fs::write(&wanted, "wanted").expect("wanted"); + std::fs::write(&other, "other").expect("other"); + let output = Command::new("sh") + .args(["-c", DELETE_TRANSCRIPT_SCRIPT, "sh", "abcd-1234"]) + .env("HOME", home.path()) + .output() + .expect("delete script"); + assert_eq!(String::from_utf8_lossy(&output.stdout).trim(), "deleted"); + assert!(!wanted.exists()); + assert!(other.exists()); + assert!(!valid_thread_id("../../something")); + } +} diff --git a/server/src/session/codex/translate.rs b/server/src/session/codex/translate.rs index c8440b8..2382026 100644 --- a/server/src/session/codex/translate.rs +++ b/server/src/session/codex/translate.rs @@ -1,9 +1,11 @@ -//! `codex exec --json` lines into the common event model. +//! Codex app-server notifications 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 std::collections::HashSet; + use serde_json::{Value, json}; use super::super::driver::{Event, SessionStatus}; @@ -13,11 +15,16 @@ pub(super) struct Translator { pub(super) thread_id: Option, completed: bool, limited: bool, + streamed_messages: HashSet, + pending_usage: Option, } impl Translator { pub(super) fn translate(&mut self, line: &Value) -> Vec { - match line.get("type").and_then(Value::as_str) { + let method = line.get("method").and_then(Value::as_str); + let body = method.and_then(|_| line.get("params")).unwrap_or(line); + let kind = method.or_else(|| line.get("type").and_then(Value::as_str)); + match kind { Some("thread.started") => { self.thread_id = line .get("thread_id") @@ -25,16 +32,65 @@ impl Translator { .map(str::to_string); Vec::new() } - Some("turn.started") => vec![Event::Status { - state: SessionStatus::Running, - }], - Some("item.started") => start_item(&line["item"]), + Some("thread/started") => { + self.thread_id = body + .pointer("/thread/id") + .and_then(Value::as_str) + .map(str::to_string); + Vec::new() + } + Some("turn.started") | Some("turn/started") => { + self.completed = false; + self.limited = false; + self.pending_usage = None; + self.streamed_messages.clear(); + vec![Event::Status { + state: SessionStatus::Running, + }] + } + Some("item.started") | Some("item/started") => start_item(&body["item"]), Some("item.updated") => update_item(&line["item"]), - Some("item.completed") => complete_item(&line["item"]), - Some("turn.completed") => { + Some("item.completed") | Some("item/completed") => { + complete_item(&body["item"], &self.streamed_messages) + } + Some("item/agentMessage/delta") => { + let Some(delta) = body.get("delta").and_then(Value::as_str) else { + return Vec::new(); + }; + if let Some(id) = body.get("itemId").and_then(Value::as_str) { + self.streamed_messages.insert(id.to_string()); + } + vec![Event::AssistantText { + delta: delta.to_string(), + }] + } + Some("item/commandExecution/outputDelta") => { + let (Some(id), Some(output)) = ( + body.get("itemId").and_then(Value::as_str), + body.get("delta").and_then(Value::as_str), + ) else { + return Vec::new(); + }; + vec![Event::ToolUpdate { + id: id.to_string(), + output: output.to_string(), + }] + } + Some("thread/tokenUsage/updated") => { + self.pending_usage = body + .pointer("/tokenUsage/last/totalTokens") + .and_then(Value::as_u64); + Vec::new() + } + Some("turn.completed") | Some("turn/completed") => { self.completed = true; let mut events = Vec::new(); - if let Some(usage) = line.get("usage") { + if let Some(tokens) = self.pending_usage.take() { + events.push(Event::UsageDelta { + tokens, + context: None, + }); + } else 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() { @@ -46,20 +102,27 @@ impl Translator { }); } } + if let Some(error) = body.pointer("/turn/error") + && !error.is_null() + { + events.extend(self.failure(error)); + } events.push(Event::Status { state: SessionStatus::Idle, }); events } - Some("turn.failed") | Some("error") => self.failure(line), + Some("turn.failed") | Some("error") => self.failure(body), _ => Vec::new(), } } + #[cfg(test)] pub(super) fn completed(&self) -> bool { self.completed } + #[cfg(test)] pub(super) fn limited(&self) -> bool { self.limited } @@ -106,19 +169,24 @@ fn update_item(item: &Value) -> Vec { .unwrap_or_default() } -fn complete_item(item: &Value) -> Vec { +fn complete_item(item: &Value, streamed_messages: &HashSet) -> Vec { match item.get("type").and_then(Value::as_str) { - Some("agent_message") => item + Some("agent_message" | "agentMessage") => item .get("text") .and_then(Value::as_str) .filter(|text| !text.is_empty()) + .filter(|_| { + item.get("id") + .and_then(Value::as_str) + .is_none_or(|id| !streamed_messages.contains(id)) + }) .map(|delta| { vec![Event::AssistantText { delta: delta.to_string(), }] }) .unwrap_or_default(), - Some("reasoning") => Vec::new(), + Some("reasoning" | "userMessage") => Vec::new(), _ => { let Some((id, _, _)) = tool(item) else { return Vec::new(); @@ -133,15 +201,15 @@ 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" => ( + "command_execution" | "commandExecution" => ( "exec_command".to_string(), json!({"command": item.get("command").cloned().unwrap_or(Value::Null)}), ), - "file_change" => ( + "file_change" | "fileChange" => ( "apply_patch".to_string(), item.get("changes").cloned().unwrap_or(Value::Null), ), - "mcp_tool_call" => ( + "mcp_tool_call" | "mcpToolCall" => ( item.get("tool") .or_else(|| item.get("name")) .and_then(Value::as_str) @@ -149,18 +217,24 @@ fn tool(item: &Value) -> Option<(String, String, Value)> { .unwrap_or_else(|| "mcp".to_string()), item.get("arguments").cloned().unwrap_or(Value::Null), ), - "web_search" => ( + "web_search" | "webSearch" => ( "web_search".to_string(), json!({"query": item.get("query").cloned().unwrap_or(Value::Null)}), ), - "todo_list" => ("update_plan".to_string(), item.clone()), + "todo_list" | "todoList" | "plan" => ("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"] { + for key in [ + "aggregated_output", + "aggregatedOutput", + "output", + "result", + "error", + ] { if let Some(text) = item.get(key).and_then(value_text) && !text.is_empty() { @@ -276,4 +350,53 @@ mod tests { ); assert!(translator.limited()); } + + #[test] + fn translates_native_app_server_streaming_without_repeating_the_final_item() { + let mut translator = Translator::default(); + assert_eq!( + translator.translate(&line( + r#"{"method":"turn/started","params":{"threadId":"thread-1","turn":{"id":"turn-1"}}}"# + )), + vec![Event::Status { + state: SessionStatus::Running + }] + ); + assert_eq!( + translator.translate(&line( + r#"{"method":"item/agentMessage/delta","params":{"itemId":"message-1","delta":"hello"}}"# + )), + vec![Event::AssistantText { + delta: "hello".to_string() + }] + ); + assert!( + translator + .translate(&line( + r#"{"method":"item/completed","params":{"item":{"id":"message-1","type":"agentMessage","text":"hello"}}}"# + )) + .is_empty() + ); + assert!( + translator + .translate(&line( + r#"{"method":"thread/tokenUsage/updated","params":{"tokenUsage":{"last":{"totalTokens":42}}}}"# + )) + .is_empty() + ); + assert_eq!( + translator.translate(&line( + r#"{"method":"turn/completed","params":{"turn":{"id":"turn-1","status":"completed","error":null}}}"# + )), + vec![ + Event::UsageDelta { + tokens: 42, + context: None + }, + Event::Status { + state: SessionStatus::Idle + } + ] + ); + } } diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index 5714cfa..b07e997 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -205,6 +205,9 @@ pub struct SessionInfo { /// Reported rather than worked out on the phone, because the phone has /// the provider's *name* and this is a property of its *kind*. pub keeps_own_transcript: bool, + /// The CLI whose copy the delete dialog can optionally remove. + #[serde(skip_serializing_if = "Option::is_none")] + pub own_transcript_name: Option<&'static str>, /// How much this session asks before acting. Reported so the phone can /// *show* the current mode rather than assume one -- a picker that /// guesses its own value is how you change something you thought you @@ -600,6 +603,7 @@ impl LiveSession { usage_provider: kind.and_then(DriverKind::usage_provider), imported, keeps_own_transcript: kind.is_some_and(DriverKind::keeps_own_transcript), + own_transcript_name: kind.and_then(DriverKind::own_transcript_name), cwd: cwd.map(Path::to_path_buf), status: *self.shared.status.lock().unwrap(), last_activity: *self.shared.last_activity.lock().unwrap(), @@ -625,9 +629,8 @@ pub struct SessionManager { /// Where every session's pump reports what this layer does not act on -- /// see [`Announcements`]. announce: Announcements, - /// Imports and deletes running against a machine's Claude Code - /// sessions: like the notifications, state the phone reads but does not - /// own. + /// Provider-transcript operations running against a machine: like the + /// notifications, state the phone reads but does not own. pending: Arc, /// What to mark sessions spawned here as -- see /// [`SessionManager::marking_new_sessions_throwaway`]. @@ -640,6 +643,38 @@ pub struct SessionManager { inner: RwLock, } +/// The CLI-owned copy optionally removed with an ai-app session. +#[derive(Debug, PartialEq, Eq)] +pub struct ForeignTranscript { + pub setup: String, + pub id: String, + kind: DriverKind, +} + +impl ForeignTranscript { + /// Removes the provider's own durable record. The route remains generic: + /// adding another transcript-owning driver extends this provider boundary. + pub async fn delete(&self, transport: &Transport) -> Result<()> { + match self.kind { + DriverKind::ClaudeCli => import::delete(transport, std::slice::from_ref(&self.id)) + .await? + .remove(&self.id) + .unwrap_or_else(|| Err(format!("nothing was reported about {}", self.id))) + .map_err(anyhow::Error::msg), + DriverKind::CodexCli => codex::delete_transcript(transport, &self.id).await, + DriverKind::Echo | DriverKind::LlamaCpp => { + anyhow::bail!("this provider keeps no transcript of its own") + } + } + } + + pub fn owner_name(&self) -> &'static str { + self.kind + .own_transcript_name() + .expect("a foreign transcript has an owner") + } +} + impl SessionManager { /// Loads the config and brings every persisted session back: its /// transcript, its pump, and the process it left running where it left @@ -1034,15 +1069,26 @@ impl SessionManager { Some((setup.ssh.clone()?, meta.cwd.clone())) } - pub fn foreign_transcript(&self, id: &str) -> Option<(String, String)> { + pub fn foreign_transcript(&self, id: &str) -> Option { let inner = self.inner.read().unwrap(); let meta = inner.config.sessions.iter().find(|meta| meta.id == id)?; - let (followed, resuming) = foreign_ids(&self.data_dir.join(&meta.id)); - // The cursor first: an imported session follows a file that exists - // whether or not a CLI has resumed it yet. - followed - .or(resuming) - .map(|foreign| (meta.setup.clone(), foreign)) + let kind = kind_of(&inner.config, &meta.setup, &meta.provider)?; + let session_dir = self.data_dir.join(&meta.id); + let foreign = match kind { + DriverKind::ClaudeCli => { + let (followed, resuming) = foreign_ids(&session_dir); + // The cursor first: an imported session follows a file that exists + // whether or not a CLI has resumed it yet. + followed.or(resuming) + } + DriverKind::CodexCli => codex::read_thread(&session_dir), + DriverKind::Echo | DriverKind::LlamaCpp => None, + }?; + Some(ForeignTranscript { + setup: meta.setup.clone(), + id: foreign, + kind, + }) } /// Every session, in config order, with live status joined in. A @@ -1088,6 +1134,8 @@ impl SessionManager { &meta.setup, &meta.provider, ), + own_transcript_name: kind_of(&inner.config, &meta.setup, &meta.provider) + .and_then(DriverKind::own_transcript_name), cwd: meta.cwd.clone(), status: status_of_unlaunched(&self.data_dir.join(&meta.id)), last_activity: meta.created, @@ -2804,6 +2852,8 @@ mod tests { #[test] fn only_a_driver_that_keeps_its_own_record_survives_deletion() { assert!(DriverKind::ClaudeCli.keeps_own_transcript()); + assert!(DriverKind::CodexCli.keeps_own_transcript()); + assert_eq!(DriverKind::CodexCli.own_transcript_name(), Some("Codex")); assert!(!DriverKind::Echo.keeps_own_transcript()); assert!(!DriverKind::LlamaCpp.keeps_own_transcript()); } @@ -3268,14 +3318,17 @@ mod tests { let dir = tempfile::tempdir().expect("tempdir"); let config_path = dir.path().join("config.ron"); let data_dir = dir.path().join("sessions"); - seed_echo_only(&config_path); + let provider = seed_stand_in_cli(&config_path, dir.path()); let manager = SessionManager::new( config_path.clone(), data_dir.clone(), data_dir.join("models"), ) - .expect("manager"); - let info = manager.spawn_session(echo_spec()).expect("spawn"); + .expect("manager") + .marking_new_sessions_throwaway(true); + let info = manager + .spawn_session(stand_in_spec(&provider)) + .expect("spawn"); // Nothing recorded yet, so there is nothing a delete would reach. assert_eq!(manager.foreign_transcript(&info.id), None); @@ -3283,7 +3336,11 @@ mod tests { claude::write_resume_token(&data_dir.join(&info.id), "5ecf21da-d53f"); assert_eq!( manager.foreign_transcript(&info.id), - Some((info.setup.clone(), "5ecf21da-d53f".to_string())) + Some(ForeignTranscript { + setup: info.setup.clone(), + id: "5ecf21da-d53f".to_string(), + kind: DriverKind::ClaudeCli, + }) ); // A session that is not there has no transcript to name, rather diff --git a/server/src/session/process.rs b/server/src/session/process.rs index 202b1a2..e3b0539 100644 --- a/server/src/session/process.rs +++ b/server/src/session/process.rs @@ -187,6 +187,39 @@ pub fn clear(session_dir: &Path) { } } +/// Creates a session stdin fifo and opens it read-write for the child. +/// +/// The child holding the write end is what keeps a detached JSON server from +/// reading EOF when ai-server restarts and temporarily closes its own writer. +pub fn make_fifo(path: &Path) -> Result { + if !path.exists() { + let c_path = std::ffi::CString::new(path.as_os_str().as_encoded_bytes()) + .with_context(|| format!("{} is not a usable path", path.display()))?; + // SAFETY: `c_path` is nul-terminated and this call only reads it. + let made = unsafe { libc::mkfifo(c_path.as_ptr(), 0o600) }; + if made != 0 { + return Err(std::io::Error::last_os_error()) + .with_context(|| format!("creating the fifo {}", path.display())); + } + } + std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(path) + .with_context(|| format!("opening the fifo {}", path.display())) +} + +/// A fresh owner-only log for a detached session process. +pub 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())) +} + /// Grace period between asking a session's process to stop and killing it. /// Here rather than beside each caller: two drivers plus the manager had /// written the same five seconds down separately.