use std::path::PathBuf;
use std::time::{Duration, Instant};
use crate::envelope::{Envelope, JsonEnvelope, NoEnvelope};
use crate::readiness::{
AgyReadinessDetector, CodexReadinessDetector, GeminiReadinessDetector,
OpencodeReadinessDetector, ReadinessDetector,
};
use crate::supervisor::RestartPolicy;
use crate::ParsedEvent;
#[derive(Debug, Clone)]
pub struct CreateContext {
pub name: String,
pub message: String,
pub cwd: PathBuf,
pub from_name: Option<String>,
pub session_id: Option<String>,
pub yolo: bool,
pub reasoning_effort: Option<String>,
pub append_system_prompt: Option<String>,
}
#[derive(Debug, Clone)]
pub struct ResumeContext {
pub session_id: String,
pub message: String,
pub cwd: PathBuf,
pub from_name: Option<String>,
pub yolo: bool,
}
#[derive(Debug, Clone)]
pub struct AgentEntry {
pub name: String,
pub provider: String,
pub session_id: Option<String>,
pub cwd: PathBuf,
}
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
#[error("reachability probe inconclusive for provider '{provider}': {reason}")]
pub struct ReachabilityProbeError {
pub provider: String,
pub reason: String,
}
impl ReachabilityProbeError {
pub fn new(provider: &str, reason: impl Into<String>) -> Self {
ReachabilityProbeError {
provider: provider.to_string(),
reason: reason.into(),
}
}
}
pub trait Provider: Send + Sync {
fn name(&self) -> &'static str;
fn create_argv(&self, ctx: &CreateContext) -> Vec<String>;
fn resume_argv(&self, ctx: &ResumeContext) -> Vec<String>;
fn parse_stream_event(&self, chunk: &str) -> ParsedEvent;
fn reachability(
&self,
entry: &AgentEntry,
timeout: Duration,
) -> Result<bool, ReachabilityProbeError>;
fn as_pty(&self) -> Option<&dyn ProviderWithPty> {
None
}
}
pub trait ProviderWithPty: Provider {
fn readiness_detector(&self) -> Box<dyn ReadinessDetector>;
fn envelope(&self) -> Box<dyn Envelope>;
fn default_restart_policy(&self) -> RestartPolicy;
}
pub struct ClaudeProvider;
pub fn claude_stream_json_resume_argv(session_uuid: &str) -> Vec<String> {
vec![
"claude".into(),
"-p".into(),
"--resume".into(),
session_uuid.into(),
"--input-format".into(),
"stream-json".into(),
"--output-format".into(),
"stream-json".into(),
"--include-partial-messages".into(),
"--replay-user-messages".into(),
]
}
impl Provider for ClaudeProvider {
fn name(&self) -> &'static str {
"claude"
}
fn create_argv(&self, ctx: &CreateContext) -> Vec<String> {
vec![
"claude".into(),
"--bg".into(),
"--name".into(),
ctx.name.clone(),
ctx.message.clone(),
]
}
fn resume_argv(&self, ctx: &ResumeContext) -> Vec<String> {
vec![
"claude".into(),
"--resume".into(),
ctx.session_id.clone(),
"--print".into(),
ctx.message.clone(),
]
}
fn parse_stream_event(&self, chunk: &str) -> ParsedEvent {
match parse_claude_short_id(chunk) {
Some(id) => ParsedEvent::SessionCreated { session_id: id },
None => ParsedEvent::Unknown {
raw: chunk.to_string(),
},
}
}
fn reachability(
&self,
entry: &AgentEntry,
_timeout: Duration,
) -> Result<bool, ReachabilityProbeError> {
let short_id = entry
.session_id
.as_deref()
.filter(|s| !s.is_empty())
.ok_or_else(|| ReachabilityProbeError::new("claude", "no session id in entry"))?;
let jobs = home_dir()
.ok_or_else(|| ReachabilityProbeError::new("claude", "HOME unset"))?
.join(".claude")
.join("jobs");
if !jobs.exists() {
return Err(ReachabilityProbeError::new(
"claude",
"~/.claude/jobs absent",
));
}
Ok(jobs.join(short_id).exists())
}
}
fn parse_claude_short_id(line: &str) -> Option<String> {
line.split(|c: char| !c.is_ascii_hexdigit())
.find(|tok| tok.len() == 8 && tok.chars().all(|c| !c.is_ascii_uppercase()))
.map(|s| s.to_string())
}
pub struct ClaudeInteractiveProvider;
impl ClaudeInteractiveProvider {
fn interactive_argv(ctx: &CreateContext) -> Vec<String> {
let mut argv = vec!["claude".into()];
if let Some(sid) = ctx.session_id.as_deref().filter(|s| !s.is_empty()) {
argv.push("--session-id".into());
argv.push(sid.to_string());
}
if let Some(prompt) = ctx
.append_system_prompt
.as_deref()
.filter(|s| !s.is_empty())
{
argv.push("--append-system-prompt".into());
argv.push(prompt.to_string());
}
if !ctx.message.is_empty() {
argv.push(ctx.message.clone());
}
argv
}
}
impl Provider for ClaudeInteractiveProvider {
fn name(&self) -> &'static str {
"claude"
}
fn create_argv(&self, ctx: &CreateContext) -> Vec<String> {
Self::interactive_argv(ctx)
}
fn resume_argv(&self, ctx: &ResumeContext) -> Vec<String> {
let mut argv = vec!["claude".into(), "--resume".into(), ctx.session_id.clone()];
if !ctx.message.is_empty() {
argv.push(ctx.message.clone());
}
argv
}
fn parse_stream_event(&self, chunk: &str) -> ParsedEvent {
ParsedEvent::Unknown {
raw: chunk.to_string(),
}
}
fn reachability(
&self,
_entry: &AgentEntry,
_timeout: Duration,
) -> Result<bool, ReachabilityProbeError> {
Err(ReachabilityProbeError::new(
"claude",
"interactive claude liveness is PTY-governed (pid/ConnState), not store-probed",
))
}
fn as_pty(&self) -> Option<&dyn ProviderWithPty> {
Some(self)
}
}
impl ProviderWithPty for ClaudeInteractiveProvider {
fn readiness_detector(&self) -> Box<dyn ReadinessDetector> {
Box::new(crate::readiness::ClaudeReadinessDetector)
}
fn envelope(&self) -> Box<dyn Envelope> {
Box::new(JsonEnvelope)
}
fn default_restart_policy(&self) -> RestartPolicy {
RestartPolicy::default()
}
}
pub struct CodexProvider;
pub fn normalize_codex_command(message: &str) -> String {
let command = message.trim();
if let Some(verb) = command.strip_prefix("/fno:") {
format!("$fno:{verb}")
} else if let Some(verb) = command.strip_prefix('/') {
format!("$fno:{verb}")
} else if command.starts_with("$fno:") {
command.to_string()
} else {
message.to_string()
}
}
impl CodexProvider {
fn sandbox_create(yolo: bool) -> Vec<String> {
if yolo {
vec!["--dangerously-bypass-approvals-and-sandbox".into()]
} else {
vec!["--sandbox".into(), "workspace-write".into()]
}
}
fn sandbox_resume(yolo: bool) -> Vec<String> {
if yolo {
vec!["--dangerously-bypass-approvals-and-sandbox".into()]
} else {
vec![]
}
}
}
impl Provider for CodexProvider {
fn name(&self) -> &'static str {
"codex"
}
fn create_argv(&self, ctx: &CreateContext) -> Vec<String> {
let mut argv = vec![
"codex".into(),
"exec".into(),
"--json".into(),
"-C".into(),
ctx.cwd.to_string_lossy().into_owned(),
"--skip-git-repo-check".into(),
];
argv.extend(Self::sandbox_create(ctx.yolo));
if let Some(effort) = ctx.reasoning_effort.as_deref().filter(|e| !e.is_empty()) {
argv.push("-c".into());
argv.push(format!("model_reasoning_effort={effort}"));
}
argv.push(normalize_codex_command(&ctx.message));
argv
}
fn resume_argv(&self, ctx: &ResumeContext) -> Vec<String> {
let mut argv = vec![
"codex".into(),
"exec".into(),
"resume".into(),
ctx.session_id.clone(),
"--json".into(),
"--skip-git-repo-check".into(),
];
argv.extend(Self::sandbox_resume(ctx.yolo));
argv.push(normalize_codex_command(&ctx.message));
argv
}
fn parse_stream_event(&self, chunk: &str) -> ParsedEvent {
parse_codex_line(chunk)
}
fn reachability(
&self,
entry: &AgentEntry,
_timeout: Duration,
) -> Result<bool, ReachabilityProbeError> {
codex_reachable(entry)
}
fn as_pty(&self) -> Option<&dyn ProviderWithPty> {
Some(self)
}
}
fn codex_reachable(entry: &AgentEntry) -> Result<bool, ReachabilityProbeError> {
let sid = entry
.session_id
.as_deref()
.filter(|s| !s.is_empty())
.ok_or_else(|| ReachabilityProbeError::new("codex", "no session id in entry"))?;
let home = home_dir().ok_or_else(|| ReachabilityProbeError::new("codex", "HOME unset"))?;
let index = home.join(".codex").join("session_index.jsonl");
match std::fs::read_to_string(&index) {
Ok(text) => Ok(text.contains(sid)),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Err(ReachabilityProbeError::new(
"codex",
format!("session index absent (fresh install?): {}", index.display()),
)),
Err(e) => Err(ReachabilityProbeError::new(
"codex",
format!("cannot read session index {}: {e}", index.display()),
)),
}
}
impl ProviderWithPty for CodexProvider {
fn readiness_detector(&self) -> Box<dyn ReadinessDetector> {
Box::new(CodexReadinessDetector)
}
fn envelope(&self) -> Box<dyn Envelope> {
Box::new(JsonEnvelope)
}
fn default_restart_policy(&self) -> RestartPolicy {
RestartPolicy::default()
}
}
fn parse_codex_line(line: &str) -> ParsedEvent {
let trimmed = line.trim();
if trimmed.is_empty() {
return ParsedEvent::Unknown {
raw: line.to_string(),
};
}
let v: serde_json::Value = match serde_json::from_str(trimmed) {
Ok(v) => v,
Err(_) => {
return ParsedEvent::Unknown {
raw: line.to_string(),
}
}
};
let typ = v.get("type").and_then(|t| t.as_str()).unwrap_or("");
match typ {
"thread.started" => match v.get("thread_id").and_then(|t| t.as_str()) {
Some(id) => ParsedEvent::SessionCreated {
session_id: id.to_string(),
},
None => ParsedEvent::Unknown {
raw: line.to_string(),
},
},
"item.started" | "item.completed" => {
let item = v.get("item");
let item_type = item
.and_then(|i| i.get("type"))
.and_then(|t| t.as_str())
.unwrap_or("");
match item_type {
"agent_message" => ParsedEvent::OutputChunk {
text: item
.and_then(|i| i.get("text"))
.and_then(|t| t.as_str())
.unwrap_or("")
.to_string(),
},
"error" => ParsedEvent::ProviderError {
message: item
.and_then(|i| i.get("message"))
.and_then(|t| t.as_str())
.unwrap_or("")
.to_string(),
},
"command_execution" => ParsedEvent::ToolUse {
name: "command_execution".to_string(),
args: item.cloned(),
},
_ => ParsedEvent::Unknown {
raw: line.to_string(),
},
}
}
"turn.completed" => ParsedEvent::ReplyComplete {
text: String::new(),
duration_ms: 0,
},
_ => ParsedEvent::Unknown {
raw: line.to_string(),
},
}
}
pub struct GeminiProvider;
impl GeminiProvider {
fn sandbox(yolo: bool) -> Vec<String> {
if yolo {
vec!["--yolo".into()]
} else {
vec!["--approval-mode".into(), "default".into()]
}
}
}
impl Provider for GeminiProvider {
fn name(&self) -> &'static str {
"gemini"
}
fn create_argv(&self, ctx: &CreateContext) -> Vec<String> {
let mut argv = vec![
"gemini".into(),
"--skip-trust".into(),
"-p".into(),
ctx.message.clone(),
"--output-format".into(),
"json".into(),
];
argv.extend(Self::sandbox(ctx.yolo));
if let Some(sid) = ctx.session_id.as_deref() {
argv.push("--session-id".into());
argv.push(sid.to_string());
}
argv
}
fn resume_argv(&self, ctx: &ResumeContext) -> Vec<String> {
let mut argv = vec![
"gemini".into(),
"--skip-trust".into(),
"-p".into(),
ctx.message.clone(),
"--output-format".into(),
"json".into(),
];
argv.extend(Self::sandbox(ctx.yolo));
argv.push("--resume".into());
argv.push(ctx.session_id.clone());
argv
}
fn parse_stream_event(&self, chunk: &str) -> ParsedEvent {
parse_gemini_blob(chunk)
}
fn reachability(
&self,
entry: &AgentEntry,
timeout: Duration,
) -> Result<bool, ReachabilityProbeError> {
gemini_reachable(entry, timeout)
}
fn as_pty(&self) -> Option<&dyn ProviderWithPty> {
Some(self)
}
}
fn gemini_reachable(entry: &AgentEntry, budget: Duration) -> Result<bool, ReachabilityProbeError> {
let sid = entry
.session_id
.as_deref()
.filter(|s| !s.is_empty())
.ok_or_else(|| ReachabilityProbeError::new("gemini", "no session id in entry"))?;
let short = sid.get(..8).ok_or_else(|| {
ReachabilityProbeError::new(
"gemini",
format!("session_id too short or non-char-boundary at 8: {sid:?}"),
)
})?;
let home = home_dir().ok_or_else(|| ReachabilityProbeError::new("gemini", "HOME unset"))?;
let basename = entry.cwd.file_name().ok_or_else(|| {
ReachabilityProbeError::new(
"gemini",
format!("cwd has no basename: {}", entry.cwd.display()),
)
})?;
let chats_dir = home
.join(".gemini")
.join("tmp")
.join(basename)
.join("chats");
if !chats_dir.exists() {
return Err(ReachabilityProbeError::new(
"gemini",
format!("chats dir absent: {}", chats_dir.display()),
));
}
let deadline = std::time::Instant::now() + budget;
let dir = std::fs::read_dir(&chats_dir).map_err(|e| {
ReachabilityProbeError::new("gemini", format!("read_dir {}: {e}", chats_dir.display()))
})?;
for ent in dir {
if std::time::Instant::now() >= deadline {
return Err(ReachabilityProbeError::new(
"gemini",
"probe budget exceeded before definitive result",
));
}
let ent = ent.map_err(|e| {
ReachabilityProbeError::new(
"gemini",
format!("dir entry in {}: {e}", chats_dir.display()),
)
})?;
let name = ent.file_name();
if !name.to_string_lossy().contains(short) {
continue;
}
let file = std::fs::File::open(ent.path()).map_err(|e| {
ReachabilityProbeError::new("gemini", format!("open {}: {e}", ent.path().display()))
})?;
let mut first_line = String::new();
std::io::BufRead::read_line(&mut std::io::BufReader::new(file), &mut first_line).map_err(
|e| {
ReachabilityProbeError::new("gemini", format!("read {}: {e}", ent.path().display()))
},
)?;
if first_line.contains(sid) {
return Ok(true);
}
}
Ok(false)
}
impl ProviderWithPty for GeminiProvider {
fn readiness_detector(&self) -> Box<dyn ReadinessDetector> {
Box::new(GeminiReadinessDetector)
}
fn envelope(&self) -> Box<dyn Envelope> {
Box::new(JsonEnvelope)
}
fn default_restart_policy(&self) -> RestartPolicy {
RestartPolicy::default()
}
}
pub struct AgyProvider;
impl AgyProvider {}
impl Provider for AgyProvider {
fn name(&self) -> &'static str {
"agy"
}
fn create_argv(&self, ctx: &CreateContext) -> Vec<String> {
let mut argv = vec!["agy".into(), "--dangerously-skip-permissions".into()];
argv.push("-p".into());
argv.push(ctx.message.clone());
argv
}
fn resume_argv(&self, ctx: &ResumeContext) -> Vec<String> {
let mut argv = vec![
"agy".into(),
"--dangerously-skip-permissions".into(),
"--conversation".into(),
ctx.session_id.clone(),
];
argv.push("-p".into());
argv.push(ctx.message.clone());
argv
}
fn parse_stream_event(&self, chunk: &str) -> ParsedEvent {
if chunk.trim().is_empty() {
ParsedEvent::Unknown {
raw: chunk.to_string(),
}
} else {
ParsedEvent::ReplyComplete {
text: chunk.to_string(),
duration_ms: 0,
}
}
}
fn reachability(
&self,
_entry: &AgentEntry,
_timeout: Duration,
) -> Result<bool, ReachabilityProbeError> {
Err(ReachabilityProbeError::new(
"agy",
"agy sessions are not probeable (plain-text, no session store)",
))
}
fn as_pty(&self) -> Option<&dyn ProviderWithPty> {
Some(self)
}
}
impl ProviderWithPty for AgyProvider {
fn readiness_detector(&self) -> Box<dyn ReadinessDetector> {
Box::new(AgyReadinessDetector)
}
fn envelope(&self) -> Box<dyn Envelope> {
Box::new(NoEnvelope)
}
fn default_restart_policy(&self) -> RestartPolicy {
RestartPolicy::default()
}
}
pub(crate) fn opencode_run_tail(message: &str) -> Vec<String> {
if let Some(rest) = message.strip_prefix('/') {
let mut parts = rest.splitn(2, ' ');
if let Some(cmd) = parts.next().filter(|c| !c.is_empty()) {
let mut tail = vec!["--command".to_string(), cmd.to_string()];
if let Some(args) = parts.next().filter(|a| !a.is_empty()) {
tail.push(args.to_string());
}
return tail;
}
}
vec![message.to_string()]
}
pub struct OpencodeProvider;
impl Provider for OpencodeProvider {
fn name(&self) -> &'static str {
"opencode"
}
fn create_argv(&self, ctx: &CreateContext) -> Vec<String> {
let mut argv = vec![
"opencode".into(),
"run".into(),
"--dangerously-skip-permissions".into(),
];
argv.extend(opencode_run_tail(&ctx.message));
argv
}
fn resume_argv(&self, ctx: &ResumeContext) -> Vec<String> {
let mut argv = vec![
"opencode".into(),
"run".into(),
"--dangerously-skip-permissions".into(),
"--session".into(),
ctx.session_id.clone(),
];
argv.extend(opencode_run_tail(&ctx.message));
argv
}
fn parse_stream_event(&self, chunk: &str) -> ParsedEvent {
if chunk.trim().is_empty() {
ParsedEvent::Unknown {
raw: chunk.to_string(),
}
} else {
ParsedEvent::ReplyComplete {
text: chunk.to_string(),
duration_ms: 0,
}
}
}
fn reachability(
&self,
entry: &AgentEntry,
timeout: Duration,
) -> Result<bool, ReachabilityProbeError> {
opencode_reachable_with(
entry,
timeout.max(OPENCODE_PROBE_MIN_BUDGET),
&(run_opencode_db as OpencodeDbRunner),
)
}
fn as_pty(&self) -> Option<&dyn ProviderWithPty> {
Some(self)
}
}
type OpencodeDbRunner = fn(&str, Duration) -> Result<(bool, String), String>;
const OPENCODE_PROBE_MIN_BUDGET: Duration = Duration::from_secs(2);
fn is_opencode_session_id(s: &str) -> bool {
match s.strip_prefix("ses_") {
Some(tail) => !tail.is_empty() && tail.chars().all(|c| c.is_ascii_alphanumeric()),
None => false,
}
}
fn opencode_reachable_with(
entry: &AgentEntry,
timeout: Duration,
run: &OpencodeDbRunner,
) -> Result<bool, ReachabilityProbeError> {
let sid = entry
.session_id
.as_deref()
.filter(|s| !s.is_empty())
.ok_or_else(|| ReachabilityProbeError::new("opencode", "no session id in entry"))?;
if !is_opencode_session_id(sid) {
return Err(ReachabilityProbeError::new(
"opencode",
format!("malformed opencode session id {sid:?} (expected ses_<alnum>)"),
));
}
let sql = format!("select id from session where id='{sid}'");
match run(&sql, timeout) {
Ok((true, stdout)) => Ok(stdout.contains(sid)),
Ok((false, _)) => Err(ReachabilityProbeError::new(
"opencode",
"`opencode db` exited nonzero (store unreadable or query rejected)",
)),
Err(e) => Err(ReachabilityProbeError::new(
"opencode",
format!("cannot run `opencode db`: {e}"),
)),
}
}
fn run_opencode_db(sql: &str, timeout: Duration) -> Result<(bool, String), String> {
use std::process::{Command, Stdio};
let mut child = Command::new("opencode")
.arg("db")
.arg(sql)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.map_err(|e| e.to_string())?;
let deadline = Instant::now() + timeout;
loop {
match child.try_wait() {
Ok(Some(_)) => break,
Ok(None) => {
if Instant::now() >= deadline {
let _ = child.kill();
let _ = child.wait();
return Err(format!("probe timed out after {timeout:?}"));
}
std::thread::sleep(Duration::from_millis(25));
}
Err(e) => {
let _ = child.kill();
let _ = child.wait();
return Err(e.to_string());
}
}
}
let out = child.wait_with_output().map_err(|e| e.to_string())?;
Ok((
out.status.success(),
String::from_utf8_lossy(&out.stdout).into_owned(),
))
}
impl ProviderWithPty for OpencodeProvider {
fn readiness_detector(&self) -> Box<dyn ReadinessDetector> {
Box::new(OpencodeReadinessDetector)
}
fn envelope(&self) -> Box<dyn Envelope> {
Box::new(NoEnvelope)
}
fn default_restart_policy(&self) -> RestartPolicy {
RestartPolicy::default()
}
}
fn parse_gemini_blob(blob: &str) -> ParsedEvent {
let v: serde_json::Value = match serde_json::from_str(blob.trim()) {
Ok(v) => v,
Err(_) => {
return ParsedEvent::Unknown {
raw: blob.to_string(),
}
}
};
if let Some(resp) = v.get("response").and_then(|r| r.as_str()) {
return ParsedEvent::ReplyComplete {
text: resp.to_string(),
duration_ms: gemini_total_latency_ms(&v),
};
}
if let Some(sid) = v.get("session_id").and_then(|s| s.as_str()) {
return ParsedEvent::SessionCreated {
session_id: sid.to_string(),
};
}
ParsedEvent::Unknown {
raw: blob.to_string(),
}
}
fn gemini_total_latency_ms(v: &serde_json::Value) -> u64 {
let Some(models) = v
.get("stats")
.and_then(|s| s.get("models"))
.and_then(|m| m.as_object())
else {
return 0;
};
models
.values()
.filter_map(|m| m.get("api"))
.filter_map(|api| api.get("totalLatencyMs"))
.filter_map(|l| l.as_u64())
.sum()
}
fn home_dir() -> Option<PathBuf> {
std::env::var_os("HOME").map(PathBuf::from)
}
pub fn gemini_session_id_from_blob(blob: &str) -> Option<String> {
serde_json::from_str::<serde_json::Value>(blob.trim())
.ok()?
.get("session_id")?
.as_str()
.map(|s| s.to_string())
}
pub const KNOWN_PROVIDERS: &[&str] = &["claude", "codex", "gemini", "agy", "opencode"];
pub fn known_providers_csv() -> String {
KNOWN_PROVIDERS.join(", ")
}
pub fn for_name(name: &str) -> Option<Box<dyn Provider>> {
match name {
"claude" => Some(Box::new(ClaudeProvider)),
"codex" => Some(Box::new(CodexProvider)),
"gemini" => Some(Box::new(GeminiProvider)),
"agy" => Some(Box::new(AgyProvider)),
"opencode" => Some(Box::new(OpencodeProvider)),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn create_ctx() -> CreateContext {
CreateContext {
name: "worker-A".into(),
message: "build feature X".into(),
cwd: PathBuf::from("/tmp/example-repo"),
from_name: None,
session_id: None,
yolo: false,
reasoning_effort: None,
append_system_prompt: None,
}
}
#[test]
fn claude_create_argv_uses_bg_not_print() {
let argv = ClaudeProvider.create_argv(&create_ctx());
assert_eq!(
argv,
vec!["claude", "--bg", "--name", "worker-A", "build feature X"]
);
assert!(!argv.iter().any(|a| a == "-p"), "LD38: never claude -p");
}
#[test]
fn claude_resume_argv_is_resume_print() {
let ctx = ResumeContext {
session_id: "7c5dcf5d".into(),
message: "follow up".into(),
cwd: PathBuf::from("/x"),
from_name: None,
yolo: false,
};
assert_eq!(
ClaudeProvider.resume_argv(&ctx),
vec!["claude", "--resume", "7c5dcf5d", "--print", "follow up"]
);
}
#[test]
fn claude_stream_json_resume_argv_uses_p_and_full_uuid() {
let argv = claude_stream_json_resume_argv("019e7157-4236-7bb1-b274-ebbac6040ace");
assert_eq!(
argv,
vec![
"claude",
"-p",
"--resume",
"019e7157-4236-7bb1-b274-ebbac6040ace",
"--input-format",
"stream-json",
"--output-format",
"stream-json",
"--include-partial-messages",
"--replay-user-messages",
]
);
}
#[test]
fn codex_create_argv_defaults_to_workspace_write_sandbox() {
let argv = CodexProvider.create_argv(&create_ctx());
assert_eq!(
argv,
vec![
"codex",
"exec",
"--json",
"-C",
"/tmp/example-repo",
"--skip-git-repo-check",
"--sandbox",
"workspace-write",
"build feature X"
]
);
}
#[test]
fn codex_create_argv_yolo_is_mutually_exclusive_with_sandbox() {
let mut ctx = create_ctx();
ctx.yolo = true;
let argv = CodexProvider.create_argv(&ctx);
assert!(argv.contains(&"--dangerously-bypass-approvals-and-sandbox".to_string()));
assert!(!argv.iter().any(|a| a == "--sandbox"));
}
#[test]
fn codex_create_argv_appends_reasoning_effort() {
let mut ctx = create_ctx();
ctx.reasoning_effort = Some("high".into());
let argv = CodexProvider.create_argv(&ctx);
assert!(argv
.windows(2)
.any(|w| w == ["-c", "model_reasoning_effort=high"]));
}
#[test]
fn codex_create_and_resume_normalize_direct_slash_commands() {
let mut create = create_ctx();
create.message = " /fno:target x-81ad ".into();
assert_eq!(
CodexProvider
.create_argv(&create)
.last()
.map(String::as_str),
Some("$fno:target x-81ad")
);
let resume = ResumeContext {
session_id: "uuid-1".into(),
message: " /fno:target x-81ad ".into(),
cwd: PathBuf::from("/x"),
from_name: None,
yolo: false,
};
assert_eq!(
CodexProvider
.resume_argv(&resume)
.last()
.map(String::as_str),
Some("$fno:target x-81ad")
);
assert_eq!(
normalize_codex_command(" review this\n code "),
" review this\n code "
);
}
#[test]
fn codex_resume_argv_omits_sandbox_unless_yolo() {
let ctx = ResumeContext {
session_id: "uuid-1".into(),
message: "m".into(),
cwd: PathBuf::from("/x"),
from_name: None,
yolo: false,
};
assert_eq!(
CodexProvider.resume_argv(&ctx),
vec![
"codex",
"exec",
"resume",
"uuid-1",
"--json",
"--skip-git-repo-check",
"m"
]
);
}
#[test]
fn gemini_create_argv_passes_session_id_and_default_approval() {
let mut ctx = create_ctx();
ctx.session_id = Some("uuid-g".into());
let argv = GeminiProvider.create_argv(&ctx);
assert_eq!(
argv,
vec![
"gemini",
"--skip-trust",
"-p",
"build feature X",
"--output-format",
"json",
"--approval-mode",
"default",
"--session-id",
"uuid-g"
]
);
}
#[test]
fn gemini_resume_argv_uses_resume_flag() {
let ctx = ResumeContext {
session_id: "uuid-g".into(),
message: "m".into(),
cwd: PathBuf::from("/x"),
from_name: None,
yolo: true,
};
let argv = GeminiProvider.resume_argv(&ctx);
assert_eq!(
argv,
vec![
"gemini",
"--skip-trust",
"-p",
"m",
"--output-format",
"json",
"--yolo",
"--resume",
"uuid-g"
]
);
}
#[test]
fn claude_is_not_pty_managed_others_are() {
assert!(ClaudeProvider.as_pty().is_none());
assert!(ClaudeInteractiveProvider.as_pty().is_some());
assert!(CodexProvider.as_pty().is_some());
assert!(GeminiProvider.as_pty().is_some());
}
#[test]
fn opencode_create_argv_is_headless_run_never_bare_tui() {
let argv = OpencodeProvider.create_argv(&create_ctx());
assert_eq!(
argv,
vec![
"opencode",
"run",
"--dangerously-skip-permissions",
"build feature X"
]
);
}
#[test]
fn opencode_create_argv_routes_slash_command_via_command_flag() {
let mut ctx = create_ctx();
ctx.message = "/fno:target no-merge x-abcd".into();
assert_eq!(
OpencodeProvider.create_argv(&ctx),
vec![
"opencode",
"run",
"--dangerously-skip-permissions",
"--command",
"fno:target",
"no-merge x-abcd"
]
);
}
#[test]
fn opencode_run_tail_prose_through_and_bare_verb() {
assert_eq!(
opencode_run_tail("build feature X"),
vec!["build feature X"]
);
assert_eq!(opencode_run_tail("/fno:pr"), vec!["--command", "fno:pr"]);
}
#[test]
fn opencode_resume_argv_uses_session_flag() {
let ctx = ResumeContext {
session_id: "ses_abc".into(),
message: "m".into(),
cwd: PathBuf::from("/x"),
from_name: None,
yolo: false,
};
assert_eq!(
OpencodeProvider.resume_argv(&ctx),
vec![
"opencode",
"run",
"--dangerously-skip-permissions",
"--session",
"ses_abc",
"m"
]
);
}
#[test]
fn opencode_is_pty_managed_and_id_less_probe_is_inconclusive() {
assert!(OpencodeProvider.as_pty().is_some());
let entry = AgentEntry {
name: "oc".into(),
provider: "opencode".into(),
session_id: None,
cwd: PathBuf::from("/x"),
};
assert!(OpencodeProvider
.reachability(&entry, Duration::from_secs(1))
.is_err());
}
#[test]
fn for_name_round_trips_every_known_provider() {
for name in KNOWN_PROVIDERS.iter().copied() {
let p = for_name(name).unwrap_or_else(|| panic!("for_name({name}) returned None"));
assert_eq!(
p.name(),
name,
"for_name({name}) resolved to wrong provider"
);
}
assert!(for_name("nope").is_none(), "unknown provider must be None");
}
#[test]
fn claude_parses_short_id_from_bg_line() {
let ev = ClaudeProvider.parse_stream_event("backgrounded · 7c5dcf5d · worker-A");
assert_eq!(
ev,
ParsedEvent::SessionCreated {
session_id: "7c5dcf5d".into()
}
);
}
#[test]
fn claude_non_id_line_is_unknown() {
assert!(matches!(
ClaudeProvider.parse_stream_event("starting up"),
ParsedEvent::Unknown { .. }
));
assert!(matches!(
ClaudeProvider.parse_stream_event("ABCDEF12"),
ParsedEvent::Unknown { .. }
));
}
#[test]
fn codex_thread_started_is_session_created() {
let ev = parse_codex_line(
r#"{"type":"thread.started","thread_id":"019e4958-80d1-7492-8054-2854dfda502c"}"#,
);
assert_eq!(
ev,
ParsedEvent::SessionCreated {
session_id: "019e4958-80d1-7492-8054-2854dfda502c".into()
}
);
}
#[test]
fn codex_agent_message_is_output_chunk() {
let ev = parse_codex_line(
r#"{"type":"item.completed","item":{"id":"item_3","type":"agent_message","text":"hello"}}"#,
);
assert_eq!(
ev,
ParsedEvent::OutputChunk {
text: "hello".into()
}
);
}
#[test]
fn codex_error_item_is_provider_error() {
let ev = parse_codex_line(
r#"{"type":"item.completed","item":{"id":"item_0","type":"error","message":"boom"}}"#,
);
assert_eq!(
ev,
ParsedEvent::ProviderError {
message: "boom".into()
}
);
}
#[test]
fn codex_command_execution_is_tool_use() {
let ev = parse_codex_line(
r#"{"type":"item.started","item":{"id":"item_2","type":"command_execution","command":"echo hi"}}"#,
);
match ev {
ParsedEvent::ToolUse { name, args } => {
assert_eq!(name, "command_execution");
assert_eq!(args.unwrap()["command"], "echo hi");
}
other => panic!("expected ToolUse, got {other:?}"),
}
}
#[test]
fn codex_turn_completed_is_reply_complete_marker() {
let ev = parse_codex_line(r#"{"type":"turn.completed","usage":{"output_tokens":91}}"#);
assert_eq!(
ev,
ParsedEvent::ReplyComplete {
text: String::new(),
duration_ms: 0
}
);
}
#[test]
fn codex_preamble_and_control_frames_are_unknown() {
assert!(matches!(
parse_codex_line("Reading additional input from stdin..."),
ParsedEvent::Unknown { .. }
));
assert!(matches!(
parse_codex_line(r#"{"type":"turn.started"}"#),
ParsedEvent::Unknown { .. }
));
}
#[test]
fn gemini_blob_response_is_reply_complete_with_latency() {
let blob = r#"{
"session_id": "abc",
"response": "PONG",
"stats": {"models": {"gemini-3.1-flash-lite": {"api": {"totalLatencyMs": 3359}}}}
}"#;
assert_eq!(
parse_gemini_blob(blob),
ParsedEvent::ReplyComplete {
text: "PONG".into(),
duration_ms: 3359
}
);
}
#[test]
fn gemini_latency_sums_across_models() {
let blob = r#"{
"response": "ok",
"stats": {"models": {
"m1": {"api": {"totalLatencyMs": 100}},
"m2": {"api": {"totalLatencyMs": 250}}
}}
}"#;
assert_eq!(
parse_gemini_blob(blob),
ParsedEvent::ReplyComplete {
text: "ok".into(),
duration_ms: 350
}
);
}
#[test]
fn gemini_session_only_blob_is_session_created() {
let ev = parse_gemini_blob(r#"{"session_id":"xyz"}"#);
assert_eq!(
ev,
ParsedEvent::SessionCreated {
session_id: "xyz".into()
}
);
}
#[test]
fn gemini_partial_or_garbage_is_unknown() {
assert!(matches!(
parse_gemini_blob(r#"{"session_id": "incomplete"#),
ParsedEvent::Unknown { .. }
));
}
#[test]
fn gemini_session_id_recoverable_from_create_blob_even_with_reply() {
let blob = r#"{"session_id":"abc-123","response":"hi","stats":{}}"#;
assert_eq!(
parse_gemini_blob(blob),
ParsedEvent::ReplyComplete {
text: "hi".into(),
duration_ms: 0
}
);
assert_eq!(gemini_session_id_from_blob(blob), Some("abc-123".into()));
assert_eq!(gemini_session_id_from_blob("not json"), None);
assert_eq!(gemini_session_id_from_blob(r#"{"response":"x"}"#), None);
}
#[test]
fn reachability_no_session_id_is_inconclusive() {
let entry = AgentEntry {
name: "a".into(),
provider: "codex".into(),
session_id: None,
cwd: PathBuf::from("/x"),
};
let err = CodexProvider
.reachability(&entry, Duration::from_millis(250))
.unwrap_err();
assert_eq!(err.provider, "codex");
}
fn codex_entry(session_id: &str) -> AgentEntry {
AgentEntry {
name: "a".into(),
provider: "codex".into(),
session_id: Some(session_id.into()),
cwd: PathBuf::from("/x"),
}
}
const OC_SES: &str = "ses_09679f284ffeJv7NdBAoLQLnLZ";
fn opencode_entry(session_id: Option<&str>) -> AgentEntry {
AgentEntry {
name: "o".into(),
provider: "opencode".into(),
session_id: session_id.map(Into::into),
cwd: PathBuf::from("/x"),
}
}
#[test]
fn opencode_reachable_when_store_returns_the_id() {
fn run(_sql: &str, _t: Duration) -> Result<(bool, String), String> {
Ok((
true,
format!("[claude-mem] OpenCode plugin loading\nid\n{OC_SES}\n"),
))
}
assert_eq!(
opencode_reachable_with(
&opencode_entry(Some(OC_SES)),
Duration::from_secs(2),
&(run as OpencodeDbRunner)
),
Ok(true)
);
}
#[test]
fn opencode_probe_embeds_only_the_validated_id_in_the_query() {
use std::sync::{Mutex, OnceLock};
static SEEN: OnceLock<Mutex<String>> = OnceLock::new();
fn run(sql: &str, _t: Duration) -> Result<(bool, String), String> {
*SEEN.get_or_init(Default::default).lock().unwrap() = sql.to_string();
Ok((true, OC_SES.to_string()))
}
let _ = opencode_reachable_with(
&opencode_entry(Some(OC_SES)),
Duration::from_secs(2),
&(run as OpencodeDbRunner),
);
assert_eq!(
*SEEN.get_or_init(Default::default).lock().unwrap(),
format!("select id from session where id='{OC_SES}'")
);
}
#[test]
fn opencode_clean_query_without_the_id_is_gone() {
fn run(_sql: &str, _t: Duration) -> Result<(bool, String), String> {
Ok((true, String::new()))
}
assert_eq!(
opencode_reachable_with(
&opencode_entry(Some(OC_SES)),
Duration::from_secs(2),
&(run as OpencodeDbRunner)
),
Ok(false)
);
}
#[test]
fn opencode_infrastructure_failure_is_inconclusive_never_gone() {
fn spawn_failed(_sql: &str, _t: Duration) -> Result<(bool, String), String> {
Err("No such file or directory (os error 2)".into())
}
fn nonzero(_sql: &str, _t: Duration) -> Result<(bool, String), String> {
Ok((false, String::new()))
}
for run in [
spawn_failed as OpencodeDbRunner,
nonzero as OpencodeDbRunner,
] {
let err = opencode_reachable_with(
&opencode_entry(Some(OC_SES)),
Duration::from_secs(2),
&run,
)
.unwrap_err();
assert_eq!(err.provider, "opencode");
}
}
#[test]
fn opencode_malformed_id_never_reaches_the_subprocess() {
fn run(_sql: &str, _t: Duration) -> Result<(bool, String), String> {
panic!("probe must reject a malformed id before spawning");
}
for bad in ["ses_'; drop table session--", "ses_ x", "ses_", "not-a-ses"] {
let err = opencode_reachable_with(
&opencode_entry(Some(bad)),
Duration::from_secs(2),
&(run as OpencodeDbRunner),
)
.unwrap_err();
assert!(err.reason.contains(bad), "reason should quote {bad:?}");
}
}
#[test]
fn opencode_missing_session_id_is_inconclusive() {
fn run(_sql: &str, _t: Duration) -> Result<(bool, String), String> {
panic!("no id means nothing to probe");
}
assert!(opencode_reachable_with(
&opencode_entry(None),
Duration::from_secs(2),
&(run as OpencodeDbRunner)
)
.is_err());
}
#[test]
fn opencode_probe_is_stateless_across_calls() {
fn failing(_sql: &str, _t: Duration) -> Result<(bool, String), String> {
Err("binary briefly unavailable".into())
}
fn healthy(_sql: &str, _t: Duration) -> Result<(bool, String), String> {
Ok((true, OC_SES.to_string()))
}
let entry = opencode_entry(Some(OC_SES));
assert!(opencode_reachable_with(
&entry,
Duration::from_secs(2),
&(failing as OpencodeDbRunner)
)
.is_err());
assert_eq!(
opencode_reachable_with(
&entry,
Duration::from_secs(2),
&(healthy as OpencodeDbRunner)
),
Ok(true)
);
}
#[test]
fn codex_reachable_when_id_in_session_index() {
let tmp = tempdir();
let idx = tmp.join(".codex").join("session_index.jsonl");
std::fs::create_dir_all(idx.parent().unwrap()).unwrap();
std::fs::write(
&idx,
"{\"id\":\"019e4958-80d1-7492-8054-2854dfda502c\",\"status\":\"live\"}\n",
)
.unwrap();
with_home(&tmp, || {
let entry = codex_entry("019e4958-80d1-7492-8054-2854dfda502c");
assert_eq!(
CodexProvider.reachability(&entry, Duration::from_secs(2)),
Ok(true)
);
});
}
#[test]
fn codex_index_present_id_absent_is_false() {
let tmp = tempdir();
let idx = tmp.join(".codex").join("session_index.jsonl");
std::fs::create_dir_all(idx.parent().unwrap()).unwrap();
std::fs::write(&idx, "{\"id\":\"some-other-uuid\"}\n").unwrap();
with_home(&tmp, || {
let entry = codex_entry("019e4958-80d1-7492-8054-2854dfda502c");
assert_eq!(
CodexProvider.reachability(&entry, Duration::from_secs(2)),
Ok(false)
);
});
}
#[test]
fn codex_index_absent_is_inconclusive() {
let tmp = tempdir();
with_home(&tmp, || {
let entry = codex_entry("019e4958-80d1-7492-8054-2854dfda502c");
assert!(CodexProvider
.reachability(&entry, Duration::from_secs(2))
.is_err());
});
}
fn gemini_entry(session_id: &str, cwd: &str) -> AgentEntry {
AgentEntry {
name: "g".into(),
provider: "gemini".into(),
session_id: Some(session_id.into()),
cwd: PathBuf::from(cwd),
}
}
const G_UUID: &str = "35624650-b11e-4300-ad85-0fc87baeb3af";
#[test]
fn gemini_reachability_is_cwd_pinned_and_verifies_full_uuid() {
let tmp = tempdir();
let chats = tmp
.join(".gemini")
.join("tmp")
.join("myproject")
.join("chats");
std::fs::create_dir_all(&chats).unwrap();
std::fs::write(
chats.join("session-35624650.json"),
format!("{{\"sessionId\":\"{G_UUID}\",\"messages\":[]}}\n").as_bytes(),
)
.unwrap();
with_home(&tmp, || {
let entry = gemini_entry(G_UUID, "/work/myproject");
assert_eq!(
GeminiProvider.reachability(&entry, Duration::from_secs(2)),
Ok(true)
);
let other = gemini_entry(G_UUID, "/work/elsewhere");
assert!(GeminiProvider
.reachability(&other, Duration::from_secs(2))
.is_err()); });
}
#[test]
fn gemini_short_prefix_collision_without_full_uuid_is_false() {
let tmp = tempdir();
let chats = tmp.join(".gemini").join("tmp").join("proj").join("chats");
std::fs::create_dir_all(&chats).unwrap();
std::fs::write(
chats.join("session-35624650.json"),
b"{\"sessionId\":\"35624650-ffff-ffff-ffff-ffffffffffff\"}\n",
)
.unwrap();
with_home(&tmp, || {
let entry = gemini_entry(G_UUID, "/x/proj");
assert_eq!(
GeminiProvider.reachability(&entry, Duration::from_secs(2)),
Ok(false)
);
});
}
#[test]
fn gemini_reachability_chats_present_no_match_is_false() {
let tmp = tempdir();
let chats = tmp.join(".gemini").join("tmp").join("proj").join("chats");
std::fs::create_dir_all(&chats).unwrap();
std::fs::write(chats.join("session-deadbeef.json"), b"{}").unwrap();
with_home(&tmp, || {
let entry = gemini_entry("00000000-1111-2222-3333-444444444444", "/x/proj");
assert_eq!(
GeminiProvider.reachability(&entry, Duration::from_secs(2)),
Ok(false)
);
});
}
#[test]
fn gemini_reachability_short_session_id_is_inconclusive() {
let entry = gemini_entry("uuid", "/x/proj");
let err = GeminiProvider
.reachability(&entry, Duration::from_millis(250))
.unwrap_err();
assert_eq!(err.provider, "gemini");
assert!(err.reason.contains("too short"));
}
fn tempdir() -> PathBuf {
let mut p = std::env::temp_dir();
let unique = format!(
"fno-agents-test-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos()
);
p.push(unique);
std::fs::create_dir_all(&p).unwrap();
p
}
static HOME_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
fn with_home(home: &std::path::Path, f: impl FnOnce()) {
let _guard = HOME_LOCK.lock().unwrap_or_else(|e| e.into_inner());
let prev = std::env::var_os("HOME");
std::env::set_var("HOME", home);
f();
match prev {
Some(v) => std::env::set_var("HOME", v),
None => std::env::remove_var("HOME"),
}
}
}