use std::borrow::Cow;
use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use seher::sdk::{
CodexBarProbe, EffortLevel, PiRunner, PiRunnerOptions, PollOptions, SeherTool, StreamChunk,
poll_for_agent, split_thinking_suffix,
};
use crate::cancellation::CancellationToken;
use crate::error::{CruiseError, Result};
use crate::step::prompt::{PromptResult, StreamCallbacks, run_prompt};
pub const DEFAULT_STEP_MODE_KEY: &str = "build";
pub const DEFAULT_PLAN_MODE_KEY: &str = "plan";
const SDK_POLL_INTERVAL_MS: u64 = 60_000;
pub struct PromptRun<'a> {
pub prompt: &'a str,
pub model_or_mode: Option<&'a str>,
pub max_retries: usize,
pub env: &'a HashMap<String, String>,
pub on_notice: Option<&'a (dyn Fn(&str) + Send + Sync)>,
pub cancel_token: Option<&'a CancellationToken>,
pub working_dir: Option<&'a Path>,
pub stream: Option<&'a StreamCallbacks<'a>>,
pub tools: Vec<SeherTool>,
pub resume: Option<String>,
}
#[derive(Debug, Clone)]
pub struct PromptOutcome {
pub result: PromptResult,
pub session_id: Option<String>,
}
pub enum Executor {
Command { command: Vec<String> },
Sdk,
Pi,
}
impl Executor {
#[must_use]
pub fn new(sdk: Option<&str>, command: &[String]) -> Self {
match sdk {
Some("pi") => Executor::Pi,
Some(_) => Executor::Sdk,
None => Executor::Command {
command: command.to_vec(),
},
}
}
#[must_use]
pub fn is_sdk(&self) -> bool {
matches!(self, Executor::Sdk | Executor::Pi)
}
#[must_use]
pub fn step_model_or_mode(
&self,
step_model: Option<&str>,
global_model: Option<&str>,
) -> Option<String> {
match self {
Executor::Command { .. } | Executor::Pi => {
step_model.or(global_model).map(str::to_string)
}
Executor::Sdk => Some(mode_key_for_step(step_model, global_model)),
}
}
#[must_use]
pub fn plan_model_or_mode(
&self,
plan_model: Option<&str>,
global_model: Option<&str>,
) -> Option<String> {
match self {
Executor::Command { .. } | Executor::Pi => {
plan_model.or(global_model).map(str::to_string)
}
Executor::Sdk => Some(mode_key_for_plan(plan_model, global_model)),
}
}
pub async fn run(&self, req: PromptRun<'_>) -> Result<PromptOutcome> {
match self {
Executor::Command { command } => run_command(command, req).await,
Executor::Sdk => run_sdk(req).await,
Executor::Pi => run_pi_direct(req).await,
}
}
}
async fn resolve_provider(
mode_key: String,
require_tools: bool,
cancel: Arc<AtomicBool>,
) -> Result<seher::sdk::ResolvedAgent> {
let (tx, rx) = tokio::sync::oneshot::channel();
std::thread::spawn(move || {
let result = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| CruiseError::Other(format!("failed to build seher resolver runtime: {e}")))
.and_then(|rt| {
rt.block_on(async {
let mut probe = CodexBarProbe;
poll_for_agent(
PollOptions {
mode_key,
require_tools,
interval_ms: SDK_POLL_INTERVAL_MS,
cancel: Some(cancel),
..Default::default()
},
&mut probe,
)
.await
.map_err(|e| {
CruiseError::CommandError(format!("seher provider resolution failed: {e}"))
})
})
});
let _ = tx.send(result);
});
rx.await
.map_err(|_| CruiseError::Other("seher resolver thread terminated".to_string()))?
}
struct AbortOnDrop(Arc<AtomicBool>);
impl Drop for AbortOnDrop {
fn drop(&mut self) {
self.0.store(true, Ordering::SeqCst);
}
}
struct CancelOnDrop(seher::sdk::CancelToken);
impl Drop for CancelOnDrop {
fn drop(&mut self) {
self.0.cancel();
}
}
async fn maybe_cancelled(token: Option<&CancellationToken>) {
match token {
Some(t) => t.cancelled().await,
None => std::future::pending().await,
}
}
async fn run_command(command: &[String], req: PromptRun<'_>) -> Result<PromptOutcome> {
let resolved = crate::engine::resolve_command_with_model(command, req.model_or_mode)?;
let model_arg = if resolved.consumed_model_placeholder {
None
} else {
req.model_or_mode.map(str::to_string)
};
let resolved_command = resolved.command;
let retry = |msg: &str| {
if let Some(cb) = req.on_notice {
cb(msg);
}
};
let result = run_prompt(
&resolved_command,
model_arg.as_deref(),
req.prompt,
req.max_retries,
req.env,
Some(&retry),
req.cancel_token,
req.working_dir,
req.stream,
)
.await?;
Ok(PromptOutcome {
result,
session_id: None,
})
}
fn spawn_agent_stream(
resolved: &seher::sdk::ResolvedAgent,
req: &PromptRun<'_>,
omp_cancel: seher::sdk::CancelToken,
) -> std::sync::mpsc::Receiver<StreamChunk> {
let cwd_string = req.working_dir.map(|p| p.to_string_lossy().into_owned());
match resolved.sdk.as_str() {
"claude" => {
let config = seher::claude_agent::ClaudeAgentRunnerConfig {
model: claude_family_model(&resolved.model_id),
effort: claude_family_effort(&resolved.model_id, resolved.effort),
cwd: cwd_string,
resume_session_id: req.resume.clone(),
tools: req.tools.clone(),
env: req.env.clone(),
..Default::default()
};
seher::claude_agent::stream_agent(
config,
req.prompt.to_string(),
resolved.provider.clone(),
)
}
"claude-headless" => {
let mut headless_cfg = seher::claude_headless::ClaudeHeadlessRunnerConfig::default();
headless_cfg.model = claude_family_model(&resolved.model_id);
headless_cfg.effort = claude_family_effort(&resolved.model_id, resolved.effort);
headless_cfg.cwd = cwd_string;
headless_cfg.resume_session_id.clone_from(&req.resume);
headless_cfg.env = req
.env
.iter()
.map(|(key, value)| (key.clone(), value.clone()))
.collect();
let runner = seher::claude_headless::ClaudeHeadlessRunner::new(headless_cfg);
seher::claude_headless::stream_headless(
runner,
req.prompt.to_string(),
resolved.provider.clone(),
)
}
"claude-terminal" => {
let sdk = seher::claude_terminal::new_sdk_with_defaults(
None,
None,
claude_family_model(&resolved.model_id),
None,
claude_family_effort(&resolved.model_id, resolved.effort),
None,
cwd_string,
req.env.clone(),
);
seher::claude_terminal::stream_via_thread(
sdk,
req.prompt.to_string(),
resolved.provider.clone(),
req.resume.clone(),
)
}
"pi" | "omp" | "pi-rust" => {
let resume = match resolved.sdk.as_str() {
"omp" | "pi" => None,
"pi-rust" => req.resume.as_deref().and_then(|id| {
seher::sdk::pi_session_path(req.working_dir, id)
.is_file()
.then(|| id.to_string())
}),
_ => unreachable!("shared RPC/PiRust dispatch branch"),
};
let mut resolved = resolved.clone();
if matches!(resolved.sdk.as_str(), "omp" | "pi") {
merge_helper_env(&mut resolved, req.env);
} else {
resolved
.env
.extend(req.env.iter().map(|(k, v)| (k.clone(), v.clone())));
}
seher::sdk::stream_for_resolved(
&resolved,
req.prompt.to_string(),
seher::sdk::RunAgentOptions {
working_dir: req.working_dir.map(Path::to_path_buf),
resume,
tools: req.tools.clone(),
cancel: omp_cancel,
..Default::default()
},
)
}
other => unreachable!(
"seher resolver returned an unsupported sdk kind: {other:?} \
(cruise dispatch is out of sync with the seher version in use)"
),
}
}
fn merge_helper_env(
resolved: &mut seher::sdk::ResolvedAgent,
request_env: &HashMap<String, String>,
) {
for (key, value) in std::env::vars() {
resolved.env.entry(key).or_insert(value);
}
for (key, value) in request_env {
if matches!(key.as_str(), "PATH" | "PATHEXT") {
continue;
}
resolved.env.insert(key.clone(), value.clone());
}
}
fn finish_sdk_session(
sdk: &str,
working_dir: Option<&Path>,
session: Option<String>,
) -> Option<String> {
match sdk {
"omp" => {
if let Some(session_id) = session.as_deref() {
let _ = seher::sdk::close_omp_session(session_id, working_dir);
}
None
}
"pi" => {
if let Some(session_id) = session.as_deref() {
let _ = seher::sdk::close_pi_session(session_id, working_dir);
}
None
}
_ => session,
}
}
async fn run_sdk(req: PromptRun<'_>) -> Result<PromptOutcome> {
let mode_key = req
.model_or_mode
.unwrap_or(DEFAULT_STEP_MODE_KEY)
.to_string();
let on_delta = req.stream.and_then(|s| s.on_stdout);
let require_tools = !req.tools.is_empty() || req.resume.is_some();
let mut attempts = 0;
loop {
if let Some(cb) = req.on_notice {
cb(&resolving_notice(&mode_key, require_tools));
}
let cancel_flag = Arc::new(AtomicBool::new(false));
let abort_guard = AbortOnDrop(Arc::clone(&cancel_flag));
let resolved = tokio::select! {
biased;
() = maybe_cancelled(req.cancel_token) => return Err(CruiseError::Interrupted),
out = resolve_provider(mode_key.clone(), require_tools, cancel_flag) => out,
}?;
drop(abort_guard);
if let Some(cb) = req.on_notice {
cb(&resolution_notice(&resolved));
}
let omp_cancel = seher::sdk::CancelToken::new();
let _omp_cancel_guard = CancelOnDrop(omp_cancel.clone());
let session_slot =
matches!(resolved.sdk.as_str(), "omp" | "pi").then(|| Arc::new(Mutex::new(None)));
let rx_std = spawn_agent_stream(&resolved, &req, omp_cancel);
let outcome =
match stream_to_outcome(rx_std, on_delta, req.cancel_token, session_slot.clone()).await
{
Ok(outcome) => outcome,
Err(error) => {
if matches!(&error, CruiseError::Interrupted) {
let session = session_slot
.as_ref()
.and_then(|slot| slot.lock().ok().and_then(|session| session.clone()));
let _ = finish_sdk_session(&resolved.sdk, req.working_dir, session);
}
return Err(error);
}
};
match outcome {
ChunkOutcome::Done { output, session } => {
let session = finish_sdk_session(&resolved.sdk, req.working_dir, session);
return Ok(PromptOutcome {
result: PromptResult {
output,
stderr: String::new(),
},
session_id: session,
});
}
ChunkOutcome::Failed { message, session } => {
let _ = finish_sdk_session(&resolved.sdk, req.working_dir, session);
return Err(CruiseError::CommandError(message));
}
ChunkOutcome::Limited { message, session } => {
let _ = finish_sdk_session(&resolved.sdk, req.working_dir, session);
if attempts < req.max_retries {
attempts += 1;
if let Some(cb) = req.on_notice {
cb(&rate_limited_notice(&resolved, attempts, req.max_retries));
}
continue;
}
return Err(CruiseError::CommandError(message));
}
ChunkOutcome::Closed { session, .. } => {
let _ = finish_sdk_session(&resolved.sdk, req.working_dir, session);
return Err(CruiseError::Other(
"seher stream closed before completion".to_string(),
));
}
}
}
}
async fn stream_to_outcome(
rx_std: std::sync::mpsc::Receiver<StreamChunk>,
on_delta: Option<&(dyn Fn(&str) + Send + Sync)>,
cancel_token: Option<&CancellationToken>,
session_slot: Option<Arc<Mutex<Option<String>>>>,
) -> Result<ChunkOutcome> {
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<StreamChunk>();
let bridge_session_slot = session_slot;
std::thread::spawn(move || {
while let Ok(chunk) = rx_std.recv() {
if let StreamChunk::Session(id) = &chunk
&& let Some(slot) = &bridge_session_slot
&& let Ok(mut session) = slot.lock()
{
*session = Some(id.clone());
}
if tx.send(chunk).is_err() {
break;
}
}
});
let mut line_buf = LineBuffer::new();
let mut reducer = ChunkReducer::new();
let outcome = loop {
tokio::select! {
biased;
() = maybe_cancelled(cancel_token) => return Err(CruiseError::Interrupted),
maybe = rx.recv() => match maybe {
Some(chunk) => {
let mut sink = |d: &str| {
if let Some(cb) = on_delta {
line_buf.push(d, cb);
}
};
if let Some(out) = reducer.step(chunk, &mut sink) {
break out;
}
}
None => break reducer.finish(),
}
}
};
if let Some(cb) = on_delta {
line_buf.flush(cb);
}
Ok(outcome)
}
async fn run_pi_direct(req: PromptRun<'_>) -> Result<PromptOutcome> {
let opts = build_pi_options(&req, req.model_or_mode)?;
let runner = PiRunner::new(opts);
let on_delta = req.stream.and_then(|s| s.on_stdout);
let mut attempts = 0;
loop {
let rx_std = runner.stream(req.prompt.to_string(), req.resume.clone());
let outcome = stream_to_outcome(rx_std, on_delta, req.cancel_token, None).await?;
match outcome {
ChunkOutcome::Done { output, session } => {
return Ok(PromptOutcome {
result: PromptResult {
output,
stderr: String::new(),
},
session_id: session,
});
}
ChunkOutcome::Failed { message, .. } => return Err(CruiseError::CommandError(message)),
ChunkOutcome::Limited { message, .. } => {
if attempts < req.max_retries {
attempts += 1;
let delay = crate::step::command::calculate_backoff(attempts);
if let Some(cb) = req.on_notice {
cb(&format!(
"Rate limit detected. Retrying in {:.1}s... ({attempts}/{})",
delay.as_secs_f64(),
req.max_retries
));
}
tokio::select! {
biased;
() = maybe_cancelled(req.cancel_token) => return Err(CruiseError::Interrupted),
() = tokio::time::sleep(delay) => {}
}
continue;
}
return Err(CruiseError::CommandError(message));
}
ChunkOutcome::Closed { .. } => {
return Err(CruiseError::Other(
"pi stream closed before completion".to_string(),
));
}
}
}
}
fn build_pi_options(req: &PromptRun<'_>, model_ref: Option<&str>) -> Result<PiRunnerOptions> {
let (provider, model, thinking) = parse_pi_model_ref(model_ref)?;
Ok(PiRunnerOptions {
provider,
model,
api_key: None,
thinking,
system_prompt: None,
working_directory: req.working_dir.map(Path::to_path_buf),
env: req
.env
.iter()
.map(|(key, value)| (key.clone(), value.clone()))
.collect(),
tools: req.tools.clone(),
})
}
fn parse_pi_model_ref(
model_ref: Option<&str>,
) -> Result<(Option<String>, Option<String>, Option<String>)> {
let Some(raw) = model_ref.map(str::trim).filter(|s| !s.is_empty()) else {
return Ok((None, None, None));
};
let (without_thinking, thinking) = split_thinking_suffix(raw);
if without_thinking.is_empty() {
return Ok((None, None, thinking.map(str::to_string)));
}
match without_thinking.split_once('/') {
Some((provider, model)) if !provider.is_empty() && !model.is_empty() => Ok((
Some(provider.to_string()),
Some(model.to_string()),
thinking.map(str::to_string),
)),
Some(_) => Err(CruiseError::Other(format!(
"invalid pi model reference '{raw}': provider and model must both be non-empty \
around '/' (expected \"provider/model[:thinking]\", \"model[:thinking]\", or \
empty for auto-selection)"
))),
None => Ok((
None,
Some(without_thinking.to_string()),
thinking.map(str::to_string),
)),
}
}
pub(crate) struct LineBuffer {
pending: String,
}
impl LineBuffer {
pub(crate) fn new() -> Self {
Self {
pending: String::new(),
}
}
pub(crate) fn push<F: FnMut(&str)>(&mut self, frag: &str, mut emit: F) {
self.pending.push_str(frag);
while let Some(idx) = self.pending.find('\n') {
let rest = self.pending.split_off(idx + 1);
let mut line = std::mem::replace(&mut self.pending, rest);
line.pop(); if line.ends_with('\r') {
line.pop();
}
emit(&line);
}
}
pub(crate) fn flush<F: FnMut(&str)>(&mut self, mut emit: F) {
if !self.pending.is_empty() {
emit(&self.pending);
self.pending.clear();
}
}
}
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum ChunkOutcome {
Done {
output: String,
session: Option<String>,
},
Limited {
message: String,
session: Option<String>,
},
Failed {
message: String,
session: Option<String>,
},
Closed {
partial: String,
session: Option<String>,
},
}
pub(crate) struct ChunkReducer {
buf: String,
session: Option<String>,
}
impl ChunkReducer {
pub(crate) fn new() -> Self {
Self {
buf: String::new(),
session: None,
}
}
pub(crate) fn step<F: FnMut(&str)>(
&mut self,
chunk: StreamChunk,
on_delta: &mut F,
) -> Option<ChunkOutcome> {
match chunk {
StreamChunk::Delta(d) => {
on_delta(&d);
self.buf.push_str(&d);
None
}
StreamChunk::Session(id) => {
self.session = Some(id);
None
}
StreamChunk::Done(text) => {
let output = if text.is_empty() {
std::mem::take(&mut self.buf)
} else {
text
};
Some(ChunkOutcome::Done {
output,
session: self.session.take(),
})
}
StreamChunk::Limit(e) => Some(ChunkOutcome::Limited {
message: e.to_string(),
session: self.session.take(),
}),
StreamChunk::Error(msg) => Some(ChunkOutcome::Failed {
message: msg,
session: self.session.take(),
}),
}
}
pub(crate) fn finish(&mut self) -> ChunkOutcome {
ChunkOutcome::Closed {
partial: std::mem::take(&mut self.buf),
session: self.session.take(),
}
}
}
#[must_use]
pub fn mode_key_for_step(step_model: Option<&str>, global_model: Option<&str>) -> String {
step_model
.or(global_model)
.unwrap_or(DEFAULT_STEP_MODE_KEY)
.to_string()
}
#[must_use]
pub fn mode_key_for_plan(plan_model: Option<&str>, global_model: Option<&str>) -> String {
plan_model
.or(global_model)
.unwrap_or(DEFAULT_PLAN_MODE_KEY)
.to_string()
}
fn effort_from_suffix(suffix: &str) -> Option<EffortLevel> {
match suffix.trim().to_lowercase().as_str() {
"minimal" | "min" | "low" | "1" => Some(EffortLevel::Low),
"medium" | "med" | "2" => Some(EffortLevel::Medium),
"high" | "3" => Some(EffortLevel::High),
"xhigh" | "4" => Some(EffortLevel::XHigh),
"max" => Some(EffortLevel::Max),
_ => None,
}
}
fn claude_family_model(model_id: &str) -> Option<String> {
let (model, _) = split_thinking_suffix(model_id);
if model.is_empty() {
None
} else {
Some(model.to_string())
}
}
fn claude_family_effort(
model_id: &str,
resolved_effort: Option<EffortLevel>,
) -> Option<EffortLevel> {
let (_, suffix_thinking) = split_thinking_suffix(model_id);
resolved_effort.or_else(|| suffix_thinking.and_then(effort_from_suffix))
}
fn effective_model(resolved: &seher::sdk::ResolvedAgent) -> (Cow<'_, str>, Option<EffortLevel>) {
if matches!(
resolved.sdk.as_str(),
"claude" | "claude-headless" | "claude-terminal"
) {
let model =
claude_family_model(&resolved.model_id).map_or(Cow::Borrowed("default"), Cow::Owned);
(
model,
claude_family_effort(&resolved.model_id, resolved.effort),
)
} else {
let (_, model, suffix) =
seher::sdk::split_model_ref(&resolved.provider, &resolved.model_id);
(
Cow::Owned(model),
resolved
.effort
.or_else(|| suffix.as_deref().and_then(effort_from_suffix)),
)
}
}
fn resolving_notice(mode_key: &str, require_tools: bool) -> String {
let tools_note = if require_tools {
" (tool-capable providers only)"
} else {
""
};
format!("seher: resolving provider for mode \"{mode_key}\"{tools_note}")
}
fn rate_limited_notice(
resolved: &seher::sdk::ResolvedAgent,
attempts: usize,
max_retries: usize,
) -> String {
let (model, _) = effective_model(resolved);
format!(
"seher: provider={} model={model} rate-limited; re-resolving... ({attempts}/{max_retries})",
resolved.provider
)
}
fn resolution_notice(resolved: &seher::sdk::ResolvedAgent) -> String {
let (model, effort) = effective_model(resolved);
let mut msg = format!(
"seher: selected provider={} model={model} sdk={} mode={}",
resolved.provider, resolved.sdk, resolved.mode_key
);
if let Some(effort) = effort {
msg.push_str(" effort=");
msg.push_str(effort.as_str());
}
msg
}
#[test]
fn resolution_notice_reports_provider_model_sdk_mode_and_effort() {
let resolved = seher::sdk::ResolvedAgent {
provider: "codex".to_string(),
model_id: "openai-codex/gpt-5.6-luna:high".to_string(),
mode_key: "build".to_string(),
sdk: "pi".to_string(),
api: None,
skills: seher::sdk::ResolvedSkillsConfig::default(),
retry: seher::sdk::RetryConfig::default(),
env: indexmap::IndexMap::default(),
effort: Some(EffortLevel::High),
};
assert_eq!(
resolution_notice(&resolved),
"seher: selected provider=codex model=gpt-5.6-luna sdk=pi mode=build effort=high"
);
}
#[test]
fn resolution_notice_splits_thinking_suffix_for_claude_family() {
let resolved = seher::sdk::ResolvedAgent {
provider: "claude".to_string(),
model_id: "claude-sonnet-4-6:xhigh".to_string(),
mode_key: "plan".to_string(),
sdk: "claude-terminal".to_string(),
api: None,
skills: seher::sdk::ResolvedSkillsConfig::default(),
retry: seher::sdk::RetryConfig::default(),
env: indexmap::IndexMap::default(),
effort: None,
};
assert_eq!(
resolution_notice(&resolved),
"seher: selected provider=claude model=claude-sonnet-4-6 sdk=claude-terminal mode=plan effort=xhigh"
);
}
#[test]
fn resolution_notice_reports_claude_default_model() {
let resolved = seher::sdk::ResolvedAgent {
provider: "claude".to_string(),
model_id: ":high".to_string(),
mode_key: "build".to_string(),
sdk: "claude-terminal".to_string(),
api: None,
skills: seher::sdk::ResolvedSkillsConfig::default(),
retry: seher::sdk::RetryConfig::default(),
env: indexmap::IndexMap::default(),
effort: None,
};
assert_eq!(
resolution_notice(&resolved),
"seher: selected provider=claude model=default sdk=claude-terminal mode=build effort=high"
);
}
#[test]
fn resolving_notice_reports_tool_requirement() {
assert_eq!(
resolving_notice("build", false),
"seher: resolving provider for mode \"build\""
);
assert_eq!(
resolving_notice("build", true),
"seher: resolving provider for mode \"build\" (tool-capable providers only)"
);
}
#[test]
fn rate_limited_notice_reports_effective_model() {
let resolved = seher::sdk::ResolvedAgent {
provider: "codex".to_string(),
model_id: "openai-codex/gpt-5.6-luna:high".to_string(),
mode_key: "build".to_string(),
sdk: "pi".to_string(),
api: None,
skills: seher::sdk::ResolvedSkillsConfig::default(),
retry: seher::sdk::RetryConfig::default(),
env: indexmap::IndexMap::default(),
effort: None,
};
assert_eq!(
rate_limited_notice(&resolved, 1, 2),
"seher: provider=codex model=gpt-5.6-luna rate-limited; re-resolving... (1/2)"
);
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn step_mode_key_prefers_step_over_global() {
assert_eq!(mode_key_for_step(Some("fast"), Some("build")), "fast");
}
#[test]
fn step_mode_key_falls_back_to_global() {
assert_eq!(mode_key_for_step(None, Some("build")), "build");
}
#[test]
fn step_mode_key_defaults_to_build() {
assert_eq!(mode_key_for_step(None, None), DEFAULT_STEP_MODE_KEY);
assert_eq!(mode_key_for_step(None, None), "build");
}
#[test]
fn plan_mode_key_prefers_plan_model() {
assert_eq!(mode_key_for_plan(Some("plan"), Some("build")), "plan");
}
#[test]
fn plan_mode_key_falls_back_to_global_model() {
assert_eq!(mode_key_for_plan(None, Some("build")), "build");
}
#[test]
fn plan_mode_key_defaults_to_plan() {
assert_eq!(mode_key_for_plan(None, None), DEFAULT_PLAN_MODE_KEY);
assert_eq!(mode_key_for_plan(None, None), "plan");
}
#[test]
fn claude_family_model_strips_thinking_suffix() {
assert_eq!(
claude_family_model("claude-sonnet-4-5:high").as_deref(),
Some("claude-sonnet-4-5")
);
}
#[test]
fn claude_family_effort_uses_suffix_when_unresolved() {
assert_eq!(
claude_family_effort("claude-sonnet-4-5:med", None),
Some(EffortLevel::Medium)
);
}
#[test]
fn claude_family_effort_prefers_resolved_effort_over_suffix() {
assert_eq!(
claude_family_effort("claude-sonnet-4-5:low", Some(EffortLevel::High)),
Some(EffortLevel::High)
);
}
#[test]
fn claude_family_effort_omits_off_suffix() {
assert_eq!(
claude_family_model("claude-sonnet-4-5:off").as_deref(),
Some("claude-sonnet-4-5")
);
assert_eq!(claude_family_effort("claude-sonnet-4-5:off", None), None);
}
fn sdk_executor() -> Executor {
Executor::Sdk
}
fn command_executor() -> Executor {
Executor::Command {
command: vec!["claude".to_string(), "-p".to_string()],
}
}
fn pi_executor() -> Executor {
Executor::Pi
}
#[test]
fn new_picks_sdk_when_sdk_set() {
let e = Executor::new(Some("seher"), &[]);
assert!(e.is_sdk());
assert!(matches!(e, Executor::Sdk));
}
#[test]
fn new_picks_sdk_for_any_non_pi_sdk_value() {
let e = Executor::new(Some("claude-terminal"), &[]);
assert!(matches!(e, Executor::Sdk));
}
#[test]
fn new_picks_pi_when_sdk_is_pi() {
let e = Executor::new(Some("pi"), &[]);
assert!(e.is_sdk());
assert!(matches!(e, Executor::Pi));
}
#[test]
fn new_picks_command_when_sdk_unset() {
let e = Executor::new(None, &["claude".to_string()]);
assert!(!e.is_sdk());
}
#[test]
fn command_step_model_passes_through_model_name() {
let e = command_executor();
assert_eq!(
e.step_model_or_mode(Some("sonnet"), Some("opus")),
Some("sonnet".to_string())
);
assert_eq!(e.step_model_or_mode(None, None), None);
}
#[test]
fn sdk_step_model_maps_to_mode_key_with_default() {
let e = sdk_executor();
assert_eq!(
e.step_model_or_mode(Some("fast"), None),
Some("fast".to_string())
);
assert_eq!(e.step_model_or_mode(None, None), Some("build".to_string()));
}
#[test]
fn sdk_plan_model_maps_to_plan_mode_key_with_default() {
let e = sdk_executor();
assert_eq!(e.plan_model_or_mode(None, None), Some("plan".to_string()));
assert_eq!(
e.plan_model_or_mode(None, Some("build")),
Some("build".to_string())
);
}
#[test]
fn pi_step_model_passes_through_model_reference() {
let e = pi_executor();
assert_eq!(
e.step_model_or_mode(Some("anthropic/claude-sonnet-4-6"), Some("opus")),
Some("anthropic/claude-sonnet-4-6".to_string())
);
assert_eq!(
e.step_model_or_mode(None, Some("opus")),
Some("opus".to_string())
);
assert_eq!(e.step_model_or_mode(None, None), None);
}
#[test]
fn pi_plan_model_passes_through_model_reference() {
let e = pi_executor();
assert_eq!(
e.plan_model_or_mode(Some("openai/gpt-5.5"), None),
Some("openai/gpt-5.5".to_string())
);
assert_eq!(e.plan_model_or_mode(None, None), None);
}
#[test]
fn parse_pi_model_ref_none_when_unset() {
assert_eq!(
parse_pi_model_ref(None).unwrap_or_else(|e| panic!("unexpected error: {e}")),
(None, None, None)
);
assert_eq!(
parse_pi_model_ref(Some("")).unwrap_or_else(|e| panic!("unexpected error: {e}")),
(None, None, None)
);
assert_eq!(
parse_pi_model_ref(Some(" ")).unwrap_or_else(|e| panic!("unexpected error: {e}")),
(None, None, None)
);
}
#[test]
fn parse_pi_model_ref_bare_model_leaves_provider_unset() {
assert_eq!(
parse_pi_model_ref(Some("claude-sonnet-4-6"))
.unwrap_or_else(|e| panic!("unexpected error: {e}")),
(None, Some("claude-sonnet-4-6".to_string()), None)
);
}
#[test]
fn parse_pi_model_ref_splits_provider_and_model() {
assert_eq!(
parse_pi_model_ref(Some("anthropic/claude-sonnet-4-6"))
.unwrap_or_else(|e| panic!("unexpected error: {e}")),
(
Some("anthropic".to_string()),
Some("claude-sonnet-4-6".to_string()),
None
)
);
}
#[test]
fn parse_pi_model_ref_extracts_thinking_suffix() {
assert_eq!(
parse_pi_model_ref(Some("openai-codex/gpt-5.5:xhigh"))
.unwrap_or_else(|e| panic!("unexpected error: {e}")),
(
Some("openai-codex".to_string()),
Some("gpt-5.5".to_string()),
Some("xhigh".to_string())
)
);
}
#[test]
fn parse_pi_model_ref_keeps_non_thinking_colon_suffix_in_model() {
assert_eq!(
parse_pi_model_ref(Some("openrouter/meta-llama/llama-3-8b:free"))
.unwrap_or_else(|e| panic!("unexpected error: {e}")),
(
Some("openrouter".to_string()),
Some("meta-llama/llama-3-8b:free".to_string()),
None
)
);
}
#[test]
fn parse_pi_model_ref_bare_model_with_thinking_suffix() {
assert_eq!(
parse_pi_model_ref(Some("claude-sonnet-4-6:high"))
.unwrap_or_else(|e| panic!("unexpected error: {e}")),
(
None,
Some("claude-sonnet-4-6".to_string()),
Some("high".to_string())
)
);
}
#[test]
fn parse_pi_model_ref_thinking_only_means_auto_model() {
assert_eq!(
parse_pi_model_ref(Some(":high")).unwrap_or_else(|e| panic!("unexpected error: {e}")),
(None, None, Some("high".to_string()))
);
}
#[test]
fn parse_pi_model_ref_rejects_empty_provider_or_model_around_slash() {
for bad in ["/claude-sonnet-4-6", "anthropic/", "/", "anthropic/:high"] {
match parse_pi_model_ref(Some(bad)) {
Err(err) => assert!(
err.to_string().contains("invalid pi model reference"),
"unexpected error message for {bad:?}: {err}"
),
Ok(parsed) => panic!("expected parse error for {bad:?}, got {parsed:?}"),
}
}
}
#[cfg(unix)]
#[tokio::test]
async fn sdk_run_reports_resolved_provider_through_on_notice() {
use std::os::unix::fs::PermissionsExt;
let _lock = crate::test_support::lock_process();
let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
let _home_guards = crate::test_support::set_fake_home(dir.path());
let bin_dir = dir.path().join("bin");
std::fs::create_dir(&bin_dir).unwrap_or_else(|e| panic!("create bin dir: {e}"));
let script = bin_dir.join("omp");
let tool_marker = dir.path().join("tool-registered");
std::fs::write(
&script,
r#"#!/bin/sh
printf '%s\n' '{"type":"ready","protocolVersion":1}'
while IFS= read -r line; do
case "$line" in
*get_state*) printf '%s\n' '{"id":"seher-handshake","type":"response","command":"get_state","success":true,"data":{"sessionId":"omp-test-session"}}' ;;
*set_host_tools*) printf '%s\n' '{"id":"seher-host-tools","type":"response","command":"set_host_tools","success":true,"data":{"toolNames":["echo"]}}'; : > "$OMP_TOOL_MARKER" ;;
*prompt*) if [ ! -f "$OMP_TOOL_MARKER" ]; then exit 42; fi; printf '%s\n' '{"id":"seher-prompt","type":"response","command":"prompt","success":true}' '{"type":"message_update","assistantMessageEvent":{"type":"text_delta","delta":"ok"}}' '{"type":"agent_end","isTerminal":true,"messages":[]}' ;;
*abort*) exit 0 ;;
esac
done
"#,
)
.unwrap_or_else(|e| panic!("write fake OMP: {e}"));
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o700))
.unwrap_or_else(|e| panic!("chmod fake OMP: {e}"));
for name in ["bunx", "npx"] {
let blocker = bin_dir.join(name);
std::fs::write(&blocker, "#!/bin/sh\nexit 97\n")
.unwrap_or_else(|e| panic!("write {name} blocker: {e}"));
std::fs::set_permissions(&blocker, std::fs::Permissions::from_mode(0o700))
.unwrap_or_else(|e| panic!("chmod {name} blocker: {e}"));
}
let codexbar = dir.path().join("codexbar");
std::fs::write(&codexbar, "#!/bin/sh\nexit 1\n")
.unwrap_or_else(|e| panic!("write codexbar stub: {e}"));
std::fs::set_permissions(&codexbar, std::fs::Permissions::from_mode(0o700))
.unwrap_or_else(|e| panic!("chmod codexbar stub: {e}"));
let config_path = dir.path().join("seher.yaml");
std::fs::write(
&config_path,
format!(
"providers:\n test-provider:\n sdk: omp\n env:\n PATH: {}\n OMP_TOOL_MARKER: {}\n models:\n build: test-provider/test-model\n",
bin_dir.display(),
tool_marker.display()
),
)
.unwrap_or_else(|e| panic!("write seher config: {e}"));
let _codexbar = crate::test_support::EnvGuard::set("SEHER_CODEXBAR_BIN", &codexbar);
let _seher_config = crate::test_support::EnvGuard::set("SEHER_CONFIG", &config_path);
let notices = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let notices_sink = std::sync::Arc::clone(¬ices);
let on_notice = move |msg: &str| {
notices_sink
.lock()
.unwrap_or_else(|e| panic!("notices lock: {e}"))
.push(msg.to_string());
};
let env = HashMap::new();
let echo = SeherTool::new(
"echo",
"Echo",
serde_json::json!({"type": "object"}),
std::sync::Arc::new(|input| Ok(input.to_string())),
);
let outcome = Executor::Sdk
.run(PromptRun {
prompt: "hello",
model_or_mode: Some("build"),
max_retries: 0,
env: &env,
on_notice: Some(&on_notice),
cancel_token: None,
working_dir: Some(dir.path()),
stream: None,
tools: vec![echo],
resume: None,
})
.await
.unwrap_or_else(|e| panic!("SDK run: {e}"));
assert_eq!(outcome.result.output, "ok");
let notices = notices
.lock()
.unwrap_or_else(|e| panic!("notices lock: {e}"));
assert_eq!(
notices[1],
"seher: selected provider=test-provider model=test-model sdk=omp mode=build"
);
}
fn base_req(env: &HashMap<String, String>) -> PromptRun<'_> {
PromptRun {
prompt: "hi",
model_or_mode: None,
max_retries: 0,
env,
on_notice: None,
cancel_token: None,
working_dir: None,
stream: None,
tools: Vec::new(),
resume: None,
}
}
#[test]
fn build_pi_options_leaves_api_key_none_for_pi_to_resolve() {
let env = HashMap::new();
let req = base_req(&env);
let opts = build_pi_options(&req, Some("anthropic/claude-sonnet-4-6"))
.unwrap_or_else(|e| panic!("unexpected error: {e}"));
assert_eq!(opts.provider.as_deref(), Some("anthropic"));
assert_eq!(opts.model.as_deref(), Some("claude-sonnet-4-6"));
assert!(opts.api_key.is_none());
assert!(opts.thinking.is_none());
}
#[test]
fn build_pi_options_forwards_tools_env_and_working_dir() {
let mut env = HashMap::new();
env.insert("FOO".to_string(), "bar".to_string());
let tool = SeherTool::new(
"echo",
"Echo",
serde_json::json!({"type": "object"}),
std::sync::Arc::new(|_| Ok(String::new())),
);
let dir = std::path::PathBuf::from("/tmp/cruise-pi-test");
let req = PromptRun {
prompt: "hi",
model_or_mode: Some("gpt-5.5"),
max_retries: 0,
env: &env,
on_notice: None,
cancel_token: None,
working_dir: Some(&dir),
stream: None,
tools: vec![tool],
resume: Some("sess-1".to_string()),
};
let opts = build_pi_options(&req, req.model_or_mode)
.unwrap_or_else(|e| panic!("unexpected error: {e}"));
assert_eq!(opts.working_directory, Some(dir));
assert_eq!(opts.env.get("FOO").map(String::as_str), Some("bar"));
assert_eq!(opts.tools.len(), 1);
assert_eq!(opts.tools[0].name, "echo");
}
#[cfg(unix)]
#[test]
fn omp_dispatch_streams_through_cruise_dispatch() {
use std::os::unix::fs::PermissionsExt;
let _lock = crate::test_support::lock_process();
let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
let _home_guards = crate::test_support::set_fake_home(dir.path());
let bin_dir = dir.path().join("bin");
std::fs::create_dir(&bin_dir).unwrap_or_else(|e| panic!("create bin dir: {e}"));
let script = bin_dir.join("omp");
let tool_marker = dir.path().join("tool-registered");
std::fs::write(
&script,
r#"#!/bin/sh
printf '%s\n' '{"type":"ready","protocolVersion":1}'
while IFS= read -r line; do
case "$line" in
*get_state*) printf '{"id":"seher-handshake","type":"response","command":"get_state","success":true,"data":{"sessionId":"omp-test-session"}}\n' ;;
*set_host_tools*) printf '%s\n' '{"id":"seher-host-tools","type":"response","command":"set_host_tools","success":true,"data":{"toolNames":["echo"]}}'; : > "$OMP_TOOL_MARKER" ;;
*prompt*) if [ ! -f "$OMP_TOOL_MARKER" ]; then exit 42; fi; printf '%s\n' '{"id":"seher-prompt","type":"response","command":"prompt","success":true}' '{"type":"message_update","assistantMessageEvent":{"type":"text_delta","delta":"ok"}}' '{"type":"agent_end","isTerminal":true,"messages":[]}' ;;
*abort*) exit 0 ;;
esac
done
"#,
)
.unwrap_or_else(|e| panic!("write fake OMP: {e}"));
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o700))
.unwrap_or_else(|e| panic!("chmod fake OMP: {e}"));
for name in ["bunx", "npx"] {
let blocker = bin_dir.join(name);
std::fs::write(&blocker, "#!/bin/sh\nexit 97\n")
.unwrap_or_else(|e| panic!("write {name} blocker: {e}"));
std::fs::set_permissions(&blocker, std::fs::Permissions::from_mode(0o700))
.unwrap_or_else(|e| panic!("chmod {name} blocker: {e}"));
}
let env = HashMap::new();
let echo = SeherTool::new(
"echo",
"Echo",
serde_json::json!({"type": "object"}),
std::sync::Arc::new(|input| Ok(input.to_string())),
);
let req = PromptRun {
prompt: "hello",
model_or_mode: None,
max_retries: 0,
env: &env,
on_notice: None,
cancel_token: None,
working_dir: Some(dir.path()),
stream: None,
tools: vec![echo],
resume: None,
};
let resolved = seher::sdk::ResolvedAgent {
provider: "test-provider".to_string(),
model_id: "test-provider/test-model:high".to_string(),
mode_key: "build".to_string(),
sdk: "omp".to_string(),
api: None,
skills: seher::sdk::ResolvedSkillsConfig::default(),
retry: seher::sdk::RetryConfig::default(),
env: [
(String::from("PATH"), bin_dir.display().to_string()),
(
String::from("OMP_TOOL_MARKER"),
tool_marker.display().to_string(),
),
]
.into(),
effort: None,
};
let rx = spawn_agent_stream(&resolved, &req, seher::sdk::CancelToken::new());
let mut output = String::new();
let mut session = None;
loop {
let chunk = rx
.recv_timeout(std::time::Duration::from_secs(5))
.unwrap_or_else(|e| panic!("OMP stream did not finish: {e}"));
match chunk {
StreamChunk::Session(id) => session = Some(id),
StreamChunk::Delta(delta) => output.push_str(&delta),
StreamChunk::Done(done) => {
if !done.is_empty() {
output = done;
}
break;
}
StreamChunk::Error(message) => panic!("unexpected OMP error: {message}"),
StreamChunk::Limit(_) => panic!("unexpected OMP rate limit"),
}
}
assert_eq!(output, "ok");
assert_eq!(session.as_deref(), Some("omp-test-session"));
let session_id = session.unwrap_or_else(|| panic!("OMP session id missing"));
assert!(finish_sdk_session("omp", Some(dir.path()), Some(session_id.clone())).is_none());
assert!(!seher::sdk::close_omp_session(
&session_id,
Some(dir.path())
));
}
#[test]
fn rpc_backend_env_does_not_override_helper_search_path() {
let _lock = crate::test_support::lock_process();
let _ambient = crate::test_support::EnvGuard::set("CRUISE_RPC_AMBIENT", "ambient");
let _pathext = crate::test_support::EnvGuard::remove("PATHEXT");
let mut resolved = seher::sdk::ResolvedAgent {
provider: "test-provider".to_string(),
model_id: "test-provider/test-model".to_string(),
mode_key: "build".to_string(),
sdk: "omp".to_string(),
api: None,
skills: seher::sdk::ResolvedSkillsConfig::default(),
retry: seher::sdk::RetryConfig::default(),
env: [("PATH".to_string(), "/trusted/bin".to_string())].into(),
effort: None,
};
let request_env = [
("PATH".to_string(), "/repo/bin".to_string()),
("PATHEXT".to_string(), ".COM".to_string()),
("PROJECT".to_string(), "cruise".to_string()),
]
.into_iter()
.collect();
merge_helper_env(&mut resolved, &request_env);
assert_eq!(
resolved.env.get("PATH").map(String::as_str),
Some("/trusted/bin")
);
assert!(!resolved.env.contains_key("PATHEXT"));
assert_eq!(
resolved.env.get("PROJECT").map(String::as_str),
Some("cruise")
);
assert_eq!(
resolved.env.get("CRUISE_RPC_AMBIENT").map(String::as_str),
Some("ambient")
);
}
#[cfg(unix)]
#[test]
fn pi_dispatch_streams_through_external_pi_cli() {
use std::os::unix::fs::PermissionsExt;
let _lock = crate::test_support::lock_process();
let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
let _home_guards = crate::test_support::set_fake_home(dir.path());
let bin_dir = dir.path().join("bin");
std::fs::create_dir(&bin_dir).unwrap_or_else(|e| panic!("create bin dir: {e}"));
let script = bin_dir.join("pi");
std::fs::write(
&script,
r#"#!/bin/sh
sid=
extension=
previous=
for arg in "$@"; do
if [ "$previous" = "--session-id" ]; then sid="$arg"; fi
if [ "$previous" = "--extension" ]; then extension="$arg"; fi
previous="$arg"
done
while IFS= read -r line; do
case "$line" in
*get_state*) printf '{"id":"seher-handshake","type":"response","command":"get_state","success":true,"data":{"sessionId":"%s"}}\n' "$sid" ;;
*prompt*) if [ ! -f "$extension" ] || [ ! -s "$SEHER_PI_TOOL_SPEC" ] || [ -z "$SEHER_PI_BRIDGE_HOST" ] || [ -z "$SEHER_PI_BRIDGE_PORT" ] || [ -z "$SEHER_PI_BRIDGE_TOKEN" ]; then exit 42; fi; printf '%s\n' '{"id":"seher-prompt","type":"response","command":"prompt","success":true}' '{"type":"message_update","assistantMessageEvent":{"type":"text_delta","delta":"ok"}}' '{"type":"agent_settled"}' ;;
*abort*) exit 0 ;;
esac
done
"#,
)
.unwrap_or_else(|e| panic!("write fake pi: {e}"));
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o700))
.unwrap_or_else(|e| panic!("chmod fake pi: {e}"));
for name in ["bunx", "npx"] {
let blocker = bin_dir.join(name);
std::fs::write(&blocker, "#!/bin/sh\nexit 97\n")
.unwrap_or_else(|e| panic!("write {name} blocker: {e}"));
std::fs::set_permissions(&blocker, std::fs::Permissions::from_mode(0o700))
.unwrap_or_else(|e| panic!("chmod {name} blocker: {e}"));
}
let env: HashMap<String, String> =
[(String::from("PATH"), String::from("/nonexistent/bin"))].into();
let echo = SeherTool::new(
"echo",
"Echo",
serde_json::json!({"type": "object"}),
std::sync::Arc::new(|input| Ok(input.to_string())),
);
let req = PromptRun {
prompt: "hello",
model_or_mode: None,
max_retries: 0,
env: &env,
on_notice: None,
cancel_token: None,
working_dir: Some(dir.path()),
stream: None,
tools: vec![echo],
resume: Some("foreign-session".to_string()),
};
let resolved = seher::sdk::ResolvedAgent {
provider: "codex".to_string(),
model_id: "openai-codex/gpt-5.6-luna:high".to_string(),
mode_key: "build".to_string(),
sdk: "pi".to_string(),
api: None,
skills: seher::sdk::ResolvedSkillsConfig::default(),
retry: seher::sdk::RetryConfig::default(),
env: [(String::from("PATH"), bin_dir.display().to_string())].into(),
effort: None,
};
let rx = spawn_agent_stream(&resolved, &req, seher::sdk::CancelToken::new());
let mut output = String::new();
let mut session = None;
loop {
let chunk = rx
.recv_timeout(std::time::Duration::from_secs(10))
.unwrap_or_else(|e| panic!("pi stream did not finish: {e}"));
match chunk {
StreamChunk::Session(id) => session = Some(id),
StreamChunk::Delta(delta) => output.push_str(&delta),
StreamChunk::Done(done) => {
if !done.is_empty() {
output = done;
}
break;
}
StreamChunk::Error(message) => panic!("unexpected pi error: {message}"),
StreamChunk::Limit(_) => panic!("unexpected pi rate limit"),
}
}
assert_eq!(output, "ok");
let session_id = session.unwrap_or_else(|| panic!("pi session id missing"));
assert_ne!(session_id, "foreign-session");
assert!(finish_sdk_session("pi", Some(dir.path()), Some(session_id.clone())).is_none());
assert!(!seher::sdk::close_pi_session(&session_id, Some(dir.path())));
}
#[test]
fn non_rpc_session_ids_remain_resumable() {
let session = Some("resumable-session".to_string());
assert_eq!(finish_sdk_session("claude", None, session.clone()), session);
}
#[test]
fn pi_backoff_matches_command_backoff_schedule() {
use crate::step::command::calculate_backoff;
assert_eq!(calculate_backoff(1), std::time::Duration::from_secs(2));
assert_eq!(calculate_backoff(2), std::time::Duration::from_secs(4));
assert_eq!(calculate_backoff(3), std::time::Duration::from_secs(8));
assert_eq!(calculate_backoff(10), std::time::Duration::from_mins(1));
}
fn no_sink() -> impl FnMut(&str) {
|_: &str| {}
}
#[test]
fn reducer_accumulates_deltas_and_captures_session() {
let mut r = ChunkReducer::new();
let mut collected = String::new();
let mut sink = |d: &str| collected.push_str(d);
assert_eq!(
r.step(StreamChunk::Session("sid-1".to_string()), &mut sink),
None
);
assert_eq!(
r.step(StreamChunk::Delta("Hello ".to_string()), &mut sink),
None
);
assert_eq!(
r.step(StreamChunk::Delta("world".to_string()), &mut sink),
None
);
let out = r
.step(StreamChunk::Done(String::new()), &mut sink)
.unwrap_or_else(|| panic!("expected terminal"));
assert_eq!(collected, "Hello world");
assert_eq!(
out,
ChunkOutcome::Done {
output: "Hello world".to_string(),
session: Some("sid-1".to_string()),
}
);
}
#[test]
fn reducer_done_text_overrides_buffered_deltas() {
let mut r = ChunkReducer::new();
let mut sink = no_sink();
r.step(StreamChunk::Delta("partial".to_string()), &mut sink);
let out = r
.step(StreamChunk::Done("FINAL".to_string()), &mut sink)
.unwrap_or_else(|| panic!("expected terminal"));
assert_eq!(
out,
ChunkOutcome::Done {
output: "FINAL".to_string(),
session: None,
}
);
}
#[test]
fn reducer_surfaces_error_chunk() {
let mut r = ChunkReducer::new();
let mut sink = no_sink();
let out = r
.step(StreamChunk::Error("boom".to_string()), &mut sink)
.unwrap_or_else(|| panic!("expected terminal"));
assert_eq!(
out,
ChunkOutcome::Failed {
message: "boom".to_string(),
session: None,
}
);
}
#[test]
fn reducer_surfaces_limit_chunk() {
use seher::sdk::errors::LimitError;
let mut r = ChunkReducer::new();
let mut sink = no_sink();
let out = r
.step(
StreamChunk::Limit(LimitError {
provider: "anthropic".to_string(),
reset_at: None,
}),
&mut sink,
)
.unwrap_or_else(|| panic!("expected terminal"));
match out {
ChunkOutcome::Limited { message, .. } => {
assert!(message.contains("anthropic"), "got: {message}");
}
other => panic!("expected Limited, got {other:?}"),
}
}
fn collect_lines(frags: &[&str]) -> (Vec<String>, Vec<String>) {
let mut lb = LineBuffer::new();
let mut lines = Vec::new();
for f in frags {
lb.push(f, |l| lines.push(l.to_string()));
}
let mut flushed = Vec::new();
lb.flush(|l| flushed.push(l.to_string()));
(lines, flushed)
}
#[test]
fn line_buffer_emits_complete_lines_and_flushes_remainder() {
let (lines, flushed) = collect_lines(&["Hel", "lo\nwor", "ld"]);
assert_eq!(lines, vec!["Hello".to_string()]);
assert_eq!(flushed, vec!["world".to_string()]);
}
#[test]
fn line_buffer_handles_multiple_lines_in_one_fragment() {
let (lines, flushed) = collect_lines(&["a\nb\nc\n"]);
assert_eq!(
lines,
vec!["a".to_string(), "b".to_string(), "c".to_string()]
);
assert!(flushed.is_empty(), "no partial line should remain");
}
#[test]
fn line_buffer_strips_carriage_return() {
let (lines, _) = collect_lines(&["x\r\n"]);
assert_eq!(lines, vec!["x".to_string()]);
}
#[test]
fn reducer_finish_reports_closed_with_partial() {
let mut r = ChunkReducer::new();
let mut sink = no_sink();
r.step(StreamChunk::Delta("half".to_string()), &mut sink);
assert_eq!(
r.finish(),
ChunkOutcome::Closed {
partial: "half".to_string(),
session: None,
}
);
}
}