Add Codex JSON sessions and usage limits
This commit is contained in:
iris committed
1 parent
0862b47f76
commit
6a0202b1b5
14 files changed
+1173
-52
No files matched your search
@@ -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 <id> --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<AttachmentRef>,
|
||||
}
|
||||
|
||||
#[derive(Default, Serialize, Deserialize)]
|
||||
struct Queue {
|
||||
waiting: VecDeque<Waiting>,
|
||||
#[serde(skip)]
|
||||
running: bool,
|
||||
#[serde(skip)]
|
||||
closed: bool,
|
||||
#[serde(skip)]
|
||||
interrupting: bool,
|
||||
}
|
||||
|
||||
struct Settings {
|
||||
model: Option<String>,
|
||||
permission_mode: Option<String>,
|
||||
effort: Option<String>,
|
||||
}
|
||||
|
||||
struct Inner {
|
||||
sink: EventSink,
|
||||
queue: Mutex<Queue>,
|
||||
settings: Mutex<Settings>,
|
||||
program: String,
|
||||
cwd: Option<PathBuf>,
|
||||
transport: Transport,
|
||||
session_dir: PathBuf,
|
||||
reading: AtomicBool,
|
||||
}
|
||||
|
||||
pub struct CodexDriver {
|
||||
inner: Arc<Inner>,
|
||||
}
|
||||
|
||||
impl CodexDriver {
|
||||
pub fn launch(
|
||||
meta: &SessionConfig,
|
||||
provider: &ProviderConfig,
|
||||
transport: Transport,
|
||||
session_dir: &Path,
|
||||
sink: EventSink,
|
||||
) -> Result<Self> {
|
||||
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<AttachmentRef>) {
|
||||
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::<Vec<_>>()
|
||||
};
|
||||
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<Inner>, 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<process::Record> {
|
||||
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<Inner>, 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<Inner>, 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::<Value>(line) else {
|
||||
tracing::warn!(
|
||||
"unparseable Codex JSONL line: {}",
|
||||
line.chars().take(200).collect::<String>()
|
||||
);
|
||||
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::File> {
|
||||
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<PathBuf> {
|
||||
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<String> {
|
||||
serde_json::from_str::<Value>(&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::<Vec<_>>()
|
||||
.into_iter()
|
||||
.rev()
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n")
|
||||
}
|
||||
@@ -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<String>,
|
||||
completed: bool,
|
||||
limited: bool,
|
||||
}
|
||||
|
||||
impl Translator {
|
||||
pub(super) fn translate(&mut self, line: &Value) -> Vec<Event> {
|
||||
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<Event> {
|
||||
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<Event> {
|
||||
let Some((id, tool, input)) = tool(item) else {
|
||||
return Vec::new();
|
||||
};
|
||||
vec![Event::ToolStart { id, tool, input }]
|
||||
}
|
||||
|
||||
fn update_item(item: &Value) -> Vec<Event> {
|
||||
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<Event> {
|
||||
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<String> {
|
||||
value
|
||||
.as_str()
|
||||
.map(str::to_string)
|
||||
.or_else(|| (!value.is_null()).then(|| value.to_string()))
|
||||
}
|
||||
|
||||
fn number(value: &Value, key: &str) -> Option<u64> {
|
||||
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<f64> {
|
||||
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());
|
||||
}
|
||||
}
|
||||
@@ -498,9 +498,9 @@ pub type EventSink = mpsc::UnboundedSender<Event>;
|
||||
/// 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.
|
||||
///
|
||||
|
||||
@@ -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(),
|
||||
)?),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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<serde_json::Value> {
|
||||
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::<serde_json::Value>(&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::<serde_json::Value>(&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 {
|
||||
|
||||
Reference in new issue
Block a user