//! Streaming path for OpenAI-compatible providers.
//!
//! Mirrors `ApiMethods::call_api_stream_inner` but speaks OpenAI chat/completions
//! and translates back to Anthropic-shaped events for the rest of the runtime.
use super::translate;
use super::types::{
ChatContentPart, ChatMessage, OaiEvent, ProviderConfig, StreamOptions, ToolCall,
};
use super::wire::StreamDecoder;
use crate::runtime::trace::openai as tr;
use crate::runtime::types::StreamEvent;
use futures::StreamExt;
use serde_json::{json, Value};
use tokio::sync::mpsc;
/// Trace capture for one batch of decoded Chat Completions events: marks the
/// first parsed model event and records provider-reported usage. Metadata
/// only — token counts, no content.
fn capture_oai_trace_signals(
events: &[OaiEvent],
attempt: &mut tr::StreamAttempt,
usage: &mut Option<crate::runtime::trace::UsageMeta>,
) {
for ev in events {
if !matches!(ev, OaiEvent::Warning(_)) {
attempt.mark_first_model_event();
}
if let OaiEvent::Usage {
prompt_tokens,
completion_tokens,
cached_tokens,
} = ev
{
*usage = Some(tr::provider_usage(
u64::from(*prompt_tokens),
u64::from(*completion_tokens),
u64::from(*cached_tokens),
));
}
}
}
/// Persistent retry posture for the Codex transport, mirroring the Anthropic
/// OAuth overload budget (`OAUTH_OVERLOAD_RETRIES` in `runtime/api.rs`).
///
/// chatgpt.com edge turbulence — 503 upstream connect errors, Cloudflare
/// 520s, raw TCP connect/read timeouts (incident: 2026-07-16 log bursts at
/// 14:19, 18:52, 20:21, 20:27, 20:55, 21:47) — clears on its own within a
/// burst; a three-attempt budget risks aborting an autonomous turn mid-burst.
pub(crate) const CODEX_PERSISTENT_RETRIES: u32 = 10;
/// Exponential-backoff exponent cap shared with the Anthropic stream retry
/// path (`stream_retry_policy`): delay = base·2^min(n−1, CAP). Without the
/// cap a 10-deep budget would sleep 1s·2⁹ = 512s on its final attempt.
const RETRY_BACKOFF_EXP_CAP: u32 = 6;
/// Effective Codex transport retry budget: the persistent floor, unless the
/// user configured something even larger. Pure — decided at the dispatch
/// seam (`try_route`), never inside `send_with_retries`, so tests can still
/// inject exact budgets.
pub(crate) fn codex_retry_budget(configured_retries: u32) -> u32 {
configured_retries.max(CODEX_PERSISTENT_RETRIES)
}
/// Backoff for retry attempt *n* (1-based): 1s·2^(n−1), capped at 64s.
fn retry_delay(attempt: u32) -> std::time::Duration {
std::time::Duration::from_millis(
1000 * 2u64.pow(attempt.saturating_sub(1).min(RETRY_BACKOFF_EXP_CAP)),
)
}
/// Retrying a response never replays tools: only a complete Ok response reaches
/// the agent tool loop. Reset uncommitted previews, not prior rounds or usage.
async fn wait_stream_retry(
tx: &mpsc::UnboundedSender<StreamEvent>,
cancel: &tokio_util::sync::CancellationToken,
retry: u32,
max_retries: u32,
suppress_stream_deltas: bool,
) -> Result<(), super::net::BoxedProviderError> {
use crate::{LlmEvent, SessionEvent};
if cancel.is_cancelled() {
return Err("request canceled".into());
}
if !suppress_stream_deltas {
let _ = tx.send(StreamEvent::Llm(LlmEvent::ResponseReset));
}
let _ = tx.send(StreamEvent::Session(SessionEvent::Retry(format!(
"Response incomplete — retrying response ({retry}/{max_retries}); incomplete output discarded, no tools replayed."
))));
tokio::select! {
biased;
_ = cancel.cancelled() => Err("request canceled".into()),
_ = tokio::time::sleep(retry_delay(retry)) => Ok(()),
}
}
pub(crate) const STREAM_INTERRUPTED: &str = "openai request failed: connection interrupted before response completed (stream retry budget exhausted)";
/// The endpoint one `send_with_retries` call talks to: its log label, the
/// URL (for the localhost-refusal fast-fail) and whether its 429 responses
/// carry Codex quota semantics worth probing. Static per call site; the
/// retry budget and trace clock are threaded separately because they mutate.
#[derive(Clone, Copy)]
struct SendTarget<'a> {
label: &'a str,
url: &'a str,
/// Codex only: read a bounded slice of a 429 body to distinguish proven
/// quota exhaustion (fail fast → account failover) from a generic 429
/// (normal transport retry). Every other endpoint drops the body unread.
quota_probe: bool,
}
/// Send a provider streaming request, retrying transient failures.
///
/// Parity fix: the Anthropic path retries transient errors with backoff
/// (`call_api_stream_inner`) but the provider routes were single-shot — one
/// transport blip against e.g. `chatgpt.com` aborted an entire autonomous
/// turn (incident: session 20260714-025948-3dab). Attempt *n* sleeps
/// 1s·2^(n−1) capped at 2^`RETRY_BACKOFF_EXP_CAP`, mirroring the Anthropic
/// budget semantics.
///
/// Retryable: transport-level send failures (timeout / connect / request)
/// and 408 / 429 / 5xx responses. Deterministic client errors (other 4xx)
/// and localhost connection refusals (Ollama/LM Studio not running — a
/// setup problem, not a blip) fail fast. Backoff sleeps are cancel-aware.
///
/// Trace (Task 10A): one record per actual HTTP send — every retried
/// failure is recorded via `attempt_failed` (status + retry class, never
/// provider text) and the per-attempt clock restarts right before each
/// re-send. Terminal failures emit their final record here; on success the
/// caller finishes the attempt after consuming the stream.
async fn send_with_retries(
target: SendTarget<'_>,
build: impl Fn() -> reqwest::RequestBuilder,
cancel: &tokio_util::sync::CancellationToken,
max_retries: u32,
retries_used: &mut u32,
trace_attempt: &mut tr::StreamAttempt,
) -> Result<reqwest::Response, Box<dyn std::error::Error + Send + Sync>> {
use super::account_routing::{
classify_codex_429, read_body_for_classification, Codex429, CodexQuotaExhausted,
QUOTA_PROBE_BODY_CAP,
};
let SendTarget {
label,
url,
quota_probe,
} = target;
loop {
if cancel.is_cancelled() {
return Err("request canceled".into());
}
let sent = tokio::select! {
biased;
_ = cancel.cancelled() => { trace_attempt.finish_canceled(None, None); return Err("request canceled".into()); }
sent = build().send() => sent,
};
match sent {
Ok(resp) if resp.status().is_success() => {
trace_attempt.mark_headers();
return Ok(resp);
}
Ok(resp) => {
// A complete HTTP response was observed even when its status
// is non-success. Preserve that timing for retried and
// terminal failure records just as the success branch does.
trace_attempt.mark_headers();
let status = resp.status();
// Provider-assigned request id from validated headers only.
let trace_rid = tr::provider_request_id_from_headers(resp.headers());
let mut retryable =
status.as_u16() == 408 || status.as_u16() == 429 || status.is_server_error();
if quota_probe && status.as_u16() == 429 {
// Codex only: a 429 that PROVES the account's usage
// window is spent must not burn the transport retry
// budget (minutes of backoff that cannot succeed). The
// body is read into a bounded buffer purely to produce
// the classification enum and is dropped unread beyond
// that — never stored, surfaced, or logged.
let headers = resp.headers().clone();
let body = tokio::select! {
biased;
_ = cancel.cancelled() => {
trace_attempt.finish_canceled(Some(429), None);
return Err("request canceled".into());
}
body = read_body_for_classification(resp, QUOTA_PROBE_BODY_CAP) => body,
};
match classify_codex_429(&headers, &body) {
Codex429::QuotaExhausted(evidence) => {
tracing::warn!(
signal = ?evidence.signal,
reset_after_secs = ?evidence.reset_after_secs,
"{label} 429 is provider-declared quota exhaustion (no transport retry)"
);
return Err(Box::new(CodexQuotaExhausted {
evidence,
request_id: trace_rid,
}));
}
Codex429::UsageNotIncluded => {
// Deterministic for this account: retrying never helps.
tracing::warn!("{label} 429: account has no Codex usage entitlement");
retryable = false;
}
Codex429::Generic => {}
}
} else if quota_probe && !retryable && status.is_client_error() {
// Codex only: a terminal 4xx is undiagnosable without its
// body. Read a bounded slice for classification; keep only
// identifier-shaped type/code/param plus Synaps-authored
// text (codex_wire). Provider prose is never kept (§5.1).
let body = tokio::select! {
biased;
_ = cancel.cancelled() => {
trace_attempt.finish_canceled(Some(status.as_u16()), None);
return Err("request canceled".into());
}
body = read_body_for_classification(resp, super::codex_wire::CODEX_ERROR_BODY_CAP) => body,
};
let detail = super::codex_wire::CodexHttpErrorDetail::from_body(&body);
let message = detail.user_message(label, status);
tracing::warn!(error = %message, "{label} request rejected (identifiers only)");
trace_attempt.finish_failed(
&detail.trace_code(status.as_u16()),
Some(status.as_u16()),
trace_rid,
);
return Err(Box::new(super::codex_wire::CodexRequestRejected {
message,
detail,
}));
} else {
// Privacy (spec §5.1): the response body is provider-controlled
// and may echo the full request (prompts, system text, tool
// schemas, credentials). Drop it unread — it must never be
// stored, surfaced, or logged at any level. `status` Display
// uses the canonical reason phrase, never server bytes. (The
// Codex 4xx arm above reads it for identifiers only.)
drop(resp);
}
let code = format!("http_{}", status.as_u16());
if !retryable || *retries_used >= max_retries {
trace_attempt.finish_failed(&code, Some(status.as_u16()), trace_rid);
return Err(format!("{label} request failed: {status}").into());
}
*retries_used += 1;
let delay = retry_delay(*retries_used);
trace_attempt.attempt_failed(
tr::retry_class_for_status(status.as_u16()),
delay,
Some(status.as_u16()),
trace_rid,
&code,
);
tracing::warn!(
"{label} API retry {retries_used}/{max_retries} after {delay:?}: {status}"
);
tokio::select! {
_ = tokio::time::sleep(delay) => {}
_ = cancel.cancelled() => {
trace_attempt.finish_canceled(None, None);
return Err("request canceled".into());
}
}
// Retry clocks reset per attempt, right before the re-send.
trace_attempt.restart_clock();
}
Err(e) => {
let localhost_refusal = e.is_connect() && url.contains("localhost");
let transient = e.is_timeout() || e.is_connect() || e.is_request();
if localhost_refusal || !transient || *retries_used >= max_retries {
tracing::warn!(
"{label} request failed (no retry): {}",
crate::core::error::error_chain_string(&e)
);
trace_attempt.finish_failed("transport_error", None, None);
return Err(e.into());
}
*retries_used += 1;
let delay = retry_delay(*retries_used);
trace_attempt.attempt_failed(
if e.is_timeout() {
crate::runtime::trace::RetryClass::Timeout
} else {
crate::runtime::trace::RetryClass::Other
},
delay,
None,
None,
"transport_error",
);
tracing::warn!(
"{label} transport retry {retries_used}/{max_retries} after {delay:?}: {}",
crate::core::error::error_chain_string(&e)
);
tokio::select! {
_ = tokio::time::sleep(delay) => {}
_ = cancel.cancelled() => {
trace_attempt.finish_canceled(None, None);
return Err("request canceled".into());
}
}
// Retry clocks reset per attempt, right before the re-send.
trace_attempt.restart_clock();
}
}
}
}
/// Run a single streaming request against an OpenAI-compatible endpoint.
///
/// Returns the final assistant response as an Anthropic-shaped content Value
/// (`{"content": [..text.., ..tool_use..]}`) so the outer agent loop can keep
/// using the same handling as the native Anthropic path.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_oai_stream_inner(
cfg: &ProviderConfig,
broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
tools_schema: &[Value],
system_prompt: &Option<String>,
messages: &[crate::SharedMessage],
tx: &mpsc::UnboundedSender<StreamEvent>,
temperature: Option<f32>,
max_tokens: Option<u32>,
thinking_budget: u32,
reasoning_level: agent_core::reasoning::ReasoningLevel,
cancel: &tokio_util::sync::CancellationToken,
trace: &crate::runtime::trace::TraceContext,
exact_wire_bytes: bool,
suppress_stream_deltas: bool,
max_retries: u32,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
let (oai_tools, name_map) = translate::tools_to_oai(tools_schema);
let oai_messages = translate::messages_to_oai(messages, system_prompt, &name_map);
let tools_opt = if oai_tools.is_empty() {
None
} else {
Some(oai_tools)
};
// Google's OpenAI-compat endpoint rejects stream_options
let stream_options = if cfg.base_url.contains("googleapis.com") {
None
} else {
Some(StreamOptions {
include_usage: true,
})
};
let mut body = serde_json::Map::new();
body.insert("model".to_string(), json!(cfg.model.clone()));
body.insert("messages".to_string(), serde_json::to_value(oai_messages)?);
body.insert("stream".to_string(), json!(true));
// ZDR: OpenAI's Zero Data Retention requires `store:false` on the wire.
// Chat Completions already defaults to false, but we assert it explicitly so the
// guarantee is provable in request logs (parity with the Codex Responses path,
// build_codex_body). Gated to api.openai.com (covers eu.api.openai.com) so we
// never send an unknown field to OpenAI-compatible providers that strict-validate
// the body (Groq/xAI/Google), mirroring the googleapis stream_options carve-out.
if cfg.base_url.contains("api.openai.com") {
body.insert("store".to_string(), json!(false));
}
if let Some(stream_options) = stream_options {
body.insert(
"stream_options".to_string(),
serde_json::to_value(stream_options)?,
);
}
if let Some(max_tokens) = max_tokens {
body.insert("max_tokens".to_string(), json!(max_tokens));
}
if let Some(temperature) = temperature {
body.insert("temperature".to_string(), json!(temperature));
}
if let Some(tools) = tools_opt {
body.insert("tools".to_string(), serde_json::to_value(tools)?);
}
super::reasoning::apply_openai_reasoning_params(
&mut body,
super::reasoning::provider_for_key(&cfg.provider),
&cfg.model,
thinking_budget,
reasoning_level,
);
let body = Value::Object(body);
tracing::debug!(provider=%cfg.provider, model=%cfg.model, "openai stream request via broker proxy");
// Serialize ONCE via the sanctioned constructor: the returned bytes are
// both digested for the trace and the very buffer stored on the request
// (`body_bytes`), which `LocalBroker` sends verbatim — the digest is
// never computed over a re-serialization, and `body`/`body_bytes`
// cannot diverge.
let (proxy_request, body_bytes) = crate::auth::ProxyRequest::post_json_exact(
cfg.provider.clone(),
"/chat/completions",
body,
true,
)?;
// ═══ TRACE (Task 10A): one record per actual attempt ═══════════════════
// A remote broker re-serializes the JSON out of process: honest kind
// `CloudProxy`, no wire-byte claim. Rule documented in `trace::emit`.
let tracer = tr::begin_openai_tracer(
trace,
&cfg.provider,
&cfg.model,
if exact_wire_bytes {
crate::runtime::trace::TransportKind::OpenAiChatCompletions
} else {
crate::runtime::trace::TransportKind::CloudProxy
},
&format!("{}/chat/completions", cfg.base_url.trim_end_matches('/')),
exact_wire_bytes.then_some(body_bytes.as_ref()),
messages,
system_prompt.as_deref(),
tools_schema,
tr::renamed_tool_losses(&name_map),
)
.await;
// One-shot explicit content capture (`/trace next content`): a no-op in
// every context without the arm. `body_bytes` is the serialized request
// body this process built pre-send (exact wire bytes on the local
// broker; the same body a remote broker re-serializes) — body only,
// headers and credentials structurally never reach this seam.
if let Some(tracer) = &tracer {
tr::capture_request_content(trace, tracer.request_id(), body_bytes.as_ref(), messages);
}
let mut attempt = tr::StreamAttempt::new(tracer);
let mut stream_retry = 0u32;
loop {
if cancel.is_cancelled() {
return Err("request canceled".into());
}
// The broker owns the API key and executes/signs the request; this path
// never resolves or attaches a credential.
let stream = crate::runtime::api::await_or_cancel(
cancel,
broker.proxy_stream(proxy_request.clone()),
)
.await
.map_err(|_| "request canceled")?
// Privacy (spec §5.1): a broker proxy error may carry an upstream
// response-body snippet; redact it to status-only before surfacing.
.map_err(|e| {
format!(
"openai request failed: {}",
super::net::redact_provider_proxy_error(&e.to_string())
)
});
let mut stream = match stream {
Ok(stream) => {
attempt.mark_headers();
stream
}
Err(msg) => {
// Status parsed from the redacted static-prefix message only;
// codes are `http_<status>` or the static `broker_error`.
let status = tr::broker_error_status(&msg);
let code =
status.map_or_else(|| "broker_error".to_string(), |s| format!("http_{s}"));
attempt.finish_failed(&code, status, None);
return Err(msg.into());
}
};
if !suppress_stream_deltas {
let _ = tx.send(StreamEvent::Llm(crate::LlmEvent::ResponseStart));
}
let mut interrupted = false;
let mut saw_done = false;
let mut decoder = StreamDecoder::new();
let mut accumulated_text = String::new();
let mut tool_use_blocks: Vec<Value> = Vec::new();
let mut buf = bytes::BytesMut::with_capacity(8 * 1024);
let mut sink: Vec<OaiEvent> = Vec::with_capacity(4);
let mut trace_usage: Option<crate::runtime::trace::UsageMeta> = None;
while let Some(chunk) = tokio::select! {
biased;
_ = cancel.cancelled() => {
attempt.finish_canceled(None, trace_usage);
return Err("request canceled".into());
}
chunk = stream.next() => chunk,
} {
let chunk = match chunk {
Ok(chunk) => {
attempt.mark_first_byte();
chunk
}
Err(e) => {
if !matches!(e, crate::auth::BrokerError::Transport(_)) {
attempt.finish_failed("broker_error", None, None);
return Err(e.into());
}
interrupted = !saw_done && decoder.finish_reason.is_none();
break;
}
};
buf.extend_from_slice(&chunk);
// Scan for newline-delimited SSE lines (SIMD-accelerated via memchr)
while let Some(nl) = memchr::memchr(b'\n', &buf) {
let line_bytes = buf.split_to(nl + 1); // O(1) — ref-counted split
let line = std::str::from_utf8(&line_bytes[..nl]).unwrap_or("");
sink.clear();
decoder.push_line(line, &mut sink);
saw_done |= sink.iter().any(|e| matches!(e, OaiEvent::Done));
capture_oai_trace_signals(&sink, &mut attempt, &mut trace_usage);
handle_events(
&sink,
tx,
&mut accumulated_text,
&mut tool_use_blocks,
&name_map,
suppress_stream_deltas,
);
}
}
// Flush any remaining buffered line + final Done
if !interrupted && !buf.is_empty() {
let line = std::str::from_utf8(&buf).unwrap_or("");
sink.clear();
decoder.push_line(line, &mut sink);
saw_done |= sink.iter().any(|e| matches!(e, OaiEvent::Done));
capture_oai_trace_signals(&sink, &mut attempt, &mut trace_usage);
handle_events(
&sink,
tx,
&mut accumulated_text,
&mut tool_use_blocks,
&name_map,
suppress_stream_deltas,
);
}
if interrupted || (!saw_done && decoder.finish_reason.is_none()) {
if stream_retry >= max_retries {
attempt.finish_failed("stream_error", None, None);
return Err(STREAM_INTERRUPTED.into());
}
stream_retry += 1;
attempt.attempt_failed(
crate::runtime::trace::RetryClass::Network,
retry_delay(stream_retry),
None,
None,
"stream_error",
);
wait_stream_retry(
tx,
cancel,
stream_retry,
max_retries,
suppress_stream_deltas,
)
.await
.inspect_err(|_| {
attempt.finish_canceled(None, None);
})?;
attempt.restart_clock();
continue;
}
sink.clear();
decoder.finish(&mut sink);
capture_oai_trace_signals(&sink, &mut attempt, &mut trace_usage);
handle_events(
&sink,
tx,
&mut accumulated_text,
&mut tool_use_blocks,
&name_map,
suppress_stream_deltas,
);
// Normalized stop reason: only when a finish_reason was actually
// observed on the wire — never inferred.
let stop_reason = decoder
.finish_reason
.as_deref()
.map(tr::stop_reason_from_finish_reason);
// Broker paths never observe the upstream HTTP status: honest `None`.
attempt.finish_success(None, None, stop_reason, trace_usage);
// Build Anthropic-shaped final response
let mut content: Vec<Value> = Vec::new();
if !accumulated_text.is_empty() {
content.push(json!({"type": "text", "text": accumulated_text}));
}
content.extend(tool_use_blocks);
return Ok(json!({
"role": "assistant",
"content": content,
}));
}
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_codex_stream_inner(
cfg: &ProviderConfig,
client: &reqwest::Client,
broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
tools_schema: &[Value],
system_prompt: &Option<String>,
messages: &[crate::SharedMessage],
tx: &mpsc::UnboundedSender<StreamEvent>,
temperature: Option<f32>,
max_tokens: Option<u32>,
reasoning_level: agent_core::reasoning::ReasoningLevel,
codex_request_role: crate::runtime::openai::catalog::CodexRequestRole,
cancel: &tokio_util::sync::CancellationToken,
max_retries: u32,
trace: &crate::runtime::trace::TraceContext,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
// Account selection crosses the broker boundary through the router seam:
// the broker vends one (credential, access token) pair, never a refresh
// token, and this path never opens auth.json.
let router = super::account_routing::BrokerCodexRouter::new(std::sync::Arc::clone(broker));
call_codex_stream_with_router(
cfg,
client,
&router,
tools_schema,
system_prompt,
messages,
tx,
temperature,
max_tokens,
reasoning_level,
codex_request_role,
cancel,
max_retries,
trace,
)
.await
}
/// What the failover gate judges: the recognized exhaustion on the current
/// attempt and the two runtime facts that can veto a switch. Built at each
/// of the two detection sites (429 probe, in-stream terminal error) so both
/// go through the identical invariant check.
#[derive(Clone, Copy)]
struct FailoverRequest<'a> {
evidence: &'a super::account_routing::CodexQuotaEvidence,
/// Any content-bearing model event reached the decoder on this attempt
/// (text, tool, reasoning — forwarded or not). Always false on the 429
/// path: no response body was streamed.
output_started: bool,
/// Account switches already spent on this logical request.
prior_failovers: u32,
}
/// Attempt a single bounded account failover after recognized quota
/// exhaustion on `failed`. Always reports the exhaustion to the router
/// (cooldown sink, best effort) — even when the runtime then refuses to
/// switch — so the broker stops advertising the seat. Returns the new pinned
/// account only when every runtime invariant holds: `Auto` selection, no
/// output produced, budget unspent, a candidate the router deems fresh and
/// eligible, and that candidate is a DIFFERENT account than the one that
/// failed. Anything else → `Ok(None)` and the caller ends the turn with the
/// static quota message. Never loops.
async fn try_codex_account_failover(
router: &dyn super::account_routing::CodexAccountRouter,
model: &str,
failed: &super::account_routing::PinnedAccount,
request: FailoverRequest<'_>,
tx: &mpsc::UnboundedSender<StreamEvent>,
cancel: &tokio_util::sync::CancellationToken,
) -> Result<Option<super::account_routing::PinnedAccount>, super::net::BoxedProviderError> {
use super::account_routing::failover_gate;
use crate::runtime::api::await_or_cancel;
let FailoverRequest {
evidence,
output_started,
prior_failovers,
} = request;
await_or_cancel(cancel, router.report_exhausted(failed, evidence))
.await
.map_err(|_| "request canceled")?;
if let Err(blocked) = failover_gate(failed.auto, output_started, prior_failovers) {
tracing::info!(
account = %failed.credential,
reason = blocked.as_str(),
"codex quota exhausted; account failover not permitted"
);
return Ok(None);
}
let candidate = await_or_cancel(
cancel,
router.failover(model, failed, evidence, prior_failovers),
)
.await
.map_err(|_| "request canceled")?;
match candidate {
Ok(Some(next)) if next.credential != failed.credential => {
tracing::warn!(
from = %failed.credential,
to = %next.credential,
"codex quota exhausted; switching account (one-time failover, no output replayed)"
);
let _ = tx.send(StreamEvent::Session(crate::SessionEvent::Notice(format!(
"Codex usage quota exhausted on account '{}' — switched to account '{}' (one-time failover; no output was replayed).",
failed.label(),
next.label()
))));
Ok(Some(next))
}
Ok(Some(_same)) => {
tracing::warn!(
account = %failed.credential,
"codex failover candidate is the exhausted account itself; refusing to retry"
);
Ok(None)
}
Ok(None) => {
tracing::info!(
account = %failed.credential,
"codex quota exhausted; no distinct account with fresh capacity"
);
Ok(None)
}
Err(e) => {
tracing::warn!(
account = %failed.credential,
error = %e,
"codex failover candidate resolution failed; not retrying"
);
Ok(None)
}
}
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_codex_stream_with_router(
cfg: &ProviderConfig,
client: &reqwest::Client,
router: &dyn super::account_routing::CodexAccountRouter,
tools_schema: &[Value],
system_prompt: &Option<String>,
messages: &[crate::SharedMessage],
tx: &mpsc::UnboundedSender<StreamEvent>,
temperature: Option<f32>,
max_tokens: Option<u32>,
reasoning_level: agent_core::reasoning::ReasoningLevel,
codex_request_role: crate::runtime::openai::catalog::CodexRequestRole,
cancel: &tokio_util::sync::CancellationToken,
max_retries: u32,
trace: &crate::runtime::trace::TraceContext,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
// Build the exact provider-qualified plan before any credential or network
// access. Logical Ultra is lowered here, never in the generic level enum.
use crate::runtime::openai::catalog::plan_codex_execution;
let qualified_model = format!("{}/{}", cfg.provider, cfg.model);
let plan =
match plan_codex_execution(&qualified_model, reasoning_level, codex_request_role, None) {
Ok(plan) => plan,
Err(error) => {
tracing::debug!(
event = "codex_mode_plan",
qualified_model = %qualified_model,
requested_level = %reasoning_level,
runtime_role = codex_request_role.as_str(),
decision = "deny",
deny_code = error.code().as_str(),
network_attempted = false,
"Codex execution plan denied"
);
return Err(Box::new(error));
}
};
if plan.automatic_delegation() && !codex_has_required_delegation_tools(tools_schema) {
tracing::debug!(
event = "codex_mode_plan",
qualified_model = %qualified_model,
requested_level = %reasoning_level,
runtime_role = codex_request_role.as_str(),
decision = "deny",
deny_code = "ultra_requires_subagent_tools",
network_attempted = false,
"Codex execution plan denied"
);
return Err(
"Ultra requires subagent_start, subagent_status, and subagent_collect tools".into(),
);
}
tracing::debug!(
event = "codex_mode_plan",
qualified_model = %plan.qualified_model,
requested_level = %plan.selected_level,
execution_mode = plan.mode.as_str(),
wire_effort = plan.wire_effort_label(),
capability_source = plan.capability_source.map_or("none", |source| source.as_str()),
multi_agent_version = plan.multi_agent_version_label(),
runtime_role = plan.request_role.as_str(),
multi_agent_mode = plan.multi_agent_mode_label(),
automatic_delegation = plan.automatic_delegation(),
decision = "allow",
network_attempted = false,
"Codex execution plan allowed"
);
// Exactly one pinned (account, token) pair per attempt. The
// `chatgpt-account-id` header is derived from THIS token's JWT claims, so
// bearer and header always name the same credential — a second lookup
// could race a policy change and pair account A's header with account
// B's token.
let mut pinned = crate::runtime::api::await_or_cancel(cancel, router.pin(&cfg.model))
.await
.map_err(|_| "request canceled")??;
let mut account_id = codex_account_id_for(&pinned)?;
let (tools, name_map) = translate::tools_to_responses(tools_schema);
// Opt-in OpenAI-hosted web search (SYNAPS_CODEX_WEB_SEARCH, read from THIS
// process's env — the daemon's when attached). Runs server-side, so hooks
// can't gate it; gated by role and model (codex_wire). The session actor
// announces the state; each call is announced by the decoder.
let mut tools = tools;
let web_search = super::codex_wire::codex_web_search_for_request(
&cfg.model,
codex_request_role,
super::codex_wire::engine_env,
);
// web_search is always the last tool. Cleared on the one resend without
// it, so that resend can't loop; the rejection is learned only if the
// resend then succeeds (`web_search_pending`).
let mut web_search_index = web_search.as_ref().map(|_| tools.len());
let mut web_search_pending = false;
tools.extend(web_search);
// Codex already receives the effective system prompt in `instructions`.
// Do not also prepend it to `input`: that doubles prompt tokens and repeats
// context/checkpoint guidance. Pass history to the translator unchanged;
// only the synthetic prepend is suppressed. Other providers keep their placement.
let oai_messages = translate::messages_to_oai(messages, &None, &name_map);
// Use the shared pure helper so production and unit tests exercise identical
// body construction — no duplication between call_codex_stream_inner and tests.
let instructions = codex_instructions(system_prompt);
let mut input_items = codex_input_messages(oai_messages);
// Key on the pre-insertion head: the mode item varies with the execution
// plan (level/role switches) and must not churn the routing key.
let prompt_cache_key = codex_prompt_cache_key(&instructions, input_items.first());
insert_codex_multi_agent_mode(&mut input_items, &plan);
let input = serde_json::Value::Array(input_items);
let mut body = build_codex_body(
&cfg.model,
&plan,
input,
instructions,
tools,
temperature,
max_tokens,
&prompt_cache_key,
);
let url = format!(
"{}/codex/responses",
cfg.base_url
.trim_end_matches('/')
.trim_end_matches("/codex")
);
tracing::debug!(url=%url, model=%cfg.model, "codex stream request");
// Serialize ONCE. These exact bytes are digested for the trace AND are
// the request body of every attempt — retries resend the identical
// buffer, so one wire digest is honest for all attempt records. The one
// exception (hosted web_search rejected) re-serializes and opens a new
// trace record for the new bytes.
let mut body_bytes = bytes::Bytes::from(serde_json::to_vec(&body)?);
// ═══ TRACE (Task 10A): one record per actual attempt ═══════════════════
let tracer = tr::begin_openai_tracer(
trace,
&cfg.provider,
&cfg.model,
crate::runtime::trace::TransportKind::OpenAiResponses,
&url,
Some(body_bytes.as_ref()),
messages,
system_prompt.as_deref(),
tools_schema,
tr::renamed_tool_losses(&name_map),
)
.await;
// One-shot explicit content capture (`/trace next content`): a no-op in
// every context without the arm. `body_bytes` is the serialized request
// body this process built pre-send (exact wire bytes on the local
// broker; the same body a remote broker re-serializes) — body only,
// headers and credentials structurally never reach this seam.
if let Some(tracer) = &tracer {
tr::capture_request_content(trace, tracer.request_id(), body_bytes.as_ref(), messages);
}
// Whether this request holds a tracer (and, on a `/trace next` context,
// the one-shot gate): only then may a web_search resend reopen it.
let traced = tracer.is_some();
let mut attempt = tr::StreamAttempt::new(tracer);
let mut stream_retry = 0u32;
// Account switches spent on this logical request. Bounded by
// `MAX_CODEX_ACCOUNT_FAILOVERS`; independent of the transport retry
// budget (a failover neither consumes nor resets `stream_retry`).
let mut failovers = 0u32;
loop {
if cancel.is_cancelled() {
return Err("request canceled".into());
}
let sent = send_with_retries(
SendTarget {
label: "codex",
url: &url,
quota_probe: true,
},
|| {
client
.post(&url)
.bearer_auth(&pinned.token)
.header("chatgpt-account-id", account_id.as_str())
.header("originator", "synaps")
.header("OpenAI-Beta", "responses=experimental")
// Cache affinity: the ChatGPT backend routes prompt-cache
// lookups by the `session-id` header (Codex CLI sends its
// conversation id as both headers and as
// `prompt_cache_key`). Without it, every request reported
// cached_tokens = 0, even with a byte-identical prefix.
.header("session-id", prompt_cache_key.as_str())
.header("thread-id", prompt_cache_key.as_str())
.header("content-type", "application/json")
.header("accept", "text/event-stream")
.body(body_bytes.clone())
},
cancel,
max_retries,
&mut stream_retry,
&mut attempt,
)
.await;
let resp = match sent {
Ok(resp) => {
if std::mem::take(&mut web_search_pending) {
super::codex_wire::confirm_web_search_rejected(&cfg.model, tx);
}
resp
}
Err(error) => match error.downcast::<super::account_routing::CodexQuotaExhausted>() {
// A 429 that proved exhaustion arrives before any response
// body was streamed: nothing was produced on this attempt, so
// the pre-output invariant holds by construction.
Ok(quota) => {
match try_codex_account_failover(
router,
&cfg.model,
&pinned,
FailoverRequest {
evidence: "a.evidence,
output_started: false,
prior_failovers: failovers,
},
tx,
cancel,
)
.await?
{
Some(next) => {
attempt.attempt_failed(
crate::runtime::trace::RetryClass::RateLimited,
std::time::Duration::ZERO,
Some(429),
quota.request_id.clone(),
"account_failover",
);
account_id = codex_account_id_for(&next)?;
pinned = next;
failovers += 1;
attempt.restart_clock();
continue;
}
None => {
let message = quota.to_string();
attempt.finish_failed("quota_exhausted", Some(429), quota.request_id);
return Err(message.into());
}
}
}
Err(other) => {
// A pre-inference 400 pointing at hosted web_search: drop
// it, re-serialize, and resend once under a fresh trace
// record (one-shot `/trace next` gates are reopened for
// it; content capture is not redone).
super::codex_wire::hosted_search_retry(
other,
&mut web_search_index,
&cfg.model,
)?;
web_search_pending = true;
super::codex_wire::strip_web_search_tool(&mut body);
body_bytes = bytes::Bytes::from(serde_json::to_vec(&body)?);
if traced {
trace.reopen_one_shot_for_resend();
}
attempt = tr::StreamAttempt::new(
tr::begin_openai_tracer(
trace,
&cfg.provider,
&cfg.model,
crate::runtime::trace::TransportKind::OpenAiResponses,
&url,
Some(body_bytes.as_ref()),
messages,
system_prompt.as_deref(),
tools_schema,
tr::renamed_tool_losses(&name_map),
)
.await,
);
continue;
}
},
};
// Direct HTTP: upstream status and provider request id are observed.
let http_status = Some(resp.status().as_u16());
let trace_rid = tr::provider_request_id_from_headers(resp.headers());
let _ = tx.send(StreamEvent::Llm(crate::LlmEvent::ResponseStart));
let mut interrupted = false;
let mut accumulated_text = String::new();
let mut parser = CodexSseDecoder::default();
let mut buf = bytes::BytesMut::with_capacity(8 * 1024);
let mut stream = resp.bytes_stream();
while let Some(chunk) = tokio::select! {
biased;
_ = cancel.cancelled() => {
attempt.finish_canceled(http_status, parser.trace_usage());
return Err("request canceled".into());
}
chunk = stream.next() => chunk,
} {
let chunk = match chunk {
Ok(chunk) => {
attempt.mark_first_byte();
chunk
}
Err(_) => {
interrupted = !parser.terminal_success && parser.terminal_failure.is_none();
break;
}
};
buf.extend_from_slice(&chunk);
while let Some(nl) = memchr::memchr(b'\n', &buf) {
let line_bytes = buf.split_to(nl + 1);
let line = std::str::from_utf8(&line_bytes[..nl]).unwrap_or("");
parser.push_line(line, tx, &mut accumulated_text);
}
if parser.saw_model_event {
attempt.mark_first_model_event();
}
}
if !interrupted && !buf.is_empty() {
let line = std::str::from_utf8(&buf).unwrap_or("");
parser.push_line(line, tx, &mut accumulated_text);
}
if interrupted || (!parser.terminal_success && parser.terminal_failure.is_none()) {
if stream_retry >= max_retries {
attempt.finish_failed("stream_error", http_status, trace_rid);
return Err(format!("Codex{RESPONSES_MISSING_TERMINAL_SUFFIX}").into());
}
stream_retry += 1;
attempt.attempt_failed(
crate::runtime::trace::RetryClass::Network,
retry_delay(stream_retry),
http_status,
trace_rid,
"stream_error",
);
wait_stream_retry(tx, cancel, stream_retry, max_retries, false)
.await
.inspect_err(|_| {
attempt.finish_canceled(None, None);
})?;
attempt.restart_clock();
continue;
}
parser.finish();
// Trailing-buffer flush and finish() can surface the first (or only)
// model event — e.g. a payload dispatched by the tail line. Mark it so
// first_model_event_ms is honest even for tail-only streams.
if parser.saw_model_event {
attempt.mark_first_model_event();
}
if let Err(failure) = parser.terminal_result() {
if failure.code == "responses_quota" {
// In-stream quota exhaustion (HTTP 200 + terminal quota
// error). Failover only if the model produced nothing on this
// attempt — text, tool, reasoning or any content-bearing
// event makes the failure terminal (no cross-account replay).
let output_started =
!accumulated_text.is_empty() || parser.model_produced_output();
let evidence = super::account_routing::CodexQuotaEvidence {
signal: super::account_routing::QuotaSignal::StreamQuotaEvent,
reset_after_secs: None,
};
match try_codex_account_failover(
router,
&cfg.model,
&pinned,
FailoverRequest {
evidence: &evidence,
output_started,
prior_failovers: failovers,
},
tx,
cancel,
)
.await?
{
Some(next) => {
attempt.attempt_failed(
crate::runtime::trace::RetryClass::RateLimited,
std::time::Duration::ZERO,
http_status,
trace_rid,
"account_failover",
);
account_id = codex_account_id_for(&next)?;
pinned = next;
failovers += 1;
attempt.restart_clock();
continue;
}
None => {
attempt.finish_failed(failure.code, http_status, trace_rid);
return Err(failure.message.into());
}
}
}
if failure.code == "responses_empty" && stream_retry < max_retries {
stream_retry += 1;
let delay = retry_delay(stream_retry);
attempt.attempt_failed(
crate::runtime::trace::RetryClass::Other,
delay,
http_status,
trace_rid,
failure.code,
);
tracing::warn!(
"codex stream retry {stream_retry}/{max_retries} after {delay:?}: completed without usable output"
);
wait_stream_retry(tx, cancel, stream_retry, max_retries, false)
.await
.inspect_err(|_| {
attempt.finish_canceled(None, None);
})?;
attempt.restart_clock();
continue;
}
attempt.finish_failed(failure.code, http_status, trace_rid);
return Err(failure.message.into());
}
attempt.finish_success(http_status, trace_rid, None, parser.trace_usage());
let mut content: Vec<Value> = Vec::new();
if !accumulated_text.is_empty() {
content.push(json!({"type": "text", "text": accumulated_text}));
}
content.extend(translate::tool_calls_to_content_blocks(
&parser.completed_tools,
&name_map,
));
return Ok(json!({
"role": "assistant",
"content": content,
}));
}
}
/// The `chatgpt-account-id` header value for a pinned account: provider-owned
/// metadata carried inside THAT token's JWT. Deriving it from the same bearer
/// the request will present is what keeps header and token paired.
fn codex_account_id_for(
pinned: &super::account_routing::PinnedAccount,
) -> Result<String, super::net::BoxedProviderError> {
crate::auth::extract_codex_account_id(&pinned.token)
.ok_or_else(|| {
"Failed to extract ChatGPT account id from Codex token — run `synaps login --provider openai-codex`".into()
})
}
/// Stable prompt-cache routing key for this conversation.
///
/// Sent as `prompt_cache_key` in the body AND as the `session-id` /
/// `thread-id` headers: the ChatGPT backend derives cache affinity from the
/// `session-id` header (upstream codex-rs sends its conversation UUID in all
/// three places). The body key alone yielded zero cache hits. Synaps
/// doesn't thread a session id down to the transport, so derive a
/// deterministic key from the stable head of the prompt: the instructions
/// plus the first input item. Identical heads hash to the same key — which
/// is exactly right, since those requests share the very prefix the cache
/// stores — and a /compact rewrite changes the key together with the prefix
/// it invalidates. Without this key, cache lookups are at the mercy of
/// load-balancer routing and long conversations re-pay their full prefix.
pub(crate) fn codex_prompt_cache_key(instructions: &str, first_item: Option<&Value>) -> String {
use sha2::{Digest, Sha256};
let mut hasher = Sha256::new();
hasher.update(instructions.as_bytes());
if let Some(item) = first_item {
hasher.update(item.to_string().as_bytes());
}
format!("synaps-{:x}", hasher.finalize())
}
/// Pure body construction for Codex Responses-API requests.
///
/// Separated from the async function so it can be unit-tested without
/// any credential access or network I/O.
///
/// # Arguments
/// - `model`: the Codex model id (already validated by the caller)
/// - `plan`: reasoning plan (must already be validated against model capability)
/// - `input`: pre-built input items (from `codex_input_messages`)
/// - `instructions`: pre-built instructions string (from `codex_instructions`)
/// - `tools`: pre-built tool array
/// - `temperature`: optional temperature override
/// - `max_tokens`: optional max_output_tokens override
/// - `prompt_cache_key`: pre-computed routing key; computed on the
/// pre-insertion input head by the caller, so it cannot be derived here
#[allow(clippy::too_many_arguments)] // pure builder mirroring the wire body; grouping would obscure the 1:1 field mapping
pub(crate) fn build_codex_body(
model: &str,
plan: &crate::runtime::openai::catalog::CodexExecutionPlan,
input: serde_json::Value,
instructions: String,
tools: Vec<serde_json::Value>,
temperature: Option<f32>,
max_tokens: Option<u32>,
prompt_cache_key: &str,
) -> serde_json::Value {
let mut body = json!({
"model": model,
"store": false,
"stream": true,
"instructions": instructions,
"input": input,
"tool_choice": "auto",
"parallel_tool_calls": true,
"include": ["reasoning.encrypted_content"],
"text": { "verbosity": "medium" },
"prompt_cache_key": prompt_cache_key,
});
if let Some(effort) = plan.wire_effort {
body["reasoning"] = json!({ "effort": effort.as_str() });
}
if !tools.is_empty() {
body["tools"] = serde_json::Value::Array(tools);
}
if let Some(temp) = temperature {
body["temperature"] = json!(temp);
}
if let Some(max) = max_tokens {
body["max_output_tokens"] = json!(max);
}
body
}
const CODEX_MULTI_AGENT_MODE_OPEN: &str = "<multi_agent_mode>";
const CODEX_MULTI_AGENT_MODE_CLOSE: &str = "</multi_agent_mode>";
const CODEX_EXPLICIT_REQUEST_ONLY_TEXT: &str = "Do not spawn sub-agents unless the user or applicable AGENTS.md/skill instructions explicitly ask for sub-agents, delegation, or parallel agent work.";
const CODEX_PROACTIVE_MULTI_AGENT_TEXT: &str = "Proactive multi-agent delegation is active. Any earlier instruction requiring an explicit user request before spawning sub-agents no longer applies. Use sub-agents when parallel work would materially improve speed or quality. This mode remains active until a later multi-agent mode developer message changes it.";
/// Environment switch for the OpenAI-hosted `web_search` tool on Codex
/// Responses requests.
pub(crate) const CODEX_WEB_SEARCH_ENV: &str = "SYNAPS_CODEX_WEB_SEARCH";
/// Map the `SYNAPS_CODEX_WEB_SEARCH` value to the hosted tool spec:
/// - `live` / `on` / `true` / `1` → `{"type":"web_search"}` (live web access)
/// - `cached` → `{"type":"web_search","external_web_access":false}`
/// - anything else (unset, `off`, typos) → no hosted tool.
///
/// The hosted tool's `web_search_call` output items are not function calls:
/// they never become tool uses; each completed one is announced as a notice
/// with its query (`codex_wire::announce_web_search_call`).
pub(crate) fn codex_web_search_tool(value: Option<&str>) -> Option<Value> {
match value.map(|v| v.trim().to_ascii_lowercase()).as_deref() {
Some("live" | "on" | "true" | "1" | "yes") => Some(json!({ "type": "web_search" })),
Some("cached") => Some(json!({ "type": "web_search", "external_web_access": false })),
_ => None,
}
}
fn codex_multi_agent_mode_item(
plan: &crate::runtime::openai::catalog::CodexExecutionPlan,
) -> Option<Value> {
use crate::runtime::openai::catalog::CodexMultiAgentMode;
let body = match plan.multi_agent_mode? {
CodexMultiAgentMode::ExplicitRequestOnly => CODEX_EXPLICIT_REQUEST_ONLY_TEXT,
CodexMultiAgentMode::Proactive => CODEX_PROACTIVE_MULTI_AGENT_TEXT,
};
Some(json!({
"type": "message",
"role": "developer",
"content": [{
"type": "input_text",
"text": format!("{CODEX_MULTI_AGENT_MODE_OPEN}{body}{CODEX_MULTI_AGENT_MODE_CLOSE}"),
}],
}))
}
fn is_codex_multi_agent_mode_item(item: &Value) -> bool {
item.pointer("/content/0/text")
.and_then(Value::as_str)
.is_some_and(|text| {
text.trim_start().starts_with(CODEX_MULTI_AGENT_MODE_OPEN)
&& text.trim_end().ends_with(CODEX_MULTI_AGENT_MODE_CLOSE)
})
}
fn insert_codex_multi_agent_mode(
input: &mut Vec<Value>,
plan: &crate::runtime::openai::catalog::CodexExecutionPlan,
) {
input.retain(|item| !is_codex_multi_agent_mode_item(item));
let Some(item) = codex_multi_agent_mode_item(plan) else {
return;
};
// Prefix-cache stability: the mode item goes at a FIXED position (head of
// input), never relative to the latest user message. The previous
// placement (before the last user item) moved every user turn, so the
// prompt prefix diverged at the prior turn's user message and the entire
// preceding agentic turn — often the bulk of the context — was re-billed
// uncached (incident: 2026-07-16 codex cache investigation). At index 0
// the item is byte-stable across turns; it only changes when the
// execution plan's mode changes, which legitimately invalidates the
// prefix once.
input.insert(0, item);
}
fn codex_has_required_delegation_tools(tools_schema: &[Value]) -> bool {
["subagent_start", "subagent_status", "subagent_collect"]
.into_iter()
.all(|required| {
tools_schema.iter().any(|tool| {
tool.get("name").and_then(Value::as_str) == Some(required)
|| tool.pointer("/function/name").and_then(Value::as_str) == Some(required)
})
})
}
const CODEX_AUTONOMOUS_LOOP_POLICY: &str = "\n\n[Synaps autonomous harness policy]\nThis harness is non-interactive after the user has provided the task/spec. Do not stop at phase boundaries, milestones, checkpoints, or after presenting a plan unless the full requested job is complete. Do not ask the user whether to continue. When a phase/checkpoint is reached, run any relevant verification and continue autonomously until the full requested job is complete, blocked by an unrecoverable error, or explicit user instructions require stopping.\n[End Synaps autonomous harness policy]";
fn codex_instructions(system_prompt: &Option<String>) -> String {
let mut instructions = system_prompt.clone().unwrap_or_default();
if instructions.contains("[Synaps autonomous harness policy]") {
return instructions;
}
instructions.push_str(CODEX_AUTONOMOUS_LOOP_POLICY);
instructions
}
fn codex_input_messages(messages: Vec<ChatMessage>) -> Vec<Value> {
responses_input_messages(messages, false)
}
/// Meta Muse input: like [`codex_input_messages`], but replayed assistant
/// text carries the Codex-style message `phase` Muse itself emits.
///
/// Muse labels a preamble before tool calls `phase:"commentary"` and a
/// turn-ending answer `phase:"final_answer"` (or no phase). Replaying every
/// past assistant message as a phase-less string taught the model that its
/// "final answers" go on to call tools, and it then ended live turns on a
/// bare preamble (A/B replay of the failing history: 2/16 runs reproduced
/// with phase-less replay, 0/16 with phase restored).
///
/// Session storage keeps no phase, so it is inferred from structure, which
/// matches what Muse emits: `messages_to_oai` folds an assistant turn's text
/// into ONE message that also carries the turn's calls, so text on a message
/// WITH tool calls is commentary, text on a message without them is the
/// final answer. Tool-only messages and the whitespace separator
/// `messages_to_oai` inserts between tool results and plain user text are
/// harness wire artefacts, not model text: they keep the legacy shape.
fn muse_input_messages(messages: Vec<ChatMessage>) -> Vec<Value> {
responses_input_messages(messages, true)
}
/// Wire phase Muse uses for a replayed assistant message's text.
fn muse_assistant_phase(has_tool_calls: bool) -> &'static str {
if has_tool_calls {
"commentary"
} else {
"final_answer"
}
}
fn responses_input_messages(messages: Vec<ChatMessage>, assistant_phase: bool) -> Vec<Value> {
let mut out = Vec::new();
for msg in messages {
if msg.role == "tool" {
// Skip tool results with no call_id — sending an empty call_id
// to the Codex API would cause a 400 with a confusing error.
if let Some(call_id) = msg.tool_call_id {
out.push(json!({
"type": "function_call_output",
"call_id": call_id,
"output": msg.content.unwrap_or_default(),
}));
}
continue;
}
// A mixed assistant message carries BOTH text and calls. Emit its
// text first rather than continuing after the calls and losing it.
// Text-only messages retain the existing string wire representation.
let phased_text =
if assistant_phase && msg.role == "assistant" && msg.content_parts.is_none() {
msg.content
.as_deref()
.filter(|text| !text.trim().is_empty())
} else {
None
};
if let Some(text) = phased_text {
out.push(json!({
"type": "message",
"role": "assistant",
"phase": muse_assistant_phase(msg.tool_calls.as_ref().is_some_and(|c| !c.is_empty())),
"content": [{"type": "output_text", "text": text}],
}));
} else if msg.content_parts.is_some() || msg.content.is_some() || msg.tool_calls.is_none() {
let content = if let Some(parts) = msg.content_parts {
Value::Array(parts.into_iter().map(|part| match part {
ChatContentPart::Text { text } => json!({
"type": if msg.role == "assistant" { "output_text" } else { "input_text" },
"text": text,
}),
ChatContentPart::ImageUrl { image_url } => {
let mut part = json!({"type": "input_image", "image_url": image_url.url});
if let Some(detail) = image_url.detail {
part["detail"] = json!(detail);
}
part
}
ChatContentPart::File { file } => json!({
"type": "input_file",
"filename": file.filename,
"file_data": file.file_data,
}),
}).collect())
} else {
Value::String(msg.content.unwrap_or_default())
};
out.push(json!({"role": msg.role, "content": content}));
}
if let Some(tool_calls) = msg.tool_calls {
for call in tool_calls {
// The Responses API rejects `id` values that are not the
// original `fc_…` output-item id. We only carry the
// `call_…` correlation id today (see types::ToolCall),
// so emit `id` *only* when the value actually starts
// with `fc`. `call_id` is sufficient on its own to
// correlate the eventual `function_call_output`.
let mut item = json!({
"type": "function_call",
"call_id": call.id,
"name": call.function.name,
"arguments": call.function.arguments,
});
if call.id.starts_with("fc") {
item["id"] = Value::from(call.id.clone());
}
out.push(item);
}
}
}
out
}
#[derive(Clone, Debug, Eq, PartialEq)]
struct ResponsesStreamFailure {
code: &'static str,
/// User-facing text built from OUR static templates plus the provider
/// display label — never provider free-text.
message: String,
}
impl ResponsesStreamFailure {
/// Terminal failures that are transient by nature and safe to re-send
/// with identical request bytes: provider capacity exhaustion, an empty
/// completion, or a stream that ended without any terminal event.
fn is_retryable(&self) -> bool {
matches!(
self.code,
"responses_capacity" | "responses_empty" | "responses_missing_terminal"
)
}
}
/// Sanitized provider error codes that mean "try again later" on the
/// Responses wire. Classification-only: the code is matched against this
/// fixed set, never echoed beyond the local sanitized log line.
const CAPACITY_ERROR_CODES: &[&str] = &[
"rate_limit_exceeded",
"server_overloaded",
"overloaded",
"capacity",
"server_error",
];
/// Fixed substrings that mark a provider free-text error message as a
/// transient capacity condition (xAI: "The model is currently at capacity
/// due to high demand. Please try again in a few minutes…"). The message is
/// consulted ONLY to produce a boolean — it is never logged, stored, or
/// surfaced (privacy invariant: provider bodies can echo request content).
const CAPACITY_MESSAGE_PATTERNS: &[&str] = &[
"at capacity",
"high demand",
"try again in a few minutes",
"overloaded",
];
/// Classification-only capacity check. `error_code` is already sanitized;
/// `message` is raw provider text used solely as a boolean signal.
fn is_capacity_failure(error_code: &str, message: Option<&str>) -> bool {
if CAPACITY_ERROR_CODES.contains(&error_code) {
return true;
}
message.is_some_and(|m| {
let lower = m.to_ascii_lowercase();
CAPACITY_MESSAGE_PATTERNS.iter().any(|p| lower.contains(p))
})
}
struct CodexSseDecoder {
/// Provider display label used in user-facing terminal messages
/// ("Codex", "xAI"). Static, ours — never provider-supplied.
provider_label: &'static str,
buffer: String,
active_tools: Vec<CodexToolAccumulator>,
completed_tools: Vec<ToolCall>,
/// Trace (Task 10A): set once the decoder parses any model event —
/// feeds the first-model-event timing bucket.
saw_model_event: bool,
/// A terminal Responses event was observed. Failure/incomplete events
/// take precedence over completion and are surfaced after the stream is
/// drained; provider-controlled text is never retained.
terminal_success: bool,
terminal_failure: Option<ResponsesStreamFailure>,
emitted_output: bool,
/// The model produced ANY content on this attempt — including events the
/// decoder does not forward to the UI (reasoning / reasoning-summary
/// deltas, output items, content parts). Replaying such a request on a
/// second account would double-spend, so account failover keys on this,
/// not merely on what reached the UI. Metadata-only events
/// (`response.created`, `response.in_progress`) never set it.
saw_content_output: bool,
/// Trace (Task 10A): provider-reported usage from `response.completed`
/// (input including cached, output, cached slice). `None` until observed.
observed_usage: Option<(u64, u64, u64)>,
/// `phase` of the most recent `message` output item (`None` when that
/// item had no phase, or no message item was seen). Muse marks an
/// unfinished preamble `commentary`; see [`muse_dangling_preamble`].
last_message_phase: Option<String>,
/// `output_index` of the most recent `message` output item.
last_message_index: Option<usize>,
/// Text streamed via `output_text.delta`, per `output_index`. Bounded by
/// what the UI already received (it duplicates `text_acc`, split by item).
message_text: std::collections::BTreeMap<usize, String>,
/// Muse only: when a completed message item carries text that extends
/// what was streamed for it, emit the missing suffix as a text delta.
/// Off for every other provider so their decoding stays unchanged.
backfill_message_text: bool,
}
impl Default for CodexSseDecoder {
fn default() -> Self {
Self::for_provider("Codex")
}
}
#[derive(Default)]
struct CodexToolAccumulator {
id: String,
name: String,
arguments: String,
started: bool,
}
/// Parse a function-call arguments string into a JSON `Value`, mirroring
/// `runtime::api::parse_tool_input` so the chat UI's `LlmEvent::ToolUse`
/// handling sees the same shape regardless of provider.
///
/// Empty / whitespace input becomes `{}`. Invalid JSON becomes
/// `{"__parse_error": "..."}` — the agent loop already understands that
/// shape and converts it into an `is_error: true` tool_result.
fn parse_tool_arguments(raw: &str) -> Value {
if raw.trim().is_empty() {
return json!({});
}
match serde_json::from_str(raw) {
Ok(v) => v,
Err(e) => json!({ "__parse_error": format!("invalid tool input JSON: {}", e) }),
}
}
/// Reduce a provider-supplied error identifier (`error.type`, `error.code`,
/// `incomplete_details.reason`) to a token safe for local logs. Lowercase
/// snake-case identifiers (`server_error`, `rate_limit_exceeded`,
/// `context_length_exceeded`, `max_output_tokens`) pass through verbatim —
/// that shape cannot carry prose or credential material. Anything with
/// uppercase, whitespace, or symbols outside the identifier charset — or
/// longer than 64 bytes — is replaced so free-text (which can echo request
/// content) and secret-shaped tokens can never leak into the log stream.
pub(crate) fn sanitize_error_identifier(raw: Option<&str>) -> &str {
match raw {
None => "absent",
Some(s) => {
let identifier_shaped = !s.is_empty()
&& s.len() <= 64
&& s.bytes()
.all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-' || b == b'.');
if identifier_shaped
&& s.bytes()
.all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
{
// Lowercase snake_case enum value: safe to record verbatim.
s
} else if identifier_shaped {
// Mixed-case / dashed identifiers can be secret-shaped
// (API keys, tokens): report shape only.
"unlisted_identifier"
} else {
"unsafe_or_freeform"
}
}
}
}
/// Static message suffixes for Responses terminal failures. The provider
/// display label is prefixed at construction; `net.rs` classifies on these
/// label-independent suffixes so every label maps to `ApiStatus`.
pub(crate) const RESPONSES_AUTH_SUFFIX: &str =
" authentication rejected in stream. Sign in again or switch to an available account.";
pub(crate) const RESPONSES_QUOTA_SUFFIX: &str =
" usage quota exhausted in stream. Add credit, wait for reset, or switch models.";
pub(crate) const RESPONSES_FAILED_SUFFIX: &str =
" response failed in stream. Provider error details withheld because they can echo request content.";
pub(crate) const RESPONSES_CONTEXT_SUFFIX: &str = " rejected the request: the conversation exceeds this model's context window. Run /compact or start a fresh session to continue.";
pub(crate) const RESPONSES_CAPACITY_SUFFIX: &str = " reports the model is at capacity (retries exhausted). Try again in a few minutes or switch models with /model.";
pub(crate) const RESPONSES_INCOMPLETE_SUFFIX: &str =
" response was incomplete. Retry the request or reduce the requested output/context size.";
pub(crate) const RESPONSES_EMPTY_SUFFIX: &str =
" completed without text or tool output. Retry the request.";
pub(crate) const RESPONSES_MISSING_TERMINAL_SUFFIX: &str =
" response stream ended without a terminal event. Retry the request.";
/// Map a Responses terminal failure to a user-facing message. Only reacts to
/// the already-sanitized enum identifiers (`error.type` / `error.code`) and
/// the boolean capacity classification — never to provider free-text — so no
/// request content can leak. Known, user-actionable conditions get a specific
/// remedy; everything else keeps the neutral, details-withheld default.
fn user_message_for_terminal_failure(
label: &str,
error_kind: &str,
error_code: &str,
capacity: bool,
) -> String {
// Context-window overflow is deterministic and user-fixable: the turn
// cannot succeed until the conversation is smaller.
if error_code == "context_length_exceeded" || error_kind == "context_length_exceeded" {
return format!("{label}{RESPONSES_CONTEXT_SUFFIX}");
}
// Both explicit identifiers must agree on a retry class. Unknown or policy
// classes (even with a capacity-looking message) are never retry signals.
let classify = |value: &str| match value {
"absent" | "error" => None,
"invalid_api_key" | "authentication_error" | "invalid_authentication_error" => Some("auth"),
"insufficient_quota" | "quota_exceeded" | "billing_error" | "usage_limit_reached" => {
Some("quota")
}
"rate_limit_exceeded"
| "rate_limit_error"
| "server_overloaded"
| "overloaded"
| "capacity"
| "server_error"
| "api_error"
| "overloaded_error" => Some("capacity"),
_ => Some("blocked"),
};
let (kind, code) = (classify(error_kind), classify(error_code));
if kind == Some("blocked")
|| code == Some("blocked")
|| (kind.is_some() && code.is_some() && kind != code)
{
return format!("{label}{RESPONSES_FAILED_SUFFIX}");
}
match code.or(kind) {
Some("auth") => return format!("{label}{RESPONSES_AUTH_SUFFIX}"),
Some("quota") => return format!("{label}{RESPONSES_QUOTA_SUFFIX}"),
Some("capacity") => return format!("{label}{RESPONSES_CAPACITY_SUFFIX}"),
// Preserve legacy flat xAI capacity handling only with no explicit
// failure identifier to contradict it. No new free-text classifiers.
None if capacity => return format!("{label}{RESPONSES_CAPACITY_SUFFIX}"),
_ => {}
}
format!("{label}{RESPONSES_FAILED_SUFFIX}")
}
impl CodexSseDecoder {
/// Decoder whose user-facing messages name `label` as the provider.
fn for_provider(label: &'static str) -> Self {
Self {
provider_label: label,
buffer: String::new(),
active_tools: Vec::new(),
completed_tools: Vec::new(),
saw_model_event: false,
terminal_success: false,
terminal_failure: None,
emitted_output: false,
saw_content_output: false,
observed_usage: None,
last_message_phase: None,
last_message_index: None,
message_text: std::collections::BTreeMap::new(),
backfill_message_text: false,
}
}
/// Record a `message` output item's phase/index (added or done event).
fn observe_message_item(&mut self, idx: usize, item: &Value) {
if item.get("type").and_then(Value::as_str) != Some("message") {
return;
}
self.last_message_index = Some(idx);
self.last_message_phase = item
.get("phase")
.and_then(Value::as_str)
.map(str::to_string);
}
/// Text of the most recent message item as streamed, if any was.
fn last_message_text(&self) -> Option<&str> {
self.last_message_index
.and_then(|idx| self.message_text.get(&idx))
.map(String::as_str)
}
/// Muse-only (C): a completed message item whose text extends what was
/// streamed for it (and starts with it) gets the missing suffix emitted
/// as a normal text delta. Never rewrites or duplicates streamed text.
fn backfill_message_suffix(
&mut self,
idx: usize,
item: &Value,
tx: &mpsc::UnboundedSender<StreamEvent>,
text_acc: &mut String,
) {
if !self.backfill_message_text
|| item.get("type").and_then(Value::as_str) != Some("message")
{
return;
}
let Some(parts) = item.get("content").and_then(Value::as_array) else {
return;
};
let full: String = parts
.iter()
.filter(|part| part.get("type").and_then(Value::as_str) == Some("output_text"))
.filter_map(|part| part.get("text").and_then(Value::as_str))
.collect();
let streamed = self.message_text.entry(idx).or_default();
if full.len() <= streamed.len() || !full.starts_with(streamed.as_str()) {
return;
}
let suffix = full[streamed.len()..].to_string();
tracing::debug!(
provider = self.provider_label,
suffix_bytes = suffix.len(),
"backfilling message text missing from the delta stream"
);
streamed.push_str(&suffix);
self.emitted_output = true;
text_acc.push_str(&suffix);
let _ = tx.send(StreamEvent::Llm(crate::runtime::types::LlmEvent::Text(
suffix,
)));
}
/// True once anything was streamed to the UI for this attempt (text
/// deltas, tool starts, or completed tools). A retry after this point
/// would duplicate output, so callers must treat the failure as terminal.
fn emitted_any_output(&self) -> bool {
self.emitted_output
|| !self.completed_tools.is_empty()
|| self.active_tools.iter().any(|tool| tool.started)
}
/// True once the model produced any content on this attempt, forwarded
/// or not. The account-failover gate uses this (stricter than
/// `emitted_any_output`): a request that already cost output on seat A
/// is never replayed on seat B.
fn model_produced_output(&self) -> bool {
self.saw_content_output || self.emitted_any_output()
}
fn push_line(
&mut self,
line: &str,
tx: &mpsc::UnboundedSender<StreamEvent>,
text_acc: &mut String,
) {
let line = line.trim_end_matches('\r');
if line.is_empty() {
if !self.buffer.is_empty() {
let payload = std::mem::take(&mut self.buffer);
self.push_payload(&payload, tx, text_acc);
}
return;
}
let Some(data) = line.strip_prefix("data:").map(str::trim_start) else {
return;
};
if data == "[DONE]" {
self.terminal_success = true;
self.finish();
return;
}
self.buffer.push_str(data);
}
fn push_payload(
&mut self,
payload: &str,
tx: &mpsc::UnboundedSender<StreamEvent>,
text_acc: &mut String,
) {
let Ok(event) = serde_json::from_str::<Value>(payload) else {
tracing::debug!(
payload_bytes = payload.len(),
"discarding malformed Responses SSE payload"
);
return;
};
let event_type = event
.get("type")
.and_then(Value::as_str)
.unwrap_or_default();
if matches!(
event_type,
"response.output_text.delta"
| "response.output_item.added"
| "response.function_call_arguments.delta"
| "response.output_item.done"
| "response.completed"
| "response.done"
| "response.failed"
| "response.incomplete"
| "error"
) {
self.saw_model_event = true;
}
// Content-bearing event families. Anything the model emitted as
// output — text, tool arguments, reasoning, content parts, output
// items — counts, even when this decoder ignores the event below.
if matches!(
event_type,
"response.output_text.delta"
| "response.output_text.done"
| "response.output_item.added"
| "response.output_item.done"
| "response.function_call_arguments.delta"
| "response.function_call_arguments.done"
) || event_type.starts_with("response.reasoning")
|| event_type.starts_with("response.content_part")
|| event_type.starts_with("response.refusal")
{
self.saw_content_output = true;
}
match event_type {
"response.output_text.delta" => {
if let Some(delta) = event.get("delta").and_then(Value::as_str) {
if !delta.is_empty() {
self.emitted_output = true;
}
let idx = event
.get("output_index")
.and_then(Value::as_u64)
.unwrap_or(0) as usize;
self.message_text.entry(idx).or_default().push_str(delta);
text_acc.push_str(delta);
let _ = tx.send(StreamEvent::Llm(crate::runtime::types::LlmEvent::Text(
delta.to_string(),
)));
}
}
"response.output_item.added" => {
if let Some(item) = event.get("item") {
let idx = event
.get("output_index")
.and_then(Value::as_u64)
.unwrap_or(0) as usize;
self.observe_message_item(idx, item);
self.add_tool_from_item(idx, item, tx);
}
}
"response.function_call_arguments.delta" => {
let idx = event
.get("output_index")
.and_then(Value::as_u64)
.unwrap_or(0) as usize;
let delta = event
.get("delta")
.and_then(Value::as_str)
.unwrap_or_default();
if !delta.is_empty() {
let tool = self.ensure_tool(idx);
tool.arguments.push_str(delta);
let tool_id = tool.id.clone();
let _ = tx.send(StreamEvent::Llm(
crate::runtime::types::LlmEvent::ToolUseDelta {
tool_id,
delta: delta.to_string(),
},
));
}
}
"response.output_item.done" => {
if let Some(item) = event.get("item") {
let idx = event
.get("output_index")
.and_then(Value::as_u64)
.unwrap_or(0) as usize;
self.observe_message_item(idx, item);
self.backfill_message_suffix(idx, item, tx, text_acc);
super::codex_wire::announce_web_search_call(self.provider_label, item, tx);
self.complete_tool_from_item(idx, item, tx);
}
}
"response.completed" | "response.done" => {
self.push_usage(&event, tx);
self.terminal_success = true;
self.finish();
}
"response.failed" | "error" => {
self.push_usage(&event, tx);
// Local-only diagnostics: log the provider's enum-like error
// identifiers (`type` / `code`) after sanitizing them down to a
// short identifier charset. The free-text `message` field is
// never logged or surfaced — it can echo request content.
//
// Two envelope shapes are accepted: the nested Responses form
// (`response.error.{type,code,message}` / `error.{…}`) and
// xAI's flat form (`{"type":"error","code":…,"message":…}`)
// where the event's own `type` is just "error", so only the
// top-level `code` is a meaningful identifier there.
let error_obj = event
.get("response")
.and_then(|r| r.get("error"))
.or_else(|| event.get("error"))
.filter(|e| e.is_object());
let error_kind = sanitize_error_identifier(
error_obj
.and_then(|e| e.get("type"))
.and_then(Value::as_str),
);
let error_code = sanitize_error_identifier(
error_obj
.and_then(|e| e.get("code"))
.or_else(|| event.get("code"))
.and_then(Value::as_str),
);
// Provider free-text: classification-only boolean, never
// retained past this expression.
let capacity = is_capacity_failure(
error_code,
error_obj
.and_then(|e| e.get("message"))
.or_else(|| event.get("message"))
.and_then(Value::as_str),
);
tracing::warn!(
provider = self.provider_label,
event_type,
error_kind,
error_code,
capacity,
"responses stream terminal failure (provider message withheld)"
);
let message = user_message_for_terminal_failure(
self.provider_label,
error_kind,
error_code,
capacity,
);
self.terminal_failure = Some(ResponsesStreamFailure {
// Exact account/context failures outrank a free-text capacity
// hint: do not spend local retries before allowing failover.
code: if message
== format!("{}{RESPONSES_CAPACITY_SUFFIX}", self.provider_label)
{
"responses_capacity"
} else if super::account_routing::stream_failure_is_quota(
error_kind, error_code,
) {
// Vetted quota identifiers only: the account's
// allowance is spent. Not retried here; the Codex
// path may fail over once if nothing was produced.
"responses_quota"
} else {
"responses_failed"
},
message,
});
self.finish();
}
"response.incomplete" => {
self.push_usage(&event, tx);
let reason = sanitize_error_identifier(
event
.get("response")
.and_then(|r| r.get("incomplete_details"))
.and_then(|d| d.get("reason"))
.and_then(Value::as_str),
);
tracing::warn!(
provider = self.provider_label,
reason,
"responses stream terminal incomplete"
);
self.terminal_failure = Some(ResponsesStreamFailure {
code: "responses_incomplete",
message: format!("{}{RESPONSES_INCOMPLETE_SUFFIX}", self.provider_label),
});
self.finish();
}
_ => {
if !event_type.is_empty() {
tracing::debug!(event_type, "ignoring unknown Responses SSE event");
}
}
}
}
fn terminal_result(&self) -> Result<(), ResponsesStreamFailure> {
if let Some(failure) = &self.terminal_failure {
return Err(failure.clone());
}
if self.terminal_success {
if self.emitted_output || !self.completed_tools.is_empty() {
Ok(())
} else {
Err(ResponsesStreamFailure {
code: "responses_empty",
message: format!("{}{RESPONSES_EMPTY_SUFFIX}", self.provider_label),
})
}
} else {
Err(ResponsesStreamFailure {
code: "responses_missing_terminal",
message: format!("{}{RESPONSES_MISSING_TERMINAL_SUFFIX}", self.provider_label),
})
}
}
fn ensure_tool(&mut self, idx: usize) -> &mut CodexToolAccumulator {
// `idx` is the provider-supplied `output_index`; a hostile or buggy
// provider sending a huge value would otherwise grow this Vec unboundedly
// (OOM/DoS). No legitimate response has thousands of concurrent tool
// calls, so clamp absurd indices to the last slot instead of allocating.
const MAX_TOOLS: usize = 4096;
let idx = idx.min(MAX_TOOLS - 1);
while self.active_tools.len() <= idx {
self.active_tools.push(CodexToolAccumulator::default());
}
&mut self.active_tools[idx]
}
fn add_tool_from_item(
&mut self,
idx: usize,
item: &Value,
tx: &mpsc::UnboundedSender<StreamEvent>,
) {
if item.get("type").and_then(Value::as_str) != Some("function_call") {
return;
}
let tool = self.ensure_tool(idx);
if let Some(id) = item
.get("call_id")
.or_else(|| item.get("id"))
.and_then(Value::as_str)
{
tool.id = id.to_string();
}
if let Some(name) = item.get("name").and_then(Value::as_str) {
tool.name = name.to_string();
}
if !tool.started && !tool.name.is_empty() {
tool.started = true;
let _ = tx.send(StreamEvent::Llm(
crate::runtime::types::LlmEvent::ToolUseStart {
tool_name: tool.name.clone(),
tool_id: tool.id.clone(),
},
));
}
}
fn complete_tool_from_item(
&mut self,
idx: usize,
item: &Value,
tx: &mpsc::UnboundedSender<StreamEvent>,
) {
if item.get("type").and_then(Value::as_str) != Some("function_call") {
return;
}
let tool = self.ensure_tool(idx);
if let Some(id) = item
.get("call_id")
.or_else(|| item.get("id"))
.and_then(Value::as_str)
{
tool.id = id.to_string();
}
if let Some(name) = item.get("name").and_then(Value::as_str) {
tool.name = name.to_string();
}
if let Some(arguments) = item.get("arguments").and_then(Value::as_str) {
tool.arguments = arguments.to_string();
}
if !tool.started && !tool.name.is_empty() {
tool.started = true;
let _ = tx.send(StreamEvent::Llm(
crate::runtime::types::LlmEvent::ToolUseStart {
tool_name: tool.name.clone(),
tool_id: tool.id.clone(),
},
));
}
let completed = if !tool.id.is_empty() && !tool.name.is_empty() {
Some(ToolCall {
id: tool.id.clone(),
kind: "function".to_string(),
function: super::types::FunctionCall {
name: tool.name.clone(),
arguments: tool.arguments.clone(),
},
})
} else {
None
};
if let Some(call) = completed {
if self.completed_tools.iter().any(|done| done.id == call.id) {
return;
}
self.emitted_output = true;
// Emit the finalized `ToolUse` event so the chat UI can collapse
// the streaming `ToolUseStart` (animated) into a stable
// `ToolUse` block. Without this the bash-trace animation
// persists forever and parallel tool blocks render as "still
// running" even after they've completed. Mirrors the
// Anthropic path in `runtime/api.rs` which emits the same
// event on tool-use content_block_stop.
let input = parse_tool_arguments(&call.function.arguments);
let _ = tx.send(StreamEvent::Llm(crate::runtime::types::LlmEvent::ToolUse {
tool_name: call.function.name.clone(),
tool_id: call.id.clone(),
input,
}));
self.completed_tools.push(ToolCall {
id: call.id,
kind: call.kind,
function: call.function,
});
}
}
fn push_usage(&mut self, event: &Value, tx: &mpsc::UnboundedSender<StreamEvent>) {
let usage = event
.get("response")
.and_then(|r| r.get("usage"))
.or_else(|| event.get("usage"));
let input = usage
.and_then(|u| u.get("input_tokens"))
.and_then(Value::as_u64)
.unwrap_or(0);
let output = usage
.and_then(|u| u.get("output_tokens"))
.and_then(Value::as_u64)
.unwrap_or(0);
// Responses-API usage nests cache hits under input_tokens_details;
// input_tokens INCLUDES them. Downstream accounting uses Anthropic
// semantics (context = input + cache_read + cache_creation, hit% =
// cache_read/total), so report the cached slice separately and
// subtract it from input — otherwise cache hits are invisible
// (reported 0) and the context total double-counts.
let cached = usage
.and_then(|u| u.pointer("/input_tokens_details/cached_tokens"))
.and_then(Value::as_u64)
.unwrap_or(0)
.min(input);
if input > 0 || output > 0 {
// Trace capture (metadata only): the provider-reported totals as
// sent — input including the cached slice.
if self.observed_usage.is_none() {
self.observed_usage = Some((input, output, cached));
}
let _ = tx.send(StreamEvent::Session(
crate::runtime::types::SessionEvent::Usage {
input_tokens: input - cached,
output_tokens: output,
cache_read_input_tokens: cached,
cache_creation_input_tokens: 0,
cache_creation_5m: None,
cache_creation_1h: None,
model: None,
},
));
}
}
/// Provider-reported usage for the trace record; `None` until observed.
fn trace_usage(&self) -> Option<crate::runtime::trace::UsageMeta> {
self.observed_usage
.map(|(input, output, cached)| tr::provider_usage(input, output, cached))
}
fn finish(&mut self) {
for tool in self.active_tools.drain(..) {
if !tool.id.is_empty()
&& !tool.name.is_empty()
&& !self.completed_tools.iter().any(|done| done.id == tool.id)
{
self.emitted_output = true;
self.completed_tools.push(ToolCall {
id: tool.id,
kind: "function".to_string(),
function: super::types::FunctionCall {
name: tool.name,
arguments: tool.arguments,
},
});
}
}
}
}
fn handle_events(
events: &[OaiEvent],
tx: &mpsc::UnboundedSender<StreamEvent>,
text_acc: &mut String,
tool_blocks: &mut Vec<Value>,
name_map: &translate::ToolNameMap,
suppress_stream_deltas: bool,
) {
for ev in events {
if let OaiEvent::TextDelta(t) = ev {
text_acc.push_str(t);
}
if let OaiEvent::ToolCallsComplete { calls, .. } = ev {
tool_blocks.extend(translate::tool_calls_to_content_blocks(calls, name_map));
}
if !suppress_stream_deltas {
if let Some(se) = translate::oai_event_to_llm(ev) {
let _ = tx.send(se);
}
}
}
}
#[cfg(test)]
mod codex_input_messages_tests {
//! Regression tests for the Codex Responses-API `input` shape.
//!
//! Background: the Responses API distinguishes two ids per tool
//! invocation — `id` (the *output item id*, prefix `fc_…`) and
//! `call_id` (the *function call id*, prefix `call_…`). When echoing
//! a previous `function_call` back as an input item, supplying an
//! `id` whose value is *not* a `fc_…` triggers
//!
//! 400 Bad Request: Invalid 'input[N].id': 'call_…'.
//! Expected an ID that begins with 'fc'.
//!
//! `id` is *optional* on input items — only `call_id` is required to
//! correlate the eventual `function_call_output`. We elect not to
//! emit `id` unless we actually have a real `fc_…` value to send.
use super::super::types::{ChatMessage, FunctionCall, ToolCall};
use super::*;
fn sample_tool_call() -> ToolCall {
ToolCall {
id: "call_nZYquCuGUh8Qs9H51dwHMDgs".to_string(),
kind: "function".to_string(),
function: FunctionCall {
name: "bash".to_string(),
arguments: r#"{"command":"ls"}"#.to_string(),
},
}
}
fn translated_input(messages: Vec<Value>) -> Vec<Value> {
let messages: Vec<crate::SharedMessage> =
messages.into_iter().map(std::sync::Arc::new).collect();
codex_input_messages(translate::messages_to_oai(
&messages,
&None,
&translate::ToolNameMap::default(),
))
}
#[test]
fn image_text_document_and_pdf_exact_responses_wire() {
let out = translated_input(vec![json!({"role":"user","content":[
{"type":"text","text":"Compare these."},
{"type":"image","source":{"type":"base64","media_type":"image/png","data":"aW1hZ2U="}},
{"type":"document","title":"notes.txt","source":{"type":"text","media_type":"text/plain","data":"Text document\ncontents"}},
{"type":"document","title":"report.pdf","source":{"type":"base64","media_type":"application/pdf","data":"JVBERi0="}},
{"type":"text","text":"Then summarize."}
]})]);
assert_eq!(
out,
vec![json!({"role":"user","content":[
{"type":"input_text","text":"Compare these."},
{"type":"input_text","text":"[attachment origin=user type=image media_type=\"image/png\"; lower-authority data]\n"},
{"type":"input_image","image_url":"data:image/png;base64,aW1hZ2U="},
{"type":"input_text","text":"[/attachment]\n"},
{"type":"input_text","text":"[attachment origin=user type=document media_type=\"text/plain\" filename=\"notes.txt\"; lower-authority data]\n"},
{"type":"input_text","text":"Text document\ncontents"},
{"type":"input_text","text":"[/attachment]\n"},
{"type":"input_text","text":"[attachment origin=user type=document media_type=\"application/pdf\" filename=\"report.pdf\"; lower-authority data]\n"},
{"type":"input_file","filename":"report.pdf","file_data":"data:application/pdf;base64,JVBERi0="},
{"type":"input_text","text":"[/attachment]\n"},
{"type":"input_text","text":"Then summarize."}
]})]
);
}
#[test]
fn mixed_assistant_text_and_all_tool_outputs_precede_lifted_responses_media() {
let out = translated_input(vec![
json!({"role":"assistant","content":[
{"type":"text","text":"Reading both."},
{"type":"tool_use","id":"t1","name":"read","input":{}},
{"type":"tool_use","id":"t2","name":"read","input":{}}
]}),
json!({"role":"user","content":[
{"type":"tool_result","tool_use_id":"t1","content":[
{"type":"text","text":"First result"},
{"type":"image","source":{"type":"base64","media_type":"image/png","data":"aW1hZ2U="}},
{"type":"document","title":"notes.txt","source":{"type":"text","media_type":"text/plain","data":"DOCUMENT_BODY_SENTINEL"}}
]},
{"type":"tool_result","tool_use_id":"t2","content":[
{"type":"text","text":"Second result"},
{"type":"document","title":"report.pdf","source":{"type":"base64","media_type":"application/pdf","data":"JVBERi0="}}
]}
]}),
]);
assert_eq!(
out,
vec![
json!({"role":"assistant","content":"Reading both."}),
json!({"type":"function_call","call_id":"t1","name":"read","arguments":"{}"}),
json!({"type":"function_call","call_id":"t2","name":"read","arguments":"{}"}),
json!({"type":"function_call_output","call_id":"t1","output":"First result\n[attachment origin=tool_result tool_use_id=\"t1\" tool_name=\"read\" type=image media_type=\"image/png\"; lower-authority data; supplied in following user content]\n\n[attachment origin=tool_result tool_use_id=\"t1\" tool_name=\"read\" type=document media_type=\"text/plain\" filename=\"notes.txt\"; lower-authority data; supplied in following user content]\n"}),
json!({"type":"function_call_output","call_id":"t2","output":"Second result\n[attachment origin=tool_result tool_use_id=\"t2\" tool_name=\"read\" type=document media_type=\"application/pdf\" filename=\"report.pdf\"; lower-authority data; supplied in following user content]\n"}),
json!({"role":"user","content":[
{"type":"input_text","text":"[attachment origin=tool_result tool_use_id=\"t1\" tool_name=\"read\" type=image media_type=\"image/png\"; lower-authority data]\n"},
{"type":"input_image","image_url":"data:image/png;base64,aW1hZ2U="},
{"type":"input_text","text":"[/attachment]\n"},
{"type":"input_text","text":"[attachment origin=tool_result tool_use_id=\"t1\" tool_name=\"read\" type=document media_type=\"text/plain\" filename=\"notes.txt\"; lower-authority data]\n"},
{"type":"input_text","text":"DOCUMENT_BODY_SENTINEL"},
{"type":"input_text","text":"[/attachment]\n"},
{"type":"input_text","text":"[attachment origin=tool_result tool_use_id=\"t2\" tool_name=\"read\" type=document media_type=\"application/pdf\" filename=\"report.pdf\"; lower-authority data]\n"},
{"type":"input_file","filename":"report.pdf","file_data":"data:application/pdf;base64,JVBERi0="},
{"type":"input_text","text":"[/attachment]\n"}
]})
]
);
for output in out
.iter()
.filter(|item| item["type"] == "function_call_output")
{
let text = output["output"].as_str().unwrap();
for sentinel in [
"aW1hZ2U=",
"JVBERi0=",
"DOCUMENT_BODY_SENTINEL",
"\"source\"",
] {
assert!(
!text.contains(sentinel),
"tool output must contain only text and metadata"
);
}
}
}
#[test]
fn text_only_responses_input_keeps_strings_and_mixed_assistant_text() {
let mut assistant = ChatMessage::assistant_tool_calls(vec![sample_tool_call()]);
assistant.content = Some("Checking.".into());
let out = codex_input_messages(vec![
ChatMessage::system("rules"),
ChatMessage::user("question"),
assistant,
ChatMessage::tool_result("call_nZYquCuGUh8Qs9H51dwHMDgs", "bash", "done"),
ChatMessage::assistant("answer"),
]);
assert_eq!(
out,
vec![
json!({"role":"system","content":"rules"}),
json!({"role":"user","content":"question"}),
json!({"role":"assistant","content":"Checking."}),
json!({"type":"function_call","call_id":"call_nZYquCuGUh8Qs9H51dwHMDgs","name":"bash","arguments":"{\"command\":\"ls\"}"}),
json!({"type":"function_call_output","call_id":"call_nZYquCuGUh8Qs9H51dwHMDgs","output":"done"}),
json!({"role":"assistant","content":"answer"}),
]
);
}
/// Anthropic-shaped stored history covering every assistant shape:
/// mixed text+tool_use, tool-only, text-only, plus a plain user message
/// right after tool results (forces the `" "` separator).
fn phase_history() -> Vec<Value> {
vec![
json!({"role":"user","content":"build it"}),
json!({"role":"assistant","content":[
{"type":"thinking","thinking":"hidden","signature":"s"},
{"type":"text","text":"Mapping the repo first."},
{"type":"tool_use","id":"call_a","name":"bash","input":{"command":"ls"}},
]}),
json!({"role":"user","content":[
{"type":"tool_result","tool_use_id":"call_a","content":"src"},
]}),
json!({"role":"assistant","content":[
{"type":"tool_use","id":"call_b","name":"bash","input":{"command":"pwd"}},
]}),
json!({"role":"user","content":[
{"type":"tool_result","tool_use_id":"call_b","content":"/w"},
]}),
json!({"role":"user","content":"also check logs"}),
json!({"role":"assistant","content":[{"type":"text","text":"Done: all good."}]}),
json!({"role":"user","content":"thanks"}),
json!({"role":"assistant","content":"Plain string answer."}),
]
}
fn oai(messages: Vec<Value>) -> Vec<ChatMessage> {
let messages: Vec<crate::SharedMessage> =
messages.into_iter().map(std::sync::Arc::new).collect();
translate::messages_to_oai(&messages, &None, &translate::ToolNameMap::default())
}
#[test]
fn muse_input_infers_phase_from_message_structure() {
let out = muse_input_messages(oai(phase_history()));
assert_eq!(
out,
vec![
json!({"role":"user","content":"build it"}),
// Mixed message: its text is commentary and precedes its call.
json!({"type":"message","role":"assistant","phase":"commentary",
"content":[{"type":"output_text","text":"Mapping the repo first."}]}),
json!({"type":"function_call","call_id":"call_a","name":"bash","arguments":"{\"command\":\"ls\"}"}),
json!({"type":"function_call_output","call_id":"call_a","output":"src"}),
// Tool-only message: unchanged, no message item invented.
json!({"type":"function_call","call_id":"call_b","name":"bash","arguments":"{\"command\":\"pwd\"}"}),
json!({"type":"function_call_output","call_id":"call_b","output":"/w"}),
// Harness separator keeps the legacy shape (not model text).
json!({"role":"assistant","content":" "}),
json!({"role":"user","content":"also check logs"}),
// Text-only messages (block and string storage) are final.
json!({"type":"message","role":"assistant","phase":"final_answer",
"content":[{"type":"output_text","text":"Done: all good."}]}),
json!({"role":"user","content":"thanks"}),
json!({"type":"message","role":"assistant","phase":"final_answer",
"content":[{"type":"output_text","text":"Plain string answer."}]}),
]
);
}
#[test]
fn codex_and_xai_input_stay_phase_less_for_the_same_history() {
// Codex and xAI both build input with `codex_input_messages`; the
// Muse change must not alter a byte of it.
let out = codex_input_messages(oai(phase_history()));
assert_eq!(
out,
vec![
json!({"role":"user","content":"build it"}),
json!({"role":"assistant","content":"Mapping the repo first."}),
json!({"type":"function_call","call_id":"call_a","name":"bash","arguments":"{\"command\":\"ls\"}"}),
json!({"type":"function_call_output","call_id":"call_a","output":"src"}),
json!({"type":"function_call","call_id":"call_b","name":"bash","arguments":"{\"command\":\"pwd\"}"}),
json!({"type":"function_call_output","call_id":"call_b","output":"/w"}),
json!({"role":"assistant","content":" "}),
json!({"role":"user","content":"also check logs"}),
json!({"role":"assistant","content":"Done: all good."}),
json!({"role":"user","content":"thanks"}),
json!({"role":"assistant","content":"Plain string answer."}),
]
);
assert!(!serde_json::to_string(&out).unwrap().contains("phase"));
}
#[test]
fn muse_input_leaves_non_assistant_and_tool_only_messages_identical() {
let messages = || {
vec![
ChatMessage::system("rules"),
ChatMessage::user("question"),
ChatMessage::assistant_tool_calls(vec![sample_tool_call()]),
ChatMessage::tool_result("call_nZYquCuGUh8Qs9H51dwHMDgs", "bash", "done"),
]
};
assert_eq!(
muse_input_messages(messages()),
codex_input_messages(messages())
);
}
#[test]
fn codex_instructions_appends_autonomous_loop_policy() {
let instructions = codex_instructions(&Some("Project-specific rules.".to_string()));
assert!(instructions.contains("Project-specific rules."));
assert!(instructions.contains("Do not stop at phase boundaries"));
assert!(instructions.contains("Do not ask the user whether to continue"));
assert!(
instructions.contains("continue autonomously until the full requested job is complete")
);
}
#[test]
fn function_call_input_omits_non_fc_id() {
let messages = vec![ChatMessage::assistant_tool_calls(vec![sample_tool_call()])];
let out = codex_input_messages(messages);
assert_eq!(out.len(), 1, "one tool_call → one input item");
let item = &out[0];
assert_eq!(
item.get("type").and_then(Value::as_str),
Some("function_call")
);
assert!(
item.get("id").is_none(),
"must not echo a non-`fc_` id back; got {:?}",
item.get("id"),
);
assert_eq!(
item.get("call_id").and_then(Value::as_str),
Some("call_nZYquCuGUh8Qs9H51dwHMDgs"),
);
assert_eq!(item.get("name").and_then(Value::as_str), Some("bash"));
}
#[test]
fn function_call_input_keeps_real_fc_id() {
// If we ever do have a genuine `fc_…` id (round-tripped from the
// Responses API), we *should* echo it.
let mut call = sample_tool_call();
call.id = "fc_abc123".to_string();
let messages = vec![ChatMessage::assistant_tool_calls(vec![call])];
let out = codex_input_messages(messages);
let item = &out[0];
assert_eq!(item.get("id").and_then(Value::as_str), Some("fc_abc123"));
assert_eq!(
item.get("call_id").and_then(Value::as_str),
Some("fc_abc123")
);
}
#[test]
fn function_call_output_round_trips_call_id() {
// The follow-up tool message must reference the original call_id.
let messages = vec![ChatMessage::tool_result(
"call_nZYquCuGUh8Qs9H51dwHMDgs",
"bash",
"total 0",
)];
let out = codex_input_messages(messages);
let item = &out[0];
assert_eq!(
item.get("type").and_then(Value::as_str),
Some("function_call_output"),
);
assert_eq!(
item.get("call_id").and_then(Value::as_str),
Some("call_nZYquCuGUh8Qs9H51dwHMDgs"),
);
assert_eq!(item.get("output").and_then(Value::as_str), Some("total 0"));
}
}
#[cfg(test)]
mod codex_decoder_tests {
//! Regression tests for `CodexSseDecoder`.
//!
//! The decoder is sync — we drive it via `push_line` and capture
//! emitted `StreamEvent`s from an `unbounded_channel` using
//! `try_recv`, no async runtime needed.
use super::*;
use crate::runtime::types::{LlmEvent, SessionEvent, StreamEvent};
fn collect_events(rx: &mut mpsc::UnboundedReceiver<StreamEvent>) -> Vec<StreamEvent> {
let mut out = Vec::new();
while let Ok(ev) = rx.try_recv() {
out.push(ev);
}
out
}
fn drive(lines: &[&str]) -> (CodexSseDecoder, String, Vec<StreamEvent>) {
let (tx, mut rx) = mpsc::unbounded_channel();
let mut decoder = CodexSseDecoder::default();
let mut text_acc = String::new();
for line in lines {
decoder.push_line(line, &tx, &mut text_acc);
}
let events = collect_events(&mut rx);
(decoder, text_acc, events)
}
fn drive_with(
mut decoder: CodexSseDecoder,
lines: &[&str],
) -> (CodexSseDecoder, String, Vec<StreamEvent>) {
let (tx, mut rx) = mpsc::unbounded_channel();
let mut text_acc = String::new();
for line in lines {
decoder.push_line(line, &tx, &mut text_acc);
}
let events = collect_events(&mut rx);
(decoder, text_acc, events)
}
#[test]
fn decoder_captures_phase_of_the_last_message_item() {
// Muse shape (ask1.sse): reasoning, commentary message, function_call.
let (decoder, text, _) = drive(&[
r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"rs_0"}}"#,
"",
r#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"message","role":"assistant","phase":"commentary","content":[]}}"#,
"",
r#"data: {"type":"response.output_text.delta","output_index":1,"delta":"Mapping first."}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":1,"item":{"type":"message","role":"assistant","phase":"commentary","content":[{"type":"output_text","text":"Mapping first."}]}}"#,
"",
r#"data: {"type":"response.output_item.added","output_index":2,"item":{"type":"function_call","call_id":"call_1","name":"bash"}}"#,
"",
]);
assert_eq!(decoder.last_message_phase.as_deref(), Some("commentary"));
assert_eq!(decoder.last_message_index, Some(1));
assert_eq!(decoder.last_message_text(), Some("Mapping first."));
assert_eq!(text, "Mapping first.");
// A later phase-less (final) message replaces it.
let (decoder, _, _) = drive(&[
r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"message","role":"assistant","phase":"commentary","content":[]}}"#,
"",
r#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"message","role":"assistant","content":[]}}"#,
"",
r#"data: {"type":"response.output_text.delta","output_index":1,"delta":"Answer"}"#,
"",
]);
assert_eq!(decoder.last_message_phase, None);
assert_eq!(decoder.last_message_text(), Some("Answer"));
}
const TRUNCATED_STREAM: [&str; 6] = [
r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"message","role":"assistant","content":[]}}"#,
"",
r#"data: {"type":"response.output_text.delta","output_index":0,"delta":"Key facts:"}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"Key facts:\nWhich format?"}]}}"#,
"",
];
#[test]
fn muse_decoder_backfills_unstreamed_message_suffix_once() {
let mut decoder = CodexSseDecoder::for_provider("Meta Muse");
decoder.backfill_message_text = true;
let mut lines = TRUNCATED_STREAM.to_vec();
// A repeated done event must not emit the suffix twice.
lines.extend_from_slice(&TRUNCATED_STREAM[4..]);
let (decoder, text, events) = drive_with(decoder, &lines);
assert_eq!(text, "Key facts:\nWhich format?");
let texts: Vec<_> = events
.iter()
.filter_map(|e| match e {
StreamEvent::Llm(LlmEvent::Text(t)) => Some(t.as_str()),
_ => None,
})
.collect();
assert_eq!(texts, vec!["Key facts:", "\nWhich format?"]);
assert_eq!(
decoder.last_message_text(),
Some("Key facts:\nWhich format?")
);
}
#[test]
fn backfill_is_off_by_default_and_never_rewrites_divergent_text() {
// Default (Codex) decoder: decoding unchanged.
let (_, text, _) = drive(&TRUNCATED_STREAM);
assert_eq!(text, "Key facts:");
// Muse decoder: final text that does not extend the stream is ignored.
let mut decoder = CodexSseDecoder::for_provider("Meta Muse");
decoder.backfill_message_text = true;
let (_, text, _) = drive_with(
decoder,
&[
r#"data: {"type":"response.output_text.delta","output_index":0,"delta":"Hello"}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"message","content":[{"type":"output_text","text":"Goodbye world"}]}}"#,
"",
],
);
assert_eq!(text, "Hello");
}
#[test]
fn hostile_output_index_does_not_grow_active_tools_unbounded() {
// A provider (or wire corruption) sending an absurd `output_index` must
// not drive an unbounded `active_tools` grow (OOM/DoS). ensure_tool clamps
// to MAX_TOOLS. Without the clamp this test would attempt to allocate ~1e9
// accumulators and hang/OOM.
let lines = [
r#"data: {"type":"response.function_call_arguments.delta","output_index":1000000000,"delta":"x"}"#,
"",
];
let (decoder, _text, _events) = drive(&lines);
assert!(
decoder.active_tools.len() <= 4096,
"active_tools grew to {} — output_index not bounded",
decoder.active_tools.len()
);
}
#[test]
fn text_delta_aggregates_into_text_acc_and_emits_text_events() {
let lines = [
r#"data: {"type":"response.output_text.delta","delta":"Hello, "}"#,
"",
r#"data: {"type":"response.output_text.delta","delta":"world!"}"#,
"",
];
let (_decoder, text_acc, events) = drive(&lines);
assert_eq!(text_acc, "Hello, world!");
let texts: Vec<_> = events
.iter()
.filter_map(|e| match e {
StreamEvent::Llm(LlmEvent::Text(t)) => Some(t.as_str()),
_ => None,
})
.collect();
assert_eq!(texts, vec!["Hello, ", "world!"]);
}
#[test]
fn hosted_web_search_items_are_not_tool_uses() {
// OpenAI-hosted web search (SYNAPS_CODEX_WEB_SEARCH) streams
// `web_search_call` output items. They run server-side, so the decoder
// must not surface them as local tool uses — only the answer text and
// real function calls come through.
let lines = [
r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"web_search_call","id":"ws_1","status":"in_progress"}}"#,
"",
r#"data: {"type":"response.web_search_call.searching","output_index":0,"item_id":"ws_1"}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"web_search_call","id":"ws_1","status":"completed","action":{"type":"search","query":"ram dass puja chat youtube"}}}"#,
"",
r#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","call_id":"call_1","name":"get_asset"}}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","call_id":"call_1","name":"get_asset","arguments":"{}"}}"#,
"",
r#"data: {"type":"response.output_text.delta","output_index":2,"delta":"Found it."}"#,
"",
r#"data: {"type":"response.output_text.annotation.added","output_index":2,"annotation":{"type":"url_citation","url":"https://www.youtube.com/watch?v=sH-u96rpYSw"}}"#,
"",
];
let (decoder, text, events) = drive(&lines);
assert_eq!(text, "Found it.");
assert_eq!(decoder.completed_tools.len(), 1);
assert_eq!(decoder.completed_tools[0].function.name, "get_asset");
let starts = events
.iter()
.filter(|e| matches!(e, StreamEvent::Llm(LlmEvent::ToolUseStart { .. })))
.count();
assert_eq!(starts, 1, "web_search_call must not emit a ToolUseStart");
}
#[test]
fn web_search_env_value_maps_to_hosted_tool() {
use super::codex_web_search_tool;
assert_eq!(codex_web_search_tool(None), None);
assert_eq!(codex_web_search_tool(Some("")), None);
assert_eq!(codex_web_search_tool(Some("off")), None);
assert_eq!(codex_web_search_tool(Some("bogus")), None);
for on in ["live", "LIVE", " on ", "true", "1", "yes"] {
assert_eq!(
codex_web_search_tool(Some(on)),
Some(serde_json::json!({ "type": "web_search" })),
"{on:?}"
);
}
assert_eq!(
codex_web_search_tool(Some("cached")),
Some(serde_json::json!({ "type": "web_search", "external_web_access": false }))
);
}
#[test]
fn single_function_call_completes_via_output_item_done() {
let lines = [
r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_abc","name":"bash"}}"#,
"",
r#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"cmd\""}"#,
"",
r#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":":\"ls\"}"}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_abc","name":"bash","arguments":"{\"cmd\":\"ls\"}"}}"#,
"",
];
let (decoder, _text, events) = drive(&lines);
assert_eq!(decoder.completed_tools.len(), 1);
let tool = &decoder.completed_tools[0];
assert_eq!(tool.id, "call_abc");
assert_eq!(tool.function.name, "bash");
assert_eq!(tool.function.arguments, r#"{"cmd":"ls"}"#);
// Exactly one ToolUseStart for the tool.
let starts: Vec<_> = events
.iter()
.filter_map(|e| match e {
StreamEvent::Llm(LlmEvent::ToolUseStart { tool_name, tool_id }) => {
Some((tool_name.as_str(), tool_id.as_str()))
}
_ => None,
})
.collect();
assert_eq!(
starts,
vec![("bash", "call_abc")],
"exactly one ToolUseStart with correct tool_id"
);
// Two argument deltas streamed (each carrying the tool_id so
// parallel calls can be routed correctly by the chat UI).
let deltas: Vec<_> = events
.iter()
.filter_map(|e| match e {
StreamEvent::Llm(LlmEvent::ToolUseDelta { tool_id, delta }) => {
Some((tool_id.as_str(), delta.as_str()))
}
_ => None,
})
.collect();
assert_eq!(
deltas,
vec![("call_abc", r#"{"cmd""#), ("call_abc", r#":"ls"}"#)]
);
}
#[test]
fn parallel_tool_calls_indexed_separately() {
let lines = [
r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"bash"}}"#,
"",
r#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","call_id":"call_2","name":"read"}}"#,
"",
r#"data: {"type":"response.function_call_arguments.delta","output_index":1,"delta":"{\"path\":\"a\"}"}"#,
"",
r#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"cmd\":\"ls\"}"}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"bash","arguments":"{\"cmd\":\"ls\"}"}}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","call_id":"call_2","name":"read","arguments":"{\"path\":\"a\"}"}}"#,
"",
];
let (decoder, _text, _events) = drive(&lines);
assert_eq!(decoder.completed_tools.len(), 2);
let mut by_id: std::collections::BTreeMap<&str, &ToolCall> =
std::collections::BTreeMap::new();
for tool in &decoder.completed_tools {
by_id.insert(tool.id.as_str(), tool);
}
assert_eq!(by_id["call_1"].function.name, "bash");
assert_eq!(by_id["call_1"].function.arguments, r#"{"cmd":"ls"}"#);
assert_eq!(by_id["call_2"].function.name, "read");
assert_eq!(by_id["call_2"].function.arguments, r#"{"path":"a"}"#);
}
#[test]
fn output_item_done_emits_tool_use_event() {
// Regression: the codex decoder must emit `LlmEvent::ToolUse` once a
// function_call's `output_item.done` arrives so the chat UI can
// collapse `ChatMessage::ToolUseStart` (animated) into the finalized
// `ChatMessage::ToolUse`. Without this the bash-trace animation
// persists forever and parallel tool blocks render as "still
// running" even after they've completed.
let lines = [
r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_abc","name":"bash"}}"#,
"",
r#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"command\":\"ls\"}"}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_abc","name":"bash","arguments":"{\"command\":\"ls\"}"}}"#,
"",
];
let (_decoder, _text, events) = drive(&lines);
let tool_uses: Vec<_> = events
.iter()
.filter_map(|e| match e {
StreamEvent::Llm(LlmEvent::ToolUse {
tool_name,
tool_id,
input,
}) => Some((tool_name.as_str(), tool_id.as_str(), input.clone())),
_ => None,
})
.collect();
assert_eq!(
tool_uses.len(),
1,
"expected exactly one ToolUse finalize event"
);
assert_eq!(tool_uses[0].0, "bash");
assert_eq!(tool_uses[0].1, "call_abc");
assert_eq!(
tool_uses[0].2,
serde_json::json!({"command": "ls"}),
"input must be parsed as a JSON Value, not a string"
);
}
#[test]
fn parallel_tool_calls_emit_tool_use_per_index() {
// Regression: parallel tool calls must each get their own ToolUse
// finalize event with the correct tool_id, so the chat UI can route
// their results back to the right block by id.
let lines = [
r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"bash"}}"#,
"",
r#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","call_id":"call_2","name":"read"}}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"bash","arguments":"{\"command\":\"ls\"}"}}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","call_id":"call_2","name":"read","arguments":"{\"path\":\"a\"}"}}"#,
"",
];
let (_decoder, _text, events) = drive(&lines);
let tool_uses: Vec<_> = events
.iter()
.filter_map(|e| match e {
StreamEvent::Llm(LlmEvent::ToolUse {
tool_name,
tool_id,
input,
}) => Some((tool_name.clone(), tool_id.clone(), input.clone())),
_ => None,
})
.collect();
assert_eq!(tool_uses.len(), 2, "one ToolUse finalize per parallel call");
let by_id: std::collections::BTreeMap<&str, &(String, String, serde_json::Value)> =
tool_uses.iter().map(|t| (t.1.as_str(), t)).collect();
assert_eq!(by_id["call_1"].0, "bash");
assert_eq!(by_id["call_1"].2, serde_json::json!({"command": "ls"}));
assert_eq!(by_id["call_2"].0, "read");
assert_eq!(by_id["call_2"].2, serde_json::json!({"path": "a"}));
}
#[test]
fn malformed_arguments_emit_tool_use_with_parse_error() {
// If the model produces invalid JSON arguments, surface a structured
// parse error in the `input` (matching how the Anthropic path
// handles it via parse_tool_input) so the agent loop can return an
// error tool_result instead of silently dropping the tool.
let lines = [
r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_bad","name":"bash"}}"#,
"",
r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_bad","name":"bash","arguments":"{not json"}}"#,
"",
];
let (_decoder, _text, events) = drive(&lines);
let tool_use = events.iter().find_map(|e| match e {
StreamEvent::Llm(LlmEvent::ToolUse { input, .. }) => Some(input.clone()),
_ => None,
});
let input = tool_use.expect("ToolUse event missing");
assert!(
input.get("__parse_error").and_then(Value::as_str).is_some(),
"malformed arguments must surface __parse_error, got {input}"
);
}
#[test]
fn response_completed_emits_usage_event() {
let lines = [
r#"data: {"type":"response.completed","response":{"usage":{"input_tokens":42,"output_tokens":17}}}"#,
"",
];
let (_decoder, _text, events) = drive(&lines);
let usage = events.iter().find_map(|e| match e {
StreamEvent::Session(SessionEvent::Usage {
input_tokens,
output_tokens,
..
}) => Some((*input_tokens, *output_tokens)),
_ => None,
});
assert_eq!(usage, Some((42, 17)));
}
/// Responses-API usage nests cache hits under input_tokens_details, and
/// input_tokens INCLUDES them. The decoder must surface the cached slice
/// as cache_read (Anthropic semantics) and subtract it from input —
/// dropping it (the old hardcoded 0) made every codex turn look 100%
/// uncached (incident: 2026-07-16 codex cache investigation).
#[test]
fn response_completed_surfaces_cached_tokens_without_double_count() {
let lines = [
r#"data: {"type":"response.completed","response":{"usage":{"input_tokens":1000,"input_tokens_details":{"cached_tokens":900},"output_tokens":17}}}"#,
"",
];
let (_decoder, _text, events) = drive(&lines);
let usage = events.iter().find_map(|e| match e {
StreamEvent::Session(SessionEvent::Usage {
input_tokens,
output_tokens,
cache_read_input_tokens,
..
}) => Some((*input_tokens, *output_tokens, *cache_read_input_tokens)),
_ => None,
});
assert_eq!(
usage,
Some((100, 17, 900)),
"input must exclude the cached slice; cache_read must carry it"
);
}
#[test]
fn response_completed_with_zero_usage_emits_nothing() {
let lines = [
r#"data: {"type":"response.output_text.delta","delta":"ok"}"#,
"",
r#"data: {"type":"response.completed","response":{"usage":{"input_tokens":0,"output_tokens":0}}}"#,
"",
];
let (decoder, _text, events) = drive(&lines);
let any_usage = events
.iter()
.any(|e| matches!(e, StreamEvent::Session(SessionEvent::Usage { .. })));
assert!(!any_usage, "zero-token usage should be suppressed");
assert_eq!(decoder.terminal_result(), Ok(()));
}
#[test]
fn response_failed_is_a_terminal_failure_without_retaining_provider_text() {
let lines = [
r#"data: {"type":"response.failed","response":{"error":{"type":"server_error","message":"ECHOED:secret prompt"}}}"#,
"",
];
let (decoder, text, events) = drive(&lines);
let failure = decoder.terminal_result().expect_err("must fail");
// Exact server_error is now preserved as a transient capacity class.
assert_eq!(failure.code, "responses_capacity");
assert!(failure.is_retryable());
assert_eq!(failure.message, format!("Codex{RESPONSES_CAPACITY_SUFFIX}"));
assert!(!failure.message.contains("ECHOED"));
assert!(text.is_empty());
assert!(events.is_empty());
}
#[test]
fn top_level_error_event_is_a_terminal_failure() {
let lines = [
r#"data: {"type":"error","error":{"type":"server_error","message":"private"}}"#,
"",
];
let (decoder, _text, _events) = drive(&lines);
assert_eq!(
decoder.terminal_result().expect_err("must fail").code,
"responses_capacity"
);
}
#[test]
fn response_incomplete_is_not_misreported_as_empty_success() {
let lines = [
r#"data: {"type":"response.incomplete","response":{"incomplete_details":{"reason":"max_output_tokens"}}}"#,
"",
];
let (decoder, _text, _events) = drive(&lines);
let failure = decoder.terminal_result().expect_err("must fail");
assert_eq!(failure.code, "responses_incomplete");
assert!(failure.message.contains("incomplete"));
assert!(!failure.message.contains("max_output_tokens"));
}
#[test]
fn completed_without_text_or_tools_fails_at_transport_boundary() {
let lines = [
r#"data: {"type":"response.completed","response":{"usage":{"input_tokens":7,"output_tokens":0}}}"#,
"",
];
let (decoder, _text, _events) = drive(&lines);
assert_eq!(
decoder.terminal_result().expect_err("must fail").code,
"responses_empty"
);
}
#[test]
fn eof_without_terminal_event_fails_closed() {
let lines = [
r#"data: {"type":"response.created","response":{"id":"resp_123"}}"#,
"",
];
let (decoder, _text, _events) = drive(&lines);
assert_eq!(
decoder.terminal_result().expect_err("must fail").code,
"responses_missing_terminal"
);
}
/// The local-log sanitizer must pass lowercase snake-case identifiers
/// verbatim, collapse secret-shaped identifiers to a shape marker, and
/// never emit free-form provider text.
#[test]
fn error_identifier_sanitizer_never_passes_freeform_text() {
assert_eq!(sanitize_error_identifier(None), "absent");
assert_eq!(
sanitize_error_identifier(Some("rate_limit_exceeded")),
"rate_limit_exceeded"
);
assert_eq!(
sanitize_error_identifier(Some("context_length_exceeded")),
"context_length_exceeded"
);
assert_eq!(
sanitize_error_identifier(Some("some_new_code2")),
"some_new_code2"
);
// Mixed-case / dashed tokens can be secret-shaped (API keys).
assert_eq!(
sanitize_error_identifier(Some("sk-Abc123XYZ")),
"unlisted_identifier"
);
assert_eq!(
sanitize_error_identifier(Some("Server.Error")),
"unlisted_identifier"
);
assert_eq!(
sanitize_error_identifier(Some("ECHOED: user secret prompt")),
"unsafe_or_freeform"
);
assert_eq!(sanitize_error_identifier(Some("")), "unsafe_or_freeform");
let long = "a".repeat(65);
assert_eq!(sanitize_error_identifier(Some(&long)), "unsafe_or_freeform");
}
/// Context-window overflow maps to a specific, actionable message; every
/// other terminal failure keeps the neutral details-withheld default. The
/// mapping only ever reads sanitized enum identifiers, never free-text.
#[test]
fn terminal_failure_user_message_is_specific_only_for_known_codes() {
assert!(user_message_for_terminal_failure(
"Codex",
"invalid_request_error",
"context_length_exceeded",
false
)
.contains("/compact"));
// Code carried on `type` rather than `code` still maps.
assert!(user_message_for_terminal_failure(
"Codex",
"context_length_exceeded",
"absent",
false
)
.contains("context window"));
// Unknown/other codes keep the neutral default (no remedy claim).
let default = user_message_for_terminal_failure(
"Codex",
"server_error",
"unlisted_identifier",
false,
);
assert!(default.starts_with("Codex response failed in stream."));
assert!(!default.contains("/compact"));
// Capacity gets the specific remedy, labelled for the provider.
let capacity = user_message_for_terminal_failure("xAI", "absent", "absent", true);
assert_eq!(
capacity,
"xAI reports the model is at capacity (retries exhausted). Try again in a few minutes or switch models with /model."
);
}
/// The exact flat envelope xAI emits (HTTP 200, one SSE event, close):
/// no nested `error` object, `code` null, free-text `message`. It must
/// classify as retryable capacity — labelled for the provider, and the
/// provider prose must never reach the surfaced message.
#[test]
fn xai_flat_capacity_error_is_retryable_capacity_failure() {
let lines = [
r#"data: {"sequence_number":0,"type":"error","code":null,"message":"The model is currently at capacity due to high demand. Please try again in a few minutes, or use a higher service tier for priority processing: https://docs.x.ai/developers/advanced-api-usage/priority-processing","param":null}"#,
"",
];
let (tx, mut rx) = mpsc::unbounded_channel();
let mut decoder = CodexSseDecoder::for_provider("xAI");
let mut text = String::new();
for line in lines {
decoder.push_line(line, &tx, &mut text);
}
decoder.finish();
let failure = decoder.terminal_result().expect_err("must fail");
assert_eq!(failure.code, "responses_capacity");
assert!(failure.is_retryable());
assert_eq!(
failure.message,
"xAI reports the model is at capacity (retries exhausted). Try again in a few minutes or switch models with /model."
);
for banned in ["high demand", "docs.x.ai", "service tier", "Codex"] {
assert!(
!failure.message.contains(banned),
"provider text `{banned}` leaked: {}",
failure.message
);
}
assert!(text.is_empty());
assert!(!decoder.emitted_any_output());
assert!(rx.try_recv().is_err(), "no events for an error-only stream");
}
/// Flat `code` identifiers are honoured when no nested error object is
/// present; only vetted codes classify as capacity.
#[test]
fn flat_code_classifies_capacity_only_for_vetted_codes() {
let capacity = [
r#"data: {"type":"error","code":"rate_limit_exceeded","message":"private"}"#,
"",
];
let (decoder, _, _) = drive(&capacity);
let failure = decoder.terminal_result().expect_err("must fail");
assert_eq!(failure.code, "responses_capacity");
assert!(failure.is_retryable());
let other = [
r#"data: {"type":"error","code":"invalid_api_key","message":"private"}"#,
"",
];
let (decoder, _, _) = drive(&other);
let failure = decoder.terminal_result().expect_err("must fail");
assert_eq!(failure.code, "responses_failed");
assert!(!failure.is_retryable());
assert!(!failure.message.contains("private"));
}
#[test]
fn provider_label_is_carried_into_every_terminal_message() {
let (tx, _rx) = mpsc::unbounded_channel();
let mut text = String::new();
let mut decoder = CodexSseDecoder::for_provider("xAI");
decoder.push_line(
r#"data: {"type":"response.completed","response":{"usage":{"input_tokens":1,"output_tokens":0}}}"#,
&tx,
&mut text,
);
decoder.push_line("", &tx, &mut text);
let failure = decoder.terminal_result().expect_err("empty");
assert_eq!(failure.code, "responses_empty");
assert!(failure.is_retryable());
assert_eq!(
failure.message,
"xAI completed without text or tool output. Retry the request."
);
let decoder = CodexSseDecoder::for_provider("xAI");
let failure = decoder.terminal_result().expect_err("missing terminal");
assert_eq!(failure.code, "responses_missing_terminal");
assert!(failure.is_retryable());
assert!(failure.message.starts_with("xAI response stream ended"));
let mut decoder = CodexSseDecoder::for_provider("xAI");
decoder.push_line(
r#"data: {"type":"response.incomplete","response":{"incomplete_details":{"reason":"max_output_tokens"}}}"#,
&tx,
&mut text,
);
decoder.push_line("", &tx, &mut text);
let failure = decoder.terminal_result().expect_err("incomplete");
assert_eq!(failure.code, "responses_incomplete");
assert!(!failure.is_retryable());
assert!(failure.message.starts_with("xAI response was incomplete."));
}
#[test]
fn done_marker_finishes_decoder() {
let lines = [
r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_x","name":"bash"}}"#,
"",
r#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":"{}"}"#,
"",
"data: [DONE]",
"",
];
let (decoder, _text, _events) = drive(&lines);
// active_tools promoted to completed_tools by finish().
assert_eq!(decoder.completed_tools.len(), 1);
assert_eq!(decoder.completed_tools[0].id, "call_x");
assert_eq!(decoder.completed_tools[0].function.arguments, "{}");
}
#[test]
fn finish_is_idempotent_no_double_emit() {
let lines = [
r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_y","name":"bash","arguments":"{}"}}"#,
"",
];
let (mut decoder, _text, _events) = drive(&lines);
assert_eq!(decoder.completed_tools.len(), 1);
// Calling finish() again must not duplicate the tool.
decoder.finish();
assert_eq!(
decoder.completed_tools.len(),
1,
"finish() called twice must not double-emit"
);
}
#[test]
fn finish_drains_active_tools_for_state_hygiene() {
// After [DONE], any leftover active tool entries should have been
// promoted *and* drained from `active_tools`. This guards against
// future code paths that re-call finish() (or new event types
// that would otherwise re-iterate the old buffer).
let lines = [
r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_z","name":"bash"}}"#,
"",
"data: [DONE]",
"",
];
let (decoder, _text, _events) = drive(&lines);
assert_eq!(decoder.completed_tools.len(), 1);
assert!(
decoder.active_tools.is_empty(),
"active_tools must be drained after finish()"
);
}
#[test]
fn unknown_event_types_are_ignored() {
let lines = [
r#"data: {"type":"response.future_unknown_event","payload":{"x":1}}"#,
"",
r#"data: {"type":"response.output_text.delta","delta":"hi"}"#,
"",
];
let (_decoder, text_acc, _events) = drive(&lines);
assert_eq!(text_acc, "hi");
}
}
#[cfg(test)]
mod broker_stream_tests {
//! Broker-boundary streaming tests: the OpenAI-compatible stream path is
//! driven end-to-end through an in-process `LocalBroker` against a fake
//! upstream. The runtime side supplies NO credential — the broker applies
//! it — and SSE deltas are forwarded to the event channel in real time.
use super::*;
use crate::auth::{CredentialBroker, LocalBroker};
use std::sync::Arc;
async fn spawn_fake_openai_sse() -> (String, Arc<std::sync::Mutex<String>>) {
use axum::routing::post;
let seen_auth = Arc::new(std::sync::Mutex::new(String::new()));
let seen = seen_auth.clone();
let app = axum::Router::new().route(
"/chat/completions",
post(move |headers: axum::http::HeaderMap| {
let seen = seen.clone();
async move {
*seen.lock().unwrap() = headers
.get("authorization")
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_string();
let body = concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"Hel\"}}]}\n\n",
"data: {\"choices\":[{\"delta\":{\"content\":\"lo\"}}]}\n\n",
"data: [DONE]\n\n",
);
([("content-type", "text/event-stream")], body)
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
(format!("http://{addr}"), seen_auth)
}
#[tokio::test]
async fn oai_stream_flows_through_broker_and_forwards_deltas() {
let (upstream, seen_auth) = spawn_fake_openai_sse().await;
let broker: Arc<dyn CredentialBroker> = Arc::new(LocalBroker::with_local_base_url(
reqwest::Client::new(),
upstream,
));
let cfg = ProviderConfig {
base_url: "unused-broker-derives-the-url".to_string(),
model: "test-model".to_string(),
provider: "local".to_string(),
};
let (tx, mut rx) = mpsc::unbounded_channel::<StreamEvent>();
let cancel = tokio_util::sync::CancellationToken::new();
let result = call_oai_stream_inner(
&cfg,
&broker,
&[],
&None,
&[],
&tx,
None,
None,
0,
agent_core::reasoning::ReasoningLevel::Medium,
&cancel,
&crate::runtime::trace::TraceContext::disabled(),
true,
false,
0,
)
.await
.expect("stream must complete");
// Final Anthropic-shaped value carries the accumulated text.
assert_eq!(result["content"][0]["text"], "Hello");
// Deltas were forwarded in real time on the event channel.
drop(tx);
let mut streamed = String::new();
while let Ok(ev) = rx.try_recv() {
if let StreamEvent::Llm(crate::runtime::types::LlmEvent::Text(t)) = ev {
streamed.push_str(&t);
}
}
assert_eq!(streamed, "Hello");
// The credential was applied by the broker, not by this call site.
assert_eq!(&*seen_auth.lock().unwrap(), "Bearer local");
}
/// Broker errors surface as typed failures without opening a stream and
/// without any credential material in the message.
#[tokio::test]
async fn suppressed_sync_route_does_not_enqueue_display_deltas() {
let (upstream, _seen_auth) = spawn_fake_openai_sse().await;
let broker: Arc<dyn CredentialBroker> = Arc::new(LocalBroker::with_local_base_url(
reqwest::Client::new(),
upstream,
));
let cfg = ProviderConfig {
base_url: "unused-broker-derives-the-url".to_string(),
model: "test-model".to_string(),
provider: "local".to_string(),
};
let (tx, mut rx) = mpsc::unbounded_channel::<StreamEvent>();
let result = call_oai_stream_inner(
&cfg,
&broker,
&[],
&None,
&[],
&tx,
None,
None,
0,
agent_core::reasoning::ReasoningLevel::Medium,
&tokio_util::sync::CancellationToken::new(),
&crate::runtime::trace::TraceContext::disabled(),
true,
true,
0,
)
.await
.expect("stream must complete");
assert_eq!(result["content"][0]["text"], "Hello");
assert!(
rx.try_recv().is_err(),
"sync route must suppress display deltas at production"
);
}
#[tokio::test]
async fn oai_stream_broker_error_fails_closed() {
// Unconfigured static provider → NotConfigured, no upstream contact.
let broker: Arc<dyn CredentialBroker> = Arc::new(LocalBroker::new(reqwest::Client::new()));
let cfg = ProviderConfig {
base_url: String::new(),
model: "m".to_string(),
provider: "definitely-not-a-provider".to_string(),
};
let (tx, _rx) = mpsc::unbounded_channel::<StreamEvent>();
let cancel = tokio_util::sync::CancellationToken::new();
let err = call_oai_stream_inner(
&cfg,
&broker,
&[],
&None,
&[],
&tx,
None,
None,
0,
agent_core::reasoning::ReasoningLevel::Medium,
&cancel,
&crate::runtime::trace::TraceContext::disabled(),
true,
false,
0,
)
.await
.unwrap_err()
.to_string();
assert!(err.contains("unknown provider"), "got: {err}");
assert!(!err.to_lowercase().contains("bearer"));
}
/// Phase 1 privacy (spec §5.1): a hostile upstream answers the streaming
/// open with HTTP 500 whose body echoes the full request. The broker's
/// transport error carries a body snippet; the runtime must not surface
/// or log it — status + provider label only.
#[tokio::test]
async fn oai_stream_upstream_error_body_never_surfaces() {
use axum::response::IntoResponse;
const SENTINEL: &str = "PH1-OAI-BROKER-SENTINEL-91d2-RAW";
let app = axum::Router::new().route(
"/chat/completions",
axum::routing::post(move |body: String| async move {
(
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
[("content-type", "application/json")],
format!("{{\"error\":{{\"message\":\"ECHOED:{body}\"}}}}"),
)
.into_response()
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let broker: Arc<dyn CredentialBroker> = Arc::new(LocalBroker::with_local_base_url(
reqwest::Client::new(),
format!("http://{addr}"),
));
let cfg = ProviderConfig {
base_url: "unused-broker-derives-the-url".to_string(),
model: "test-model".to_string(),
provider: "local".to_string(),
};
let tools = vec![serde_json::json!({
"name": "ph1_secret_tool_zz",
"description": "internal tool schema",
"input_schema": {"type": "object", "properties": {}}
})];
let msgs: Vec<crate::SharedMessage> = vec![Arc::new(
serde_json::json!({"role": "user", "content": SENTINEL}),
)];
let (tx, _rx) = mpsc::unbounded_channel::<StreamEvent>();
let cancel = tokio_util::sync::CancellationToken::new();
let err = call_oai_stream_inner(
&cfg,
&broker,
&tools,
&Some(format!("system secret {SENTINEL}")),
&msgs,
&tx,
None,
None,
0,
agent_core::reasoning::ReasoningLevel::Medium,
&cancel,
&crate::runtime::trace::TraceContext::disabled(),
true,
false,
0,
)
.await
.expect_err("500 must fail")
.to_string();
assert!(err.contains("500"), "status must survive: {err}");
for banned in ["ECHOED", SENTINEL, "ph1_secret_tool_zz", "system secret"] {
assert!(
!err.contains(banned),
"provider body content `{banned}` leaked into the surfaced error: {err}"
);
}
}
}
/// Provider display label for the xai-auth Responses route.
const XAI_PROVIDER_LABEL: &str = "xAI";
/// Broker identity + display label for one OpenAI-Responses-wire provider
/// routed through the shared streaming loop
/// ([`stream_broker_responses`]).
#[derive(Debug, Clone, Copy)]
pub(crate) struct ResponsesProfile {
/// Broker provider key (credential + pinned origin + path allowlist).
pub broker_provider: &'static str,
/// Allowlisted broker path for the Responses endpoint.
pub path: &'static str,
/// User-facing provider name in stream errors and logs.
pub label: &'static str,
}
pub(crate) const XAI_RESPONSES: ResponsesProfile = ResponsesProfile {
broker_provider: "xai-auth",
path: "/responses",
label: XAI_PROVIDER_LABEL,
};
/// Meta Muse: `https://api.meta.ai` origin, Responses at `/v1/responses`.
pub(crate) const MUSE_RESPONSES: ResponsesProfile = ResponsesProfile {
broker_provider: "muse",
path: "/v1/responses",
label: "Meta Muse",
};
#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_xai_responses_stream_inner(
cfg: &ProviderConfig,
broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
tools_schema: &[Value],
system_prompt: &Option<String>,
messages: &[crate::SharedMessage],
tx: &mpsc::UnboundedSender<StreamEvent>,
max_tokens: Option<u32>,
reasoning_level: agent_core::reasoning::ReasoningLevel,
cancel: &tokio_util::sync::CancellationToken,
max_retries: u32,
trace: &crate::runtime::trace::TraceContext,
exact_wire_bytes: bool,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
let (tools, names) = translate::tools_to_responses(tools_schema);
let input = codex_input_messages(translate::messages_to_oai(messages, system_prompt, &names));
// Pure, validated body construction — rejects unsupported reasoning
// combinations BEFORE any broker credential access or network I/O.
let body = build_xai_body(
&cfg.model,
reasoning_level,
serde_json::to_value(input)?,
tools,
max_tokens,
)
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })?;
stream_broker_responses(
XAI_RESPONSES,
cfg,
broker,
body,
&names,
tools_schema,
system_prompt,
messages,
tx,
cancel,
max_retries,
trace,
exact_wire_bytes,
)
.await
}
/// Dispatch a resolved `OpenAiResponses` route to its provider body builder:
/// `muse` → [`call_muse_responses_stream_inner`], otherwise xAI.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_responses_route_inner(
route_provider: &str,
cfg: &ProviderConfig,
broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
tools_schema: &[Value],
system_prompt: &Option<String>,
messages: &[crate::SharedMessage],
tx: &mpsc::UnboundedSender<StreamEvent>,
max_tokens: Option<u32>,
reasoning_level: agent_core::reasoning::ReasoningLevel,
cancel: &tokio_util::sync::CancellationToken,
max_retries: u32,
trace: &crate::runtime::trace::TraceContext,
exact_wire_bytes: bool,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
if route_provider == MUSE_RESPONSES.broker_provider {
call_muse_responses_stream_inner(
cfg,
broker,
tools_schema,
system_prompt,
messages,
tx,
max_tokens,
reasoning_level,
cancel,
max_retries,
trace,
exact_wire_bytes,
)
.await
} else {
call_xai_responses_stream_inner(
cfg,
broker,
tools_schema,
system_prompt,
messages,
tx,
max_tokens,
reasoning_level,
cancel,
max_retries,
trace,
exact_wire_bytes,
)
.await
}
}
/// Meta Muse Responses route (`muse/<id>`). Mirrors the official Muse Code
/// request shape: system prompt as `instructions`, conversation as typed
/// input items, `store:false` (stateless), and a `prompt_cache_key` so the
/// server-side prompt cache routes repeated prefixes to the same cache.
#[allow(clippy::too_many_arguments)]
pub(crate) async fn call_muse_responses_stream_inner(
cfg: &ProviderConfig,
broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
tools_schema: &[Value],
system_prompt: &Option<String>,
messages: &[crate::SharedMessage],
tx: &mpsc::UnboundedSender<StreamEvent>,
max_tokens: Option<u32>,
reasoning_level: agent_core::reasoning::ReasoningLevel,
cancel: &tokio_util::sync::CancellationToken,
max_retries: u32,
trace: &crate::runtime::trace::TraceContext,
exact_wire_bytes: bool,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
let (tools, names) = translate::tools_to_responses(tools_schema);
let instructions = system_prompt.clone().unwrap_or_default();
// The system prompt travels as `instructions`, never duplicated in input.
let input = muse_input_messages(translate::messages_to_oai(messages, &None, &names));
let cache_key = muse_prompt_cache_key(&instructions, input.first());
// Validated BEFORE any broker credential access or network I/O.
let body = build_muse_body(
&cfg.model,
reasoning_level,
Value::Array(input),
&instructions,
tools,
max_tokens,
&cache_key,
)
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { e.into() })?;
stream_broker_responses(
MUSE_RESPONSES,
cfg,
broker,
body,
&names,
tools_schema,
system_prompt,
messages,
tx,
cancel,
max_retries,
trace,
exact_wire_bytes,
)
.await
}
/// Deterministic Muse cache-routing key derived from the stable prompt head
/// (instructions + first input item), like the Codex key: identical heads
/// share the prefix the server caches, and a /compact rewrite changes the
/// key together with the prefix it invalidates. Namespaced so it can never
/// collide with Codex keys.
pub(crate) fn muse_prompt_cache_key(instructions: &str, first_item: Option<&Value>) -> String {
codex_prompt_cache_key(instructions, first_item).replacen("synaps-", "synaps-muse-", 1)
}
/// Pure, validated body construction for the Meta Muse Responses API.
///
/// Shape mirrors the official Muse Code client (verified by capture):
/// `model`, `instructions`, `input`, `store:false`, `stream:true`, `tools`,
/// `max_output_tokens`, `reasoning.effort`, `prompt_cache_key`. Reasoning is
/// validated against the exact model's capability first; `Adaptive` omits
/// the field (provider default). Encrypted reasoning is not requested:
/// Synaps does not replay reasoning items across turns.
#[allow(clippy::too_many_arguments)] // pure builder mirroring the wire body
pub(crate) fn build_muse_body(
model: &str,
level: agent_core::reasoning::ReasoningLevel,
input: Value,
instructions: &str,
tools: Vec<Value>,
max_tokens: Option<u32>,
prompt_cache_key: &str,
) -> Result<Value, String> {
crate::runtime::openai::catalog::validation::validate_reasoning_mutation(
&format!("muse/{model}"),
level,
)?;
let mut body = serde_json::Map::new();
body.insert("model".into(), json!(model));
if !instructions.is_empty() {
body.insert("instructions".into(), json!(instructions));
}
body.insert("input".into(), input);
body.insert("store".into(), json!(false));
body.insert("stream".into(), json!(true));
if !tools.is_empty() {
body.insert("tools".into(), Value::Array(tools));
}
if let Some(max) = max_tokens {
body.insert("max_output_tokens".into(), json!(max));
}
if let Some(effort) = crate::runtime::openai::catalog::muse_effort_label(level) {
body.insert("reasoning".into(), json!({ "effort": effort }));
}
body.insert("prompt_cache_key".into(), json!(prompt_cache_key));
Ok(Value::Object(body))
}
/// Shared streaming loop for broker-proxied OpenAI Responses providers
/// (xAI, Meta Muse): one exact-bytes request, SSE decode, bounded retries
/// that never replay after output reached the UI.
#[allow(clippy::too_many_arguments)]
async fn stream_broker_responses(
profile: ResponsesProfile,
cfg: &ProviderConfig,
broker: &std::sync::Arc<dyn crate::auth::CredentialBroker>,
body: Value,
names: &translate::ToolNameMap,
tools_schema: &[Value],
system_prompt: &Option<String>,
messages: &[crate::SharedMessage],
tx: &mpsc::UnboundedSender<StreamEvent>,
cancel: &tokio_util::sync::CancellationToken,
max_retries: u32,
trace: &crate::runtime::trace::TraceContext,
exact_wire_bytes: bool,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
// Serialize ONCE via the sanctioned constructor: the digested bytes are
// the very bytes the broker sends upstream (`body_bytes` handoff,
// LocalBroker verbatim) — `body`/`body_bytes` cannot diverge. On a
// remote broker the daemon re-serializes out of process → honest
// `CloudProxy` kind and no wire-byte claim. Every retry clones this
// same request (identical `body` and `body_bytes`), so one wire digest
// is honest for all attempt records.
let (proxy_request, body_bytes) = crate::auth::ProxyRequest::post_json_exact(
profile.broker_provider,
profile.path,
body,
true,
)?;
let tracer = tr::begin_openai_tracer(
trace,
&cfg.provider,
&cfg.model,
if exact_wire_bytes {
crate::runtime::trace::TransportKind::OpenAiResponses
} else {
crate::runtime::trace::TransportKind::CloudProxy
},
&format!("{}/responses", cfg.base_url.trim_end_matches('/')),
exact_wire_bytes.then_some(body_bytes.as_ref()),
messages,
system_prompt.as_deref(),
tools_schema,
tr::renamed_tool_losses(names),
)
.await;
// One-shot explicit content capture (`/trace next content`): a no-op in
// every context without the arm. `body_bytes` is the serialized request
// body this process built pre-send (exact wire bytes on the local
// broker; the same body a remote broker re-serializes) — body only,
// headers and credentials structurally never reach this seam.
if let Some(tracer) = &tracer {
tr::capture_request_content(trace, tracer.request_id(), body_bytes.as_ref(), messages);
}
let mut attempt = tr::StreamAttempt::new(tracer);
let mut stream_retry = 0u32;
loop {
if cancel.is_cancelled() {
return Err("request canceled".into());
}
let stream = crate::runtime::api::await_or_cancel(
cancel,
broker.proxy_stream(proxy_request.clone()),
)
.await
.map_err(|_| "request canceled")?;
// A credential refresh failure keeps its kind as a type (retryable /
// action-required vs credential); its text is ours, no upstream bytes.
let stream = match stream {
Err(crate::auth::BrokerError::Refresh(f)) => {
attempt.finish_failed("broker_error", None, None);
return Err(Box::new(super::net::BrokerRefreshError::new(f)));
}
// Privacy (spec §5.1): a broker proxy error may carry an upstream
// response-body snippet; redact it to status-only before surfacing.
other => other.map_err(|e| super::net::redact_provider_proxy_error(&e.to_string())),
};
let mut stream = match stream {
Ok(stream) => {
attempt.mark_headers();
stream
}
Err(msg) => {
let status = tr::broker_error_status(&msg);
let code =
status.map_or_else(|| "broker_error".to_string(), |s| format!("http_{s}"));
attempt.finish_failed(&code, status, None);
return Err(super::net::normalize_broker_http_error(msg).into());
}
};
let _ = tx.send(StreamEvent::Llm(crate::LlmEvent::ResponseStart));
let mut interrupted = false;
let mut text = String::new();
let mut parser = CodexSseDecoder::for_provider(profile.label);
parser.backfill_message_text = profile.broker_provider == MUSE_RESPONSES.broker_provider;
let mut buf = bytes::BytesMut::new();
while let Some(chunk) = tokio::select! {
biased;
_ = cancel.cancelled() => {
attempt.finish_canceled(None, parser.trace_usage());
return Err("request canceled".into());
}
c = stream.next() => c,
} {
let chunk = match chunk {
Ok(chunk) => {
attempt.mark_first_byte();
chunk
}
Err(e) => {
if !matches!(e, crate::auth::BrokerError::Transport(_)) {
attempt.finish_failed("broker_error", None, None);
return Err(e.into());
}
interrupted = !parser.terminal_success && parser.terminal_failure.is_none();
break;
}
};
buf.extend_from_slice(&chunk);
while let Some(n) = memchr::memchr(b'\n', &buf) {
let line = buf.split_to(n + 1);
parser.push_line(std::str::from_utf8(&line[..n]).unwrap_or(""), tx, &mut text);
}
if parser.saw_model_event {
attempt.mark_first_model_event();
}
}
if !interrupted && !buf.is_empty() {
parser.push_line(std::str::from_utf8(&buf).unwrap_or(""), tx, &mut text);
}
if interrupted || (!parser.terminal_success && parser.terminal_failure.is_none()) {
if stream_retry >= max_retries {
attempt.finish_failed("stream_error", None, None);
return Err(STREAM_INTERRUPTED.into());
}
stream_retry += 1;
attempt.attempt_failed(
crate::runtime::trace::RetryClass::Network,
retry_delay(stream_retry),
None,
None,
"stream_error",
);
wait_stream_retry(tx, cancel, stream_retry, max_retries, false)
.await
.inspect_err(|_| {
attempt.finish_canceled(None, None);
})?;
attempt.restart_clock();
continue;
}
parser.finish();
// Trailing-buffer flush and finish() can surface the first (or only)
// model event; mark it so first_model_event_ms is honest for tail-only
// streams (mirrors the Codex direct path above).
if parser.saw_model_event {
attempt.mark_first_model_event();
}
// Broker paths never observe the upstream HTTP status; the Responses
// stream does not yet expose a normalized stop reason — both stay
// honest `None` rather than guessed values.
if let Err(failure) = parser.terminal_result() {
// xAI answers HTTP 200 + a lone `{"type":"error",…}` capacity
// event while the model is saturated; a re-send seconds later
// usually succeeds. Retry only when NOTHING reached the UI on
// this attempt — text or tool events already streamed would be
// emitted twice on a re-send, so those failures stay terminal.
let output_already_streamed = !text.is_empty() || parser.emitted_any_output();
if failure.is_retryable() && !output_already_streamed && stream_retry < max_retries {
stream_retry += 1;
let delay = retry_delay(stream_retry);
attempt.attempt_failed(
crate::runtime::trace::RetryClass::Other,
delay,
None,
None,
failure.code,
);
tracing::warn!(
code = failure.code,
provider = profile.broker_provider,
"responses stream retry {stream_retry}/{max_retries} after {delay:?}"
);
tokio::select! {
_ = tokio::time::sleep(delay) => {}
_ = cancel.cancelled() => {
attempt.finish_canceled(None, None);
return Err("request canceled".into());
}
}
attempt.restart_clock();
continue;
}
attempt.finish_failed(failure.code, None, None);
return Err(failure.message.into());
}
attempt.finish_success(None, None, None, parser.trace_usage());
let mut content = Vec::new();
if !text.is_empty() {
content.push(json!({"type":"text","text":text}));
}
content.extend(translate::tool_calls_to_content_blocks(
&parser.completed_tools,
names,
));
let mut response = json!({"role":"assistant","content":content});
if profile.broker_provider == MUSE_RESPONSES.broker_provider {
if let Some(reason) = muse_dangling_preamble(
!parser.completed_tools.is_empty(),
parser.last_message_phase.as_deref(),
parser.last_message_text().unwrap_or(&text),
) {
tracing::info!(
provider = profile.broker_provider,
reason,
"muse response ended on a preamble without acting"
);
// Top-level sidecar, never inside `content`: the turn loop
// persists only `content`, so nothing reaches stored history.
response[DANGLING_PREAMBLE_KEY] = json!(reason);
}
}
return Ok(response);
}
}
/// Response sidecar key (top level, outside `content`) set by the Muse route
/// when a completed response ended on a preamble without acting. Read by the
/// turn loop's one-shot continuation; never persisted.
pub(crate) const DANGLING_PREAMBLE_KEY: &str = "synaps_dangling_preamble";
/// Upper bound (chars, trimmed) for the text heuristic. The observed failures
/// were 32–36 output tokens (~70–130 chars); 300 leaves room for a long
/// sentence of preamble while excluding real answers that merely end in a
/// colon after substantive content.
pub(crate) const PREAMBLE_MAX_CHARS: usize = 300;
/// Decide whether a completed Muse response ended on an unfinished preamble.
/// Returns the trigger reason, or `None`. Never fires when the response
/// called a tool.
///
/// - `commentary_phase`: the last message item was labelled
/// `phase:"commentary"` by Muse itself, yet no tool call followed — by the
/// provider's own contract the turn is unfinished.
/// - `colon_preamble`: the last message's trimmed text is short
/// (≤ [`PREAMBLE_MAX_CHARS`]) and ends with ':' — "Blocking question
/// before I build anything:" with nothing after it.
pub(crate) fn muse_dangling_preamble(
has_tool_calls: bool,
last_message_phase: Option<&str>,
last_message_text: &str,
) -> Option<&'static str> {
if has_tool_calls {
return None;
}
if last_message_phase == Some("commentary") {
return Some("commentary_phase");
}
let text = last_message_text.trim();
if (text.ends_with(':') || text.ends_with('\u{ff1a}'))
&& text.chars().count() <= PREAMBLE_MAX_CHARS
{
return Some("colon_preamble");
}
None
}
/// Harness-authored nudge appended (once per user turn) after a dangling
/// preamble. Framed like the context advisories so the model does not read
/// it as a new user request.
pub(crate) const PREAMBLE_NUDGE: &str = "[Host harness note, not a new user request] You ended your turn on a preamble without acting. Continue now: run the tool call you announced, or ask your question in full.";
/// Visible system note shown when the nudge fires.
pub(crate) const PREAMBLE_NUDGE_NOTICE: &str = "Muse ended its turn on a preamble without acting; continuing once automatically (disable: SYNAPS_MUSE_PREAMBLE_NUDGE=0).";
/// `SYNAPS_MUSE_PREAMBLE_NUDGE=0|false|off|no` disables the one-shot
/// continuation. Enabled by default.
pub(crate) fn preamble_nudge_enabled() -> bool {
preamble_nudge_enabled_value(std::env::var("SYNAPS_MUSE_PREAMBLE_NUDGE").ok().as_deref())
}
fn preamble_nudge_enabled_value(value: Option<&str>) -> bool {
!matches!(
value.map(|v| v.trim().to_ascii_lowercase()).as_deref(),
Some("0" | "false" | "off" | "no")
)
}
/// Turn-loop gate for the one-shot preamble continuation. Returns the reason
/// and arms `already_nudged` when the response carries the Muse sidecar, the
/// round was not cancelled, the feature is enabled, and this user turn has
/// not been nudged yet. At most one nudge per user turn: it cannot loop.
pub(crate) fn take_preamble_nudge(
response: &Value,
already_nudged: &mut bool,
cancelled: bool,
enabled: bool,
) -> Option<String> {
if *already_nudged || cancelled || !enabled {
return None;
}
let reason = response.get(DANGLING_PREAMBLE_KEY)?.as_str()?.to_string();
*already_nudged = true;
Some(reason)
}
/// Pure, validated body construction for the xAI Responses API.
///
/// Enforces the exact-model reasoning matrix from
/// `docs/specs/anthropic-xai-reasoning-modes-spec.md` BEFORE any credential or
/// network access, via the shared per-provider validator:
/// - `Off` is rejected (never silently omitted) on models whose reasoning
/// cannot be disabled; on the non-reasoning model it is trivially omission.
/// - `Adaptive` omits the `reasoning` field → documented provider default.
/// - Named levels are emitted as exact `reasoning:{effort:"..."}` only when
/// the exact model id documents them; otherwise `Err` (fail closed).
pub(crate) fn build_xai_body(
model: &str,
level: agent_core::reasoning::ReasoningLevel,
input: Value,
tools: Vec<Value>,
max_tokens: Option<u32>,
) -> Result<Value, String> {
use agent_core::reasoning::ReasoningLevel;
crate::runtime::openai::catalog::validation::validate_reasoning_mutation(
&format!("xai-auth/{model}"),
level,
)?;
let mut body = serde_json::Map::new();
body.insert("model".into(), json!(model));
body.insert("input".into(), input);
body.insert("stream".into(), json!(true));
if !tools.is_empty() {
body.insert("tools".into(), Value::Array(tools));
}
if let Some(max) = max_tokens {
body.insert("max_output_tokens".into(), json!(max));
}
match level {
// Off (validated: only reachable where reasoning is absent/disableable)
// and Adaptive: omit `reasoning` — provider default applies.
ReasoningLevel::Off | ReasoningLevel::Adaptive => {}
l => {
body.insert("reasoning".into(), json!({ "effort": l.as_str() }));
}
}
Ok(Value::Object(body))
}
#[cfg(test)]
mod xai_tests {
use super::*;
use agent_core::reasoning::ReasoningLevel;
#[test]
fn xai_fixture_is_public_responses_shape() {
let fixture = serde_json::json!({"type":"response.output_text.delta","delta":"hello"});
assert_eq!(fixture["type"], "response.output_text.delta");
assert_eq!(fixture["delta"], "hello");
}
fn body_for(model: &str, level: ReasoningLevel) -> Result<Value, String> {
build_xai_body(model, level, json!([]), Vec::new(), Some(1024))
}
// ── Exact Responses wire (spec: anthropic-xai-reasoning-modes) ───────────
#[test]
fn grok45_emits_exact_documented_efforts() {
for (level, effort) in [
(ReasoningLevel::Low, "low"),
(ReasoningLevel::Medium, "medium"),
(ReasoningLevel::High, "high"),
] {
let body = body_for("grok-4.5", level).expect("supported effort");
assert_eq!(body["reasoning"], json!({"effort": effort}), "{level}");
assert_eq!(body["model"], "grok-4.5");
assert_eq!(body["stream"], json!(true));
assert_eq!(body["max_output_tokens"], json!(1024));
}
}
#[test]
fn grok47_responses_wire_uses_exact_efforts_and_preserves_tool_shape() {
for (level, effort) in [
(ReasoningLevel::Low, "low"),
(ReasoningLevel::Medium, "medium"),
(ReasoningLevel::High, "high"),
(ReasoningLevel::XHigh, "xhigh"),
] {
let input = json!([{"role": "user", "content": "hello"}]);
let tool = json!({"type": "function", "name": "inspect", "parameters": {"type": "object"}});
let body = build_xai_body("grok-4.7", level, input.clone(), vec![tool.clone()], Some(1024)).unwrap();
assert_eq!(body["model"], "grok-4.7");
assert_eq!(body["reasoning"], json!({"effort": effort}));
assert_eq!(body["input"], input);
assert_eq!(body["tools"], json!([tool]));
assert_eq!(body["stream"], true);
assert_eq!(body["max_output_tokens"], 1024);
assert!(body.get("messages").is_none());
assert!(body.get("reasoning_effort").is_none());
// Default ciphertext does not require opting into include or
// changing storage policy (official docs say ignoring it works).
assert!(body.get("include").is_none());
assert!(body.get("store").is_none());
}
assert!(body_for("grok-4.7", ReasoningLevel::Adaptive).unwrap().get("reasoning").is_none());
for level in [ReasoningLevel::Off, ReasoningLevel::Max, ReasoningLevel::Ultra, ReasoningLevel::UltraCode] {
assert!(body_for("grok-4.7", level).is_err(), "{level}");
}
}
#[test]
fn grok46_emits_exact_documented_efforts() {
for (level, effort) in [
(ReasoningLevel::Low, "low"),
(ReasoningLevel::Medium, "medium"),
(ReasoningLevel::High, "high"),
] {
let body = body_for("grok-4.6", level).expect("supported effort");
assert_eq!(body["reasoning"], json!({"effort": effort}), "{level}");
assert_eq!(body["model"], "grok-4.6");
assert_eq!(body["stream"], json!(true));
assert_eq!(body["max_output_tokens"], json!(1024));
}
}
#[test]
fn adaptive_omits_reasoning_field_provider_default() {
for model in [
"grok-4.5",
"grok-4.5-latest",
"grok-4.6",
"grok-4.3",
"grok-4.20-0309-non-reasoning",
] {
let body = body_for(model, ReasoningLevel::Adaptive).expect("adaptive is omission");
assert!(body.get("reasoning").is_none(), "{model}");
}
}
#[test]
fn off_is_rejected_pre_network_when_reasoning_cannot_be_disabled() {
for model in [
"grok-4.5",
"grok-4.5-latest",
"grok-4.6",
"grok-4.20-multi-agent-0309",
"grok-4.3",
] {
let err = body_for(model, ReasoningLevel::Off).unwrap_err();
assert!(err.contains(model), "{model}: {err}");
}
// Non-reasoning model: Off is trivially satisfied by omission.
let body = body_for("grok-4.20-0309-non-reasoning", ReasoningLevel::Off).unwrap();
assert!(body.get("reasoning").is_none());
}
#[test]
fn unsupported_named_efforts_are_rejected_pre_network() {
// 4.5/4.6 have no documented xhigh; max/ultra never exist on xAI.
for level in [
ReasoningLevel::XHigh,
ReasoningLevel::Max,
ReasoningLevel::Ultra,
] {
assert!(body_for("grok-4.5", level).is_err(), "{level}");
assert!(body_for("grok-4.6", level).is_err(), "{level}");
}
// Intrinsic-reasoning models have no documented effort control.
assert!(body_for("grok-4.3", ReasoningLevel::Medium).is_err());
// Non-reasoning models reject named reasoning outright.
assert!(body_for("grok-4.20-0309-non-reasoning", ReasoningLevel::Medium).is_err());
// Unknown exact ids fail closed.
assert!(body_for("grok-9000", ReasoningLevel::Medium).is_err());
}
#[test]
fn multi_agent_xhigh_is_exact_agent_count_control() {
let body = body_for("grok-4.20-multi-agent-0309", ReasoningLevel::XHigh).unwrap();
assert_eq!(body["reasoning"], json!({"effort": "xhigh"}));
}
#[test]
fn responses_wire_shape_tools_flat_and_no_chat_completions_fields() {
let tools = vec![json!({
"type": "function",
"name": "get_weather",
"description": "d",
"parameters": {"type": "object"}
})];
let body =
build_xai_body("grok-4.5", ReasoningLevel::High, json!([]), tools, None).unwrap();
// Responses API: flat tool objects + `input`, never chat-completions
// `messages`/nested function wrappers.
assert_eq!(body["tools"][0]["name"], "get_weather");
assert!(body["tools"][0].get("function").is_none());
assert!(body.get("input").is_some());
assert!(body.get("messages").is_none());
assert!(body.get("max_output_tokens").is_none());
}
}
#[cfg(test)]
mod xai_capacity_retry_tests {
//! Regression tests for the xai-auth Responses route: xAI answers HTTP
//! 200 + a lone flat `{"type":"error",…}` capacity event while the model
//! is saturated (reproduced live: grok-4.6 failed 3 of 4 turns, a re-send
//! seconds later succeeded). The route must retry with identical request
//! bytes, label its messages "xAI" (not "Codex"), and never surface the
//! provider prose. A scripted fake broker serves the SSE bodies directly
//! through `proxy_stream` — no HTTP server involved.
use super::*;
use agent_core::auth::{
AccessToken, BrokerError, ProxyByteStream, ProxyRequest, ProxyResponse,
};
use async_trait::async_trait;
use std::sync::{
atomic::{AtomicUsize, Ordering},
Arc,
};
/// The exact envelope observed on the wire (flat: `type` is the event
/// type, `code` null, `message` top-level, no nested `error` object).
const XAI_CAPACITY_SSE: &str = "data: {\"sequence_number\":0,\"type\":\"error\",\"code\":null,\"message\":\"The model is currently at capacity due to high demand. Please try again in a few minutes, or use a higher service tier for priority processing: https://docs.x.ai/developers/advanced-api-usage/priority-processing\",\"param\":null}\n\n";
const XAI_SUCCESS_SSE: &str = concat!(
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\"}}\n\n",
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"GROK_OK\"}\n\n",
"data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":7,\"output_tokens\":2}}}\n\n",
);
/// Serves `bodies[n]` for the n-th `proxy_stream` call (the last body
/// repeats once exhausted) and records every request body it saw.
struct ScriptedXaiBroker {
bodies: Vec<&'static str>,
calls: Arc<AtomicUsize>,
seen: Arc<std::sync::Mutex<Vec<ProxyRequest>>>,
}
#[async_trait]
impl crate::auth::CredentialBroker for ScriptedXaiBroker {
async fn access_token(
&self,
_p: agent_core::auth::OAuthProviderId,
) -> Result<AccessToken, BrokerError> {
Err(BrokerError::Denied(
"xai route must use proxy_stream".into(),
))
}
async fn proxy(&self, _request: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
Err(BrokerError::Denied("not implemented in stub".into()))
}
async fn proxy_stream(
&self,
request: ProxyRequest,
) -> Result<ProxyByteStream, BrokerError> {
request.validate()?;
let n = self.calls.fetch_add(1, Ordering::SeqCst);
self.seen.lock().unwrap().push(request);
let body = self.bodies[n.min(self.bodies.len() - 1)];
Ok(Box::pin(futures::stream::iter(vec![Ok(
bytes::Bytes::from(body),
)])))
}
async fn anthropic_usage(&self) -> Result<serde_json::Value, BrokerError> {
Err(BrokerError::Denied("not implemented in stub".into()))
}
async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
Ok(vec![])
}
}
/// Broker whose `proxy_stream` fails with a typed refresh failure (a
/// Muse re-mint that hit the network, a corrupt store).
struct RefreshFailingBroker(agent_core::auth::RefreshFailureKind);
#[async_trait]
impl crate::auth::CredentialBroker for RefreshFailingBroker {
async fn access_token(
&self,
_p: agent_core::auth::OAuthProviderId,
) -> Result<AccessToken, BrokerError> {
Err(BrokerError::Denied("stub".into()))
}
async fn proxy(&self, _request: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
Err(BrokerError::Denied("stub".into()))
}
async fn proxy_stream(&self, _r: ProxyRequest) -> Result<ProxyByteStream, BrokerError> {
Err(BrokerError::Refresh(agent_core::auth::RefreshFailure::new(
self.0,
"transport error",
)))
}
async fn anthropic_usage(&self) -> Result<serde_json::Value, BrokerError> {
Err(BrokerError::Denied("stub".into()))
}
async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
Ok(vec![])
}
}
/// Round-2 m-a: the retryable kind is carried as a type through the real
/// pipeline (redaction, normalization, classification), not recovered
/// from our hint text. Retryable and action-required refresh failures
/// are API-side, never "Config error" and never "sign in again".
#[tokio::test]
async fn broker_refresh_failure_kind_survives_the_stream_pipeline() {
use agent_core::auth::RefreshFailureKind as K;
for (kind, api) in [
(K::Transient, true),
(K::LockBusy, true),
(K::Other, true),
(K::ActionRequired, true),
(K::Credential, false),
] {
let broker: Arc<dyn crate::auth::CredentialBroker> =
Arc::new(RefreshFailingBroker(kind));
let cfg = ProviderConfig {
base_url: "https://api.meta.ai/v1".to_string(),
model: "muse-x".to_string(),
provider: "muse".to_string(),
};
let (tx, _rx) = mpsc::unbounded_channel::<StreamEvent>();
let err = call_xai_responses_stream_inner(
&cfg,
&broker,
&[],
&None,
&[],
&tx,
None,
agent_core::reasoning::ReasoningLevel::Adaptive,
&tokio_util::sync::CancellationToken::new(),
0,
&crate::runtime::trace::TraceContext::disabled(),
true,
)
.await
.unwrap_err();
let runtime = super::super::net::provider_error_to_runtime_for("muse/muse-x", err);
match (&runtime, api) {
(crate::RuntimeError::ApiStatus(m), true) => {
assert!(m.contains("credential refresh failed"), "{kind:?}: {m}");
assert!(!m.contains("Sign in again"), "{kind:?}: {m}");
}
(crate::RuntimeError::Config(_), false) => {}
other => panic!("{kind:?}: {other:?}"),
}
}
}
struct Run {
result: Result<Value, Box<dyn std::error::Error + Send + Sync>>,
calls: usize,
seen: Vec<ProxyRequest>,
events: Vec<StreamEvent>,
}
async fn run_xai(bodies: Vec<&'static str>, max_retries: u32) -> Run {
let calls = Arc::new(AtomicUsize::new(0));
let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
let broker: Arc<dyn crate::auth::CredentialBroker> = Arc::new(ScriptedXaiBroker {
bodies,
calls: Arc::clone(&calls),
seen: Arc::clone(&seen),
});
let cfg = ProviderConfig {
base_url: "https://api.x.ai/v1".to_string(),
model: "grok-4.5".to_string(),
provider: "xai-auth".to_string(),
};
let (tx, mut rx) = mpsc::unbounded_channel::<StreamEvent>();
let result = call_xai_responses_stream_inner(
&cfg,
&broker,
&[],
&Some("system".to_string()),
&[],
&tx,
None,
agent_core::reasoning::ReasoningLevel::Adaptive,
&tokio_util::sync::CancellationToken::new(),
max_retries,
&crate::runtime::trace::TraceContext::disabled(),
true,
)
.await;
drop(tx);
let mut events = Vec::new();
while let Ok(event) = rx.try_recv() {
events.push(event);
}
let seen = std::mem::take(&mut *seen.lock().unwrap());
Run {
result,
calls: calls.load(Ordering::SeqCst),
seen,
events,
}
}
fn text_events(events: &[StreamEvent]) -> Vec<String> {
events
.iter()
.filter_map(|e| match e {
StreamEvent::Llm(crate::runtime::types::LlmEvent::Text(t)) => Some(t.clone()),
_ => None,
})
.collect()
}
#[tokio::test(start_paused = true)]
async fn xai_capacity_error_is_retried_with_identical_bytes_then_succeeds() {
let run = run_xai(vec![XAI_CAPACITY_SSE, XAI_SUCCESS_SSE], 3).await;
let value = run.result.expect("second attempt must succeed");
assert_eq!(value["content"][0]["text"], "GROK_OK");
assert_eq!(run.calls, 2, "exactly one retry");
assert_eq!(run.seen.len(), 2);
assert_eq!(
run.seen[0].body_bytes, run.seen[1].body_bytes,
"retry must re-send the identical serialized body"
);
assert_eq!(run.seen[0].body, run.seen[1].body);
assert_eq!(run.seen[0].provider, "xai-auth");
assert_eq!(run.seen[0].path, "/responses");
// No output was streamed twice: the failed attempt emitted nothing.
assert_eq!(text_events(&run.events), vec!["GROK_OK".to_string()]);
}
#[tokio::test(start_paused = true)]
async fn xai_capacity_error_with_zero_retries_surfaces_xai_capacity_message() {
let run = run_xai(vec![XAI_CAPACITY_SSE], 0).await;
let err = run.result.expect_err("no retry budget → terminal");
let msg = err.to_string();
assert_eq!(
msg,
"xAI reports the model is at capacity (retries exhausted). Try again in a few minutes or switch models with /model."
);
assert_eq!(run.calls, 1);
for banned in ["Codex", "high demand", "docs.x.ai", "service tier"] {
assert!(!msg.contains(banned), "`{banned}` leaked: {msg}");
}
// And the runtime classifies it as an API failure, not a config one.
assert!(matches!(
super::super::net::provider_error_to_runtime(err),
crate::RuntimeError::ApiStatus(_)
));
}
#[tokio::test(start_paused = true)]
async fn xai_capacity_error_exhausting_budget_counts_every_attempt() {
let run = run_xai(vec![XAI_CAPACITY_SSE], 2).await;
let err = run.result.expect_err("persistent capacity → terminal");
assert!(err
.to_string()
.starts_with("xAI reports the model is at capacity"));
assert_eq!(run.calls, 3, "initial attempt + 2 retries");
assert!(text_events(&run.events).is_empty());
}
/// If the failed attempt already streamed text, a re-send would duplicate
/// it in the UI — the failure must stay terminal even with budget left.
#[tokio::test(start_paused = true)]
async fn xai_does_not_retry_after_partial_text_was_streamed() {
const PARTIAL_THEN_ERROR: &str = concat!(
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n",
"data: {\"type\":\"error\",\"code\":null,\"message\":\"at capacity\"}\n\n",
);
let run = run_xai(vec![PARTIAL_THEN_ERROR, XAI_SUCCESS_SSE], 3).await;
let err = run.result.expect_err("must not retry after partial output");
assert!(err
.to_string()
.starts_with("xAI reports the model is at capacity"));
assert_eq!(run.calls, 1, "no retry once output reached the UI");
assert_eq!(text_events(&run.events), vec!["partial".to_string()]);
}
/// An unfinished tool preview is not execution; reset it before retrying.
#[tokio::test(start_paused = true)]
async fn xai_retries_eof_after_tool_preview_with_reset() {
const TOOL_THEN_EOF: &str = "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"bash\"}}\n\n";
let run = run_xai(vec![TOOL_THEN_EOF, XAI_SUCCESS_SSE], 3).await;
assert_eq!(run.result.unwrap()["content"][0]["text"], "GROK_OK");
assert_eq!(run.calls, 2);
assert_eq!(run.seen[0].body_bytes, run.seen[1].body_bytes);
assert!(run
.events
.iter()
.any(|e| matches!(e, StreamEvent::Llm(crate::LlmEvent::ResponseReset))));
}
#[test]
fn responses_exact_account_errors_survive_decoder_net_and_driver() {
for (code, suffix, expected) in [
("invalid_api_key", RESPONSES_AUTH_SUFFIX, "auth"),
("authentication_error", RESPONSES_AUTH_SUFFIX, "auth"),
(
"invalid_authentication_error",
RESPONSES_AUTH_SUFFIX,
"auth",
),
("insufficient_quota", RESPONSES_QUOTA_SUFFIX, "quota"),
("quota_exceeded", RESPONSES_QUOTA_SUFFIX, "quota"),
("billing_error", RESPONSES_QUOTA_SUFFIX, "quota"),
] {
for label in ["Codex", "OpenAI", "xAI"] {
let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
let mut decoder = CodexSseDecoder::for_provider(label);
let frame = format!(
"data: {}\n\n",
json!({"type":"response.failed", "response":{
"error":{"code":code,"message":"SECRET at capacity"}
}})
);
let mut text = String::new();
for line in frame.lines() {
decoder.push_line(line, &tx, &mut text);
}
let failure = decoder.terminal_result().unwrap_err();
assert_eq!(failure.message, format!("{label}{suffix}"));
assert!(!failure.is_retryable());
let runtime = super::super::net::provider_error_to_runtime(failure.message.into());
assert!(matches!(runtime, crate::RuntimeError::ApiStatus(_)));
let typed = crate::runtime::helpers::turn_error_for(&runtime, "test");
assert_eq!(
crate::extensions::session_driver::classify_turn_error(&typed),
(
crate::extensions::session_driver::Outcome::ProviderError,
expected.into()
)
);
assert!(!typed.message.contains("SECRET"));
}
}
for code in [
"permission_denied",
"context_length_exceeded",
"invalid_request_error",
"unknown",
] {
let message = user_message_for_terminal_failure("Codex", "absent", code, false);
assert_eq!(
crate::extensions::session_driver::classify_error(&message).0,
crate::extensions::session_driver::Outcome::Blocked
);
}
}
#[test]
fn explicit_policy_and_conflicting_error_classes_outrank_capacity_prose() {
for (kind, code) in [
("absent", "permission_denied"),
("permission_error", "rate_limit_exceeded"),
("invalid_request_error", "insufficient_quota"),
("authentication_error", "quota_exceeded"),
("server_error", "invalid_api_key"),
("unknown", "absent"),
] {
let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
let mut decoder = CodexSseDecoder::for_provider("Codex");
let frame = format!(
"data: {}",
json!({"type":"response.failed", "response":{
"error":{"type":kind, "code":code, "message":"private policy at capacity"}
}})
);
let mut text = String::new();
decoder.push_line(&frame, &tx, &mut text);
decoder.push_line("", &tx, &mut text);
let failure = decoder.terminal_result().unwrap_err();
assert!(!failure.is_retryable());
assert_eq!(
crate::extensions::session_driver::classify_error(&failure.message).0,
crate::extensions::session_driver::Outcome::Blocked
);
assert!(!failure.message.contains("private"));
}
}
/// Deterministic (non-capacity) failures are terminal on the first try
/// and keep the neutral, xAI-labelled, details-withheld message.
#[tokio::test(start_paused = true)]
async fn xai_non_capacity_error_is_terminal_and_labelled_xai() {
const OTHER_ERROR: &str = "data: {\"type\":\"error\",\"code\":\"invalid_api_key\",\"message\":\"ECHOED:secret prompt\"}\n\n";
let run = run_xai(vec![OTHER_ERROR, XAI_SUCCESS_SSE], 3).await;
let err = run.result.expect_err("must fail");
let msg = err.to_string();
assert_eq!(msg, format!("xAI{RESPONSES_AUTH_SUFFIX}"));
assert!(!msg.contains("ECHOED"));
assert_eq!(run.calls, 1);
}
}
// ─── Codex wire body tests ────────────────────────────────────────────────────
#[cfg(test)]
mod codex_wire_tests {
use crate::runtime::openai::catalog::{
plan_codex_execution, CodexMultiAgentMode, CodexRequestRole,
};
use agent_core::reasoning::ReasoningLevel;
fn plan(
level: ReasoningLevel,
role: CodexRequestRole,
) -> crate::runtime::openai::catalog::CodexExecutionPlan {
plan_codex_execution("openai-codex/gpt-5.6-sol", level, role, None)
.expect("Sol request plan")
}
#[allow(clippy::too_many_arguments)]
fn body_for_level(
model: &str,
level: ReasoningLevel,
input: serde_json::Value,
instructions: String,
tools: Vec<serde_json::Value>,
temperature: Option<f32>,
max_tokens: Option<u32>,
) -> serde_json::Value {
let qualified = format!("openai-codex/{model}");
let plan = plan_codex_execution(&qualified, level, CodexRequestRole::Foreground, None)
.expect("request plan");
super::build_codex_body(
model,
&plan,
input,
instructions,
tools,
temperature,
max_tokens,
"synaps-test-cache-key",
)
}
/// Validate → emit exact reasoning.effort shape for known levels.
fn codex_effort_for(model_id: &str, level: ReasoningLevel) -> Result<String, String> {
let qualified = format!("openai-codex/{model_id}");
let plan = plan_codex_execution(&qualified, level, CodexRequestRole::Foreground, None)
.map_err(|error| error.to_string())?;
plan.wire_effort
.map(|effort| effort.as_str().to_string())
.ok_or_else(|| "reasoning effort omitted".to_string())
}
#[test]
fn sol_lowers_ultra_to_max_for_requests() {
let effort = codex_effort_for("gpt-5.6-sol", ReasoningLevel::Ultra).unwrap();
assert_eq!(effort, "max");
}
#[test]
fn sol_emits_max_exact() {
let effort = codex_effort_for("gpt-5.6-sol", ReasoningLevel::Max).unwrap();
assert_eq!(effort, "max");
}
#[test]
fn sol_emits_xhigh_exact() {
let effort = codex_effort_for("gpt-5.6-sol", ReasoningLevel::XHigh).unwrap();
assert_eq!(effort, "xhigh");
}
#[test]
fn luna_emits_max_exact() {
let effort = codex_effort_for("gpt-5.6-luna", ReasoningLevel::Max).unwrap();
assert_eq!(effort, "max");
}
#[test]
fn luna_rejects_ultra_before_network() {
// Validation must fail before any credential/network access.
let err = codex_effort_for("gpt-5.6-luna", ReasoningLevel::Ultra).unwrap_err();
assert!(err.contains("ultra"), "{err}");
assert!(err.contains("gpt-5.6-luna"), "{err}");
}
#[test]
fn gpt55_emits_xhigh_exact() {
let effort = codex_effort_for("gpt-5.5", ReasoningLevel::XHigh).unwrap();
assert_eq!(effort, "xhigh");
}
#[test]
fn gpt55_rejects_max_before_network() {
assert!(codex_effort_for("gpt-5.5", ReasoningLevel::Max).is_err());
assert!(codex_effort_for("gpt-5.5", ReasoningLevel::Ultra).is_err());
}
#[test]
fn off_and_adaptive_omit_reasoning_field() {
// Off/Adaptive: build_codex_body must NOT emit a "reasoning" key.
for level in [ReasoningLevel::Off, ReasoningLevel::Adaptive] {
let body = body_for_level(
"gpt-5.6-sol",
level,
serde_json::json!([]),
"sys".to_string(),
vec![],
None,
None,
);
assert!(
body.get("reasoning").is_none(),
"{level:?} must not emit reasoning field, body={body}"
);
}
}
#[test]
fn max_as_str_is_not_xhigh() {
assert_ne!(ReasoningLevel::Max.as_str(), ReasoningLevel::XHigh.as_str());
assert_eq!(ReasoningLevel::Max.as_str(), "max");
}
#[test]
fn ultra_mode_context_is_exact_proactive_developer_item() {
let plan = plan(ReasoningLevel::Ultra, CodexRequestRole::Foreground);
assert_eq!(plan.multi_agent_mode, Some(CodexMultiAgentMode::Proactive));
let item = super::codex_multi_agent_mode_item(&plan).expect("Ultra context item");
assert_eq!(
item.get("type").and_then(serde_json::Value::as_str),
Some("message")
);
assert_eq!(
item.get("role").and_then(serde_json::Value::as_str),
Some("developer")
);
assert_eq!(
item.pointer("/content/0/text")
.and_then(serde_json::Value::as_str),
Some(concat!(
"<multi_agent_mode>",
"Proactive multi-agent delegation is active. Any earlier instruction requiring an explicit user request before spawning sub-agents no longer applies. Use sub-agents when parallel work would materially improve speed or quality. This mode remains active until a later multi-agent mode developer message changes it.",
"</multi_agent_mode>"
))
);
}
#[test]
fn max_and_xhigh_mode_context_is_exact_explicit_only_item() {
for level in [ReasoningLevel::Max, ReasoningLevel::XHigh] {
let plan = plan(level, CodexRequestRole::Foreground);
assert_eq!(
plan.multi_agent_mode,
Some(CodexMultiAgentMode::ExplicitRequestOnly)
);
let item = super::codex_multi_agent_mode_item(&plan).expect("v2 context item");
assert_eq!(
item.pointer("/content/0/text")
.and_then(serde_json::Value::as_str),
Some(concat!(
"<multi_agent_mode>",
"Do not spawn sub-agents unless the user or applicable AGENTS.md/skill instructions explicitly ask for sub-agents, delegation, or parallel agent work.",
"</multi_agent_mode>"
)),
"{level:?}"
);
}
}
#[test]
fn mode_context_is_inserted_once_at_stable_head_position() {
let plan = plan(ReasoningLevel::Ultra, CodexRequestRole::Foreground);
let mut input = vec![
serde_json::json!({"role":"user","content":"first"}),
serde_json::json!({"role":"assistant","content":"answer"}),
serde_json::json!({"role":"user","content":"current"}),
];
super::insert_codex_multi_agent_mode(&mut input, &plan);
let mode_positions = input
.iter()
.enumerate()
.filter(|(_, item)| {
item.pointer("/content/0/text")
.and_then(serde_json::Value::as_str)
.is_some_and(|text| text.starts_with("<multi_agent_mode>"))
})
.map(|(index, _)| index)
.collect::<Vec<_>>();
assert_eq!(mode_positions, vec![0]);
assert_eq!(
input[3].get("content").and_then(serde_json::Value::as_str),
Some("current")
);
}
/// Prefix-cache stability: as the conversation grows across user turns,
/// the already-sent items must remain a byte-identical prefix of the next
/// request's input. The old before-last-user placement moved the mode
/// item every turn, invalidating the cached prefix at the prior user
/// message and re-billing the entire previous agentic turn uncached.
#[test]
fn mode_context_placement_preserves_prompt_prefix_across_turns() {
let plan = plan(ReasoningLevel::Ultra, CodexRequestRole::Foreground);
let turn1 = vec![
serde_json::json!({"role":"user","content":"first"}),
serde_json::json!({"role":"assistant","content":"answer"}),
];
let mut turn2 = turn1.clone();
turn2.push(serde_json::json!({"role":"user","content":"follow-up"}));
let mut input1 = turn1;
let mut input2 = turn2;
super::insert_codex_multi_agent_mode(&mut input1, &plan);
super::insert_codex_multi_agent_mode(&mut input2, &plan);
assert_eq!(
&input2[..input1.len()],
&input1[..],
"turn N's input must be a strict prefix of turn N+1's input"
);
}
#[test]
fn prompt_cache_key_is_stable_for_the_same_conversation_head() {
let first = serde_json::json!({"role":"user","content":"hello"});
let a = super::codex_prompt_cache_key("sys", Some(&first));
let b = super::codex_prompt_cache_key("sys", Some(&first));
assert_eq!(a, b);
assert!(a.starts_with("synaps-"), "{a}");
}
#[test]
fn prompt_cache_key_differs_across_conversations() {
let first = serde_json::json!({"role":"user","content":"hello"});
let other = serde_json::json!({"role":"user","content":"different task"});
assert_ne!(
super::codex_prompt_cache_key("sys", Some(&first)),
super::codex_prompt_cache_key("sys", Some(&other)),
);
assert_ne!(
super::codex_prompt_cache_key("sys", Some(&first)),
super::codex_prompt_cache_key("other-sys", Some(&first)),
);
}
#[test]
fn body_carries_prompt_cache_key() {
let body = body_for_level(
"gpt-5.6-sol",
ReasoningLevel::Medium,
serde_json::json!([]),
"sys".to_string(),
vec![],
None,
None,
);
assert_eq!(
body.get("prompt_cache_key")
.and_then(serde_json::Value::as_str),
Some("synaps-test-cache-key")
);
}
#[test]
fn worker_ultra_has_no_mode_context_item() {
let plan = plan(ReasoningLevel::Ultra, CodexRequestRole::Worker);
assert!(super::codex_multi_agent_mode_item(&plan).is_none());
}
// ── Pure body construction tests (zero network) ──────────────────────────
#[test]
fn body_lowers_ultra_reasoning_effort_to_max() {
let body = body_for_level(
"gpt-5.6-sol",
ReasoningLevel::Ultra,
serde_json::json!([]),
"sys".to_string(),
vec![],
None,
None,
);
let effort = body
.get("reasoning")
.and_then(|r| r.get("effort"))
.and_then(serde_json::Value::as_str)
.expect("reasoning.effort must be present for Ultra");
assert_eq!(effort, "max");
}
#[test]
fn body_contains_reasoning_effort_for_max() {
let body = body_for_level(
"gpt-5.6-sol",
ReasoningLevel::Max,
serde_json::json!([]),
"sys".to_string(),
vec![],
None,
None,
);
let effort = body
.get("reasoning")
.and_then(|r| r.get("effort"))
.and_then(serde_json::Value::as_str)
.expect("reasoning.effort must be present for Max");
assert_eq!(effort, "max");
// max must NOT be xhigh (critical invariant)
assert_ne!(effort, "xhigh");
}
#[test]
fn body_contains_reasoning_effort_for_xhigh() {
let body = body_for_level(
"gpt-5.6-sol",
ReasoningLevel::XHigh,
serde_json::json!([]),
"sys".to_string(),
vec![],
None,
None,
);
let effort = body
.get("reasoning")
.and_then(|r| r.get("effort"))
.and_then(serde_json::Value::as_str)
.expect("reasoning.effort must be present for XHigh");
assert_eq!(effort, "xhigh");
}
#[test]
fn body_contains_reasoning_effort_for_low_medium_high() {
for (level, expected) in [
(ReasoningLevel::Low, "low"),
(ReasoningLevel::Medium, "medium"),
(ReasoningLevel::High, "high"),
] {
let body = body_for_level(
"gpt-5.6-sol",
level,
serde_json::json!([]),
"sys".to_string(),
vec![],
None,
None,
);
let effort = body
.get("reasoning")
.and_then(|r| r.get("effort"))
.and_then(serde_json::Value::as_str)
.unwrap_or_else(|| panic!("{level:?} must emit reasoning.effort"));
assert_eq!(effort, expected, "{level:?}");
}
}
#[test]
fn body_always_sets_model_and_stream_and_include() {
let body = body_for_level(
"gpt-5.5",
ReasoningLevel::XHigh,
serde_json::json!([]),
"instructions".to_string(),
vec![],
None,
None,
);
assert_eq!(
body.get("model").and_then(serde_json::Value::as_str),
Some("gpt-5.5")
);
assert_eq!(
body.get("stream").and_then(serde_json::Value::as_bool),
Some(true)
);
let includes = body.get("include").expect("include must be present");
assert!(
includes.as_array().is_some_and(|a| {
a.iter()
.any(|v| v.as_str() == Some("reasoning.encrypted_content"))
}),
"include must contain reasoning.encrypted_content"
);
assert_eq!(
body.get("store").and_then(serde_json::Value::as_bool),
Some(false)
);
}
#[test]
fn body_omits_temperature_and_max_tokens_when_none() {
let body = body_for_level(
"gpt-5.6-sol",
ReasoningLevel::Medium,
serde_json::json!([]),
"sys".to_string(),
vec![],
None,
None,
);
assert!(body.get("temperature").is_none());
assert!(body.get("max_output_tokens").is_none());
}
#[test]
fn body_includes_temperature_and_max_tokens_when_some() {
let body = body_for_level(
"gpt-5.6-sol",
ReasoningLevel::High,
serde_json::json!([]),
"sys".to_string(),
vec![],
Some(0.7),
Some(2048),
);
let temp = body
.get("temperature")
.and_then(serde_json::Value::as_f64)
.unwrap();
assert!((temp - 0.7f64).abs() < 0.01, "temperature={temp}");
assert_eq!(
body.get("max_output_tokens")
.and_then(serde_json::Value::as_u64),
Some(2048)
);
}
}
#[cfg(test)]
mod send_retry_tests {
//! Regression tests for transient-failure retry on the codex send path
//! (incident: session 20260714-025948-3dab — a single transport failure
//! against chatgpt.com aborted a whole autonomous turn because the codex
//! route was single-shot while the Anthropic route retried).
//!
//! Mirrors the `on401_tests` mock-server pattern in `runtime/api.rs`.
//! Credentials cross the broker boundary: a stub broker vends a
//! JWT-shaped token carrying the ChatGPT account-id claim.
use super::*;
use agent_core::auth::{
AccessToken, BrokerError, ProxyByteStream, ProxyRequest, ProxyResponse,
};
use agent_core::reasoning::ReasoningLevel;
use async_trait::async_trait;
use axum::{http::StatusCode, response::IntoResponse, routing::post as axum_post, Router};
use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
use std::sync::{
atomic::{AtomicUsize, Ordering},
Arc,
};
use tokio::sync::mpsc;
/// Minimal Responses-API SSE success body: one text delta + completed.
const CODEX_SSE_SUCCESS: &str = concat!(
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"hello\"}\n\n",
"data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n\n",
"data: [DONE]\n\n",
);
/// JWT-shaped token whose payload carries the ChatGPT account-id claim
/// `extract_codex_account_id` looks for. Signature is irrelevant — the
/// mock server never validates it.
fn fake_codex_token() -> String {
let payload = serde_json::json!({
"https://api.openai.com/auth": { "chatgpt_account_id": "acct_test" }
});
format!("h.{}.s", URL_SAFE_NO_PAD.encode(payload.to_string()))
}
/// Stub broker that vends the fake codex token. Everything else is
/// unreachable in these tests and fails closed.
struct TokenOnlyBroker;
#[async_trait]
impl crate::auth::CredentialBroker for TokenOnlyBroker {
async fn access_token(
&self,
_p: agent_core::auth::OAuthProviderId,
) -> Result<AccessToken, BrokerError> {
Ok(AccessToken {
token: fake_codex_token(),
expires: u64::MAX,
})
}
async fn proxy(&self, _request: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
Err(BrokerError::Denied("not implemented in stub".into()))
}
async fn proxy_stream(
&self,
_request: ProxyRequest,
) -> Result<ProxyByteStream, BrokerError> {
Err(BrokerError::Denied("not implemented in stub".into()))
}
async fn anthropic_usage(&self) -> Result<serde_json::Value, BrokerError> {
Err(BrokerError::Denied("not implemented in stub".into()))
}
async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
Ok(vec![])
}
}
struct DenyingCountingBroker {
access_calls: Arc<AtomicUsize>,
}
#[async_trait]
impl crate::auth::CredentialBroker for DenyingCountingBroker {
async fn access_token(
&self,
_provider: agent_core::auth::OAuthProviderId,
) -> Result<AccessToken, BrokerError> {
self.access_calls.fetch_add(1, Ordering::SeqCst);
Err(BrokerError::Denied(
"credential access must not occur".into(),
))
}
async fn proxy(&self, _request: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
Err(BrokerError::Denied("not implemented in stub".into()))
}
async fn proxy_stream(
&self,
_request: ProxyRequest,
) -> Result<ProxyByteStream, BrokerError> {
Err(BrokerError::Denied("not implemented in stub".into()))
}
async fn anthropic_usage(&self) -> Result<serde_json::Value, BrokerError> {
Err(BrokerError::Denied("not implemented in stub".into()))
}
async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
Ok(vec![])
}
}
/// Spawn a mock Codex endpoint. First `fail_count` POSTs → `fail_status`;
/// subsequent → SSE success. Returns (base_url, call_counter).
async fn spawn_mock_codex(
fail_count: usize,
fail_status: StatusCode,
) -> (String, Arc<AtomicUsize>) {
let counter = Arc::new(AtomicUsize::new(0));
let counter_clone = Arc::clone(&counter);
let app = Router::new().route(
"/codex/responses",
axum_post(move || {
let counter = Arc::clone(&counter_clone);
async move {
let n = counter.fetch_add(1, Ordering::SeqCst);
if n < fail_count {
(
fail_status,
[("content-type", "application/json")],
"{\"error\":{\"message\":\"transient upstream sadness\"}}".to_string(),
)
.into_response()
} else {
(
StatusCode::OK,
[("content-type", "text/event-stream")],
CODEX_SSE_SUCCESS.to_string(),
)
.into_response()
}
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
(format!("http://{addr}"), counter)
}
async fn spawn_codex_sse(body: &'static str) -> String {
let app = Router::new().route(
"/codex/responses",
axum_post(move || async move {
(
StatusCode::OK,
[("content-type", "text/event-stream")],
body,
)
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
format!("http://{addr}")
}
async fn spawn_empty_then_success_codex() -> (
String,
Arc<AtomicUsize>,
Arc<std::sync::Mutex<Vec<bytes::Bytes>>>,
) {
let counter = Arc::new(AtomicUsize::new(0));
let bodies = Arc::new(std::sync::Mutex::new(Vec::new()));
let app = Router::new().route(
"/codex/responses",
axum_post({
let counter = Arc::clone(&counter);
let bodies = Arc::clone(&bodies);
move |body: bytes::Bytes| {
let counter = Arc::clone(&counter);
let bodies = Arc::clone(&bodies);
async move {
bodies.lock().unwrap().push(body);
let n = counter.fetch_add(1, Ordering::SeqCst);
let response = if n == 0 {
concat!(
"data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":7,\"output_tokens\":0}}}\n\n",
"data: [DONE]\n\n",
)
} else {
CODEX_SSE_SUCCESS
};
(
StatusCode::OK,
[("content-type", "text/event-stream")],
response,
)
}
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
(format!("http://{addr}"), counter, bodies)
}
async fn run_codex(
base_url: &str,
max_retries: u32,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
let cfg = ProviderConfig {
base_url: base_url.to_string(),
// The incident model — validated against the static catalog ladder.
model: "gpt-5.6-sol".to_string(),
provider: "openai-codex".to_string(),
};
let client = reqwest::Client::new();
let broker: std::sync::Arc<dyn crate::auth::CredentialBroker> =
std::sync::Arc::new(TokenOnlyBroker);
let (tx, _rx) = mpsc::unbounded_channel();
call_codex_stream_inner(
&cfg,
&client,
&broker,
&[],
&Some("test".to_string()),
&[],
&tx,
None,
None,
agent_core::reasoning::ReasoningLevel::Medium,
crate::runtime::openai::catalog::CodexRequestRole::Foreground,
&tokio_util::sync::CancellationToken::new(),
max_retries,
&crate::runtime::trace::TraceContext::disabled(),
)
.await
}
// Raw HTTP is intentional: axum's complete bodies cannot reproduce the
// reported EOF in an HTTP/1.1 chunk-size line.
async fn broken_chunk_server(
scripts: Vec<(&'static str, bool)>,
) -> (
String,
Arc<std::sync::Mutex<Vec<Vec<u8>>>>,
tokio::task::JoinHandle<()>,
) {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let requests = Arc::new(std::sync::Mutex::new(Vec::new()));
let seen = requests.clone();
let task = tokio::spawn(async move {
let mut attempt = 0;
loop {
let (mut socket, _) = listener.accept().await.unwrap();
let mut request = Vec::new();
let end = loop {
let mut chunk = [0; 8192];
let n = socket.read(&mut chunk).await.unwrap();
if n == 0 {
return;
}
request.extend_from_slice(&chunk[..n]);
if let Some(end) = request.windows(4).position(|w| w == b"\r\n\r\n") {
break end + 4;
}
};
let length: usize = String::from_utf8_lossy(&request[..end])
.lines()
.find_map(|line| {
line.to_ascii_lowercase()
.strip_prefix("content-length: ")
.map(str::to_owned)
})
.unwrap()
.parse()
.unwrap();
while request.len() < end + length {
let mut chunk = [0; 8192];
let n = socket.read(&mut chunk).await.unwrap();
if n == 0 {
return;
}
request.extend_from_slice(&chunk[..n]);
}
seen.lock()
.unwrap()
.push(request[end..end + length].to_vec());
let (body, broken) = scripts[attempt.min(scripts.len() - 1)];
attempt += 1;
socket.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n").await.unwrap();
socket
.write_all(format!("{:x}\r\n{}\r\n", body.len(), body).as_bytes())
.await
.unwrap();
// Omitting the 0 chunk produces precisely the user's failure.
if !broken {
socket.write_all(b"0\r\n\r\n").await.unwrap();
}
socket.shutdown().await.unwrap();
}
});
(format!("http://{address}"), requests, task)
}
async fn codex_with_events(
base: &str,
retries: u32,
cancel: &tokio_util::sync::CancellationToken,
tx: &mpsc::UnboundedSender<StreamEvent>,
) -> Result<Value, super::super::net::BoxedProviderError> {
let cfg = ProviderConfig {
base_url: base.into(),
model: "gpt-5.6-sol".into(),
provider: "openai-codex".into(),
};
let broker: Arc<dyn crate::auth::CredentialBroker> = Arc::new(TokenOnlyBroker);
call_codex_stream_inner(
&cfg,
&reqwest::Client::new(),
&broker,
&[],
&Some("test".into()),
&[],
tx,
None,
None,
agent_core::reasoning::ReasoningLevel::Medium,
crate::runtime::openai::catalog::CodexRequestRole::Foreground,
cancel,
retries,
&crate::runtime::trace::TraceContext::disabled(),
)
.await
}
const PARTIAL_TOOL: &str = concat!(
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"discard me\"}\n\n",
"data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"old_call\",\"name\":\"write\",\"arguments\":\"\"}}\n\n",
"data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"path\\\":\"}\n\n",
);
const COMPLETE_TOOL: &str = concat!(
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"accepted\"}\n\n",
"data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"type\":\"function_call\",\"call_id\":\"new_call\",\"name\":\"write\",\"arguments\":\"{\\\"path\\\":\\\"ok\\\"}\"}}\n\n",
"data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":3,\"output_tokens\":2}}}\n\n",
);
#[tokio::test]
async fn interrupted_partial_tool_retries_identical_bytes_and_commits_only_success() {
use crate::LlmEvent;
let (url, requests, server) =
broken_chunk_server(vec![(PARTIAL_TOOL, true), (COMPLETE_TOOL, false)]).await;
let (tx, mut rx) = mpsc::unbounded_channel();
let result = codex_with_events(&url, 1, &tokio_util::sync::CancellationToken::new(), &tx)
.await
.unwrap();
server.abort();
let requests = requests.lock().unwrap();
assert_eq!(requests.len(), 2);
assert_eq!(requests[0], requests[1]);
let tools: Vec<_> = result["content"]
.as_array()
.unwrap()
.iter()
.filter(|v| v["type"] == "tool_use")
.collect();
assert_eq!(tools.len(), 1);
assert_eq!(tools[0]["id"], "new_call");
assert_eq!(result["content"][0]["text"], "accepted");
let mut preview = String::from("prior completed round");
let mut baseline = 0;
let mut resets = 0;
let mut input_tokens = 0;
while let Ok(event) = rx.try_recv() {
match event {
StreamEvent::Llm(LlmEvent::ResponseStart) => baseline = preview.len(),
StreamEvent::Llm(LlmEvent::ResponseReset) => {
preview.truncate(baseline);
resets += 1;
}
StreamEvent::Llm(LlmEvent::Text(t)) => preview.push_str(&t),
StreamEvent::Session(crate::SessionEvent::Usage {
input_tokens: n, ..
}) => input_tokens += n,
_ => {}
}
}
assert_eq!(resets, 1);
assert_eq!(preview, "prior completed roundaccepted");
assert_eq!(input_tokens, 3);
}
#[tokio::test]
async fn completed_response_followed_by_chunk_error_is_not_replayed() {
let (url, requests, server) = broken_chunk_server(vec![(CODEX_SSE_SUCCESS, true)]).await;
assert!(run_codex(&url, 2).await.is_ok());
server.abort();
assert_eq!(requests.lock().unwrap().len(), 1);
}
#[tokio::test]
async fn clean_truncated_eof_retries_but_never_accepts_partial_tools() {
let (url, requests, server) = broken_chunk_server(vec![(PARTIAL_TOOL, false)]).await;
assert!(run_codex(&url, 1).await.is_err());
server.abort();
assert_eq!(requests.lock().unwrap().len(), 2);
}
#[tokio::test]
async fn broken_chunk_retry_budget_is_finite() {
let (url, requests, server) = broken_chunk_server(vec![(PARTIAL_TOOL, true)]).await;
assert!(run_codex(&url, 1).await.is_err());
server.abort();
assert_eq!(requests.lock().unwrap().len(), 2);
}
#[tokio::test]
async fn cancellation_during_stream_retry_prevents_second_dispatch() {
let (url, requests, server) = broken_chunk_server(vec![(PARTIAL_TOOL, true)]).await;
let (tx, mut rx) = mpsc::unbounded_channel();
let cancel = tokio_util::sync::CancellationToken::new();
let waiter = cancel.clone();
let task = tokio::spawn(async move {
while let Some(event) = rx.recv().await {
if matches!(event, StreamEvent::Session(crate::SessionEvent::Retry(_))) {
waiter.cancel();
break;
}
}
});
let error = codex_with_events(&url, 10, &cancel, &tx).await.unwrap_err();
task.await.unwrap();
server.abort();
assert_eq!(error.to_string(), "request canceled");
assert_eq!(requests.lock().unwrap().len(), 1);
}
#[tokio::test]
async fn direct_ultra_tool_guard_precedes_broker_access() {
let access_calls = Arc::new(AtomicUsize::new(0));
let broker: Arc<dyn crate::auth::CredentialBroker> = Arc::new(DenyingCountingBroker {
access_calls: Arc::clone(&access_calls),
});
let cfg = ProviderConfig {
base_url: "http://127.0.0.1:1".to_string(),
model: "gpt-5.6-sol".to_string(),
provider: "openai-codex".to_string(),
};
let (tx, _rx) = mpsc::unbounded_channel();
let error = call_codex_stream_inner(
&cfg,
&reqwest::Client::new(),
&broker,
&[],
&Some("test".to_string()),
&[],
&tx,
None,
None,
ReasoningLevel::Ultra,
crate::runtime::openai::catalog::CodexRequestRole::Foreground,
&tokio_util::sync::CancellationToken::new(),
0,
&crate::runtime::trace::TraceContext::disabled(),
)
.await
.expect_err("foreground Ultra without delegation tools must fail closed");
assert!(error.to_string().contains("subagent_start"), "{error}");
assert_eq!(
access_calls.load(Ordering::SeqCst),
0,
"preflight denial must happen before broker credential access"
);
}
#[tokio::test]
async fn direct_codex_path_rejects_provider_identity_before_broker_access() {
let access_calls = Arc::new(AtomicUsize::new(0));
let broker: Arc<dyn crate::auth::CredentialBroker> = Arc::new(DenyingCountingBroker {
access_calls: Arc::clone(&access_calls),
});
let cfg = ProviderConfig {
base_url: "http://127.0.0.1:1".to_string(),
model: "gpt-5.6-sol".to_string(),
provider: "openrouter".to_string(),
};
let (tx, _rx) = mpsc::unbounded_channel();
let error = call_codex_stream_inner(
&cfg,
&reqwest::Client::new(),
&broker,
&[],
&Some("test".to_string()),
&[],
&tx,
None,
None,
ReasoningLevel::Medium,
crate::runtime::openai::catalog::CodexRequestRole::Foreground,
&tokio_util::sync::CancellationToken::new(),
0,
&crate::runtime::trace::TraceContext::disabled(),
)
.await
.expect_err("the Codex seam must validate provider-qualified identity");
assert!(error.to_string().contains("openrouter"), "{error}");
assert_eq!(
access_calls.load(Ordering::SeqCst),
0,
"provider identity denial must precede broker credential access"
);
}
/// A stream whose terminal event is `response.failed` must surface a
/// typed API failure at the transport boundary — never `Ok` with empty
/// content (the "model returned an empty response" misdiagnosis,
/// incident: sessions 20260720-022603-6c2a / turn-2629074-0). The
/// provider-controlled error body must not leak into the message.
#[tokio::test]
async fn codex_response_failed_event_is_a_terminal_error_not_empty_success() {
let base_url = spawn_codex_sse(concat!(
"data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"server_error\",\"message\":\"ECHOED:secret prompt\"}}}\n\n",
"data: [DONE]\n\n",
))
.await;
let err = run_codex(&base_url, 0)
.await
.expect_err("response.failed must fail the request");
let msg = err.to_string();
assert!(
msg == format!("Codex{RESPONSES_CAPACITY_SUFFIX}"),
"static transient failure message expected: {msg}"
);
assert!(
!msg.contains("ECHOED") && !msg.contains("secret prompt"),
"provider error body must never surface: {msg}"
);
}
/// `response.incomplete` is a terminal non-success: surfacing it as an
/// empty `Ok` poisoned the turn with the generic empty-response error.
#[tokio::test]
async fn codex_response_incomplete_event_is_a_terminal_error() {
let base_url = spawn_codex_sse(concat!(
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n",
"data: {\"type\":\"response.incomplete\",\"response\":{\"incomplete_details\":{\"reason\":\"max_output_tokens\"}}}\n\n",
"data: [DONE]\n\n",
))
.await;
let err = run_codex(&base_url, 0)
.await
.expect_err("response.incomplete must fail the request");
assert!(
err.to_string()
.starts_with("Codex response was incomplete."),
"got: {err}"
);
}
/// EOF without any terminal Responses event (mid-stream connection drop
/// after HTTP 200) must fail closed instead of returning empty content.
#[tokio::test]
async fn codex_eof_without_terminal_event_fails_closed() {
let base_url = spawn_codex_sse(
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\"}}\n\n",
)
.await;
let err = run_codex(&base_url, 0)
.await
.expect_err("EOF without terminal event must fail");
assert!(
err.to_string()
.starts_with("Codex response stream ended without a terminal event."),
"got: {err}"
);
}
#[tokio::test]
async fn codex_completed_without_output_retries_identical_bytes_then_succeeds() {
let (base_url, counter, bodies) = spawn_empty_then_success_codex().await;
let result = run_codex(&base_url, 1)
.await
.expect("one empty-success stream with retries available must not abort the turn");
assert_eq!(counter.load(Ordering::SeqCst), 2, "expected one retry");
let received = bodies.lock().unwrap();
assert_eq!(received.len(), 2, "both attempts must reach the endpoint");
assert_eq!(
received[0], received[1],
"the logical request must resend byte-identical content"
);
assert_eq!(result["content"][0]["text"], "hello");
}
/// Mock Codex endpoint that records every request's (session-id,
/// thread-id, prompt_cache_key). The first `fail_count` POSTs fail 500.
async fn spawn_header_recording_codex(
fail_count: usize,
) -> (String, Arc<std::sync::Mutex<Vec<(String, String, String)>>>) {
let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
let counter = Arc::new(AtomicUsize::new(0));
let app = Router::new().route(
"/codex/responses",
axum_post({
let seen = Arc::clone(&seen);
move |headers: axum::http::HeaderMap, body: bytes::Bytes| {
let seen = Arc::clone(&seen);
let counter = Arc::clone(&counter);
async move {
let get = |name: &str| {
headers
.get(name)
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_string()
};
let key = serde_json::from_slice::<Value>(&body)
.ok()
.and_then(|b| b["prompt_cache_key"].as_str().map(str::to_string))
.unwrap_or_default();
seen.lock()
.unwrap()
.push((get("session-id"), get("thread-id"), key));
if counter.fetch_add(1, Ordering::SeqCst) < fail_count {
(StatusCode::INTERNAL_SERVER_ERROR, "{}".to_string()).into_response()
} else {
(
StatusCode::OK,
[("content-type", "text/event-stream")],
CODEX_SSE_SUCCESS.to_string(),
)
.into_response()
}
}
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
(format!("http://{addr}"), seen)
}
/// The ChatGPT backend derives prompt-cache affinity from the `session-id`
/// header (Codex CLI `responses_session_id`); without it every request
/// reported `cached_tokens: 0`. Measured A/B: no header 0/0/0 cached;
/// with `session-id`/`thread-id` = prompt_cache_key, 0 then 95%, 95%.
#[tokio::test]
async fn codex_requests_carry_session_and_thread_id_equal_to_the_cache_key() {
let (base_url, seen) = spawn_header_recording_codex(1).await;
run_codex(&base_url, 2)
.await
.expect("one transient 500 then success");
let seen = seen.lock().unwrap();
assert_eq!(seen.len(), 2, "first attempt + one retry");
for (session_id, thread_id, key) in seen.iter() {
assert!(
key.starts_with("synaps-"),
"prompt_cache_key in body: {key:?}"
);
assert_eq!(session_id, key, "session-id header = prompt_cache_key");
assert_eq!(thread_id, key, "thread-id header = prompt_cache_key");
}
assert_eq!(seen[0], seen[1], "a retry keeps the same cache routing");
}
#[tokio::test]
async fn codex_retries_transient_500_then_succeeds() {
let (base_url, counter) = spawn_mock_codex(1, StatusCode::INTERNAL_SERVER_ERROR).await;
let result = run_codex(&base_url, 2)
.await
.expect("one transient 500 with retries available must not abort the turn");
assert_eq!(
counter.load(Ordering::SeqCst),
2,
"expected exactly one retry"
);
let text = result["content"][0]["text"].as_str().unwrap_or_default();
assert_eq!(text, "hello");
}
#[tokio::test]
async fn codex_response_failed_surfaces_typed_provider_error_not_empty_response() {
const BODY: &str = concat!(
"data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"server_error\",\"message\":\"ECHOED:private prompt\"}}}\n\n",
"data: [DONE]\n\n",
);
let base_url = spawn_codex_sse(BODY).await;
let err = run_codex(&base_url, 0).await.expect_err("must fail");
let msg = err.to_string();
assert_eq!(msg, format!("Codex{RESPONSES_CAPACITY_SUFFIX}"));
assert!(!msg.contains("empty response"), "{msg}");
assert!(!msg.contains("ECHOED"), "provider text leaked: {msg}");
assert!(
!msg.contains("private prompt"),
"provider text leaked: {msg}"
);
}
#[tokio::test]
async fn codex_response_incomplete_surfaces_typed_provider_error_not_empty_response() {
const BODY: &str = concat!(
"data: {\"type\":\"response.incomplete\",\"response\":{\"incomplete_details\":{\"reason\":\"max_output_tokens\"}}}\n\n",
"data: [DONE]\n\n",
);
let base_url = spawn_codex_sse(BODY).await;
let err = run_codex(&base_url, 0).await.expect_err("must fail");
let msg = err.to_string();
assert!(msg.starts_with("Codex response was incomplete."), "{msg}");
assert!(!msg.contains("empty response"), "{msg}");
assert!(
!msg.contains("max_output_tokens"),
"provider text leaked: {msg}"
);
}
#[tokio::test]
async fn codex_clean_eof_without_terminal_fails_closed() {
let base_url = spawn_codex_sse(
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_123\"}}\n\n",
)
.await;
let err = run_codex(&base_url, 0).await.expect_err("must fail");
assert!(
err.to_string()
.starts_with("Codex response stream ended without a terminal event."),
"{err}"
);
}
#[tokio::test]
async fn codex_retries_429_then_succeeds() {
let (base_url, counter) = spawn_mock_codex(1, StatusCode::TOO_MANY_REQUESTS).await;
run_codex(&base_url, 2)
.await
.expect("429 is transient — must retry");
assert_eq!(counter.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn codex_zero_retries_fails_fast_with_status_error() {
let (base_url, counter) =
spawn_mock_codex(usize::MAX, StatusCode::INTERNAL_SERVER_ERROR).await;
let err = run_codex(&base_url, 0).await.expect_err("must fail");
assert_eq!(counter.load(Ordering::SeqCst), 1, "no retries budgeted");
let msg = err.to_string();
assert!(
msg.starts_with("codex request failed: 500"),
"status must survive for classification: {msg}"
);
}
#[tokio::test]
async fn codex_does_not_retry_client_errors() {
let (base_url, counter) = spawn_mock_codex(usize::MAX, StatusCode::BAD_REQUEST).await;
let err = run_codex(&base_url, 3).await.expect_err("must fail");
assert_eq!(
counter.load(Ordering::SeqCst),
1,
"400 is deterministic — retrying it never helps"
);
assert!(err.to_string().starts_with("codex request failed: 400"));
}
#[tokio::test]
async fn codex_exhausted_retries_reports_last_status() {
let (base_url, counter) =
spawn_mock_codex(usize::MAX, StatusCode::SERVICE_UNAVAILABLE).await;
let err = run_codex(&base_url, 1).await.expect_err("must fail");
assert_eq!(
counter.load(Ordering::SeqCst),
2,
"initial attempt + 1 retry"
);
assert!(err.to_string().starts_with("codex request failed: 503"));
}
// ─── G5: account-aware selection + bounded failover ──────────────────────
use crate::runtime::openai::account_routing::{
CodexAccountRouter, CodexQuotaEvidence, PinnedAccount,
};
use agent_core::auth::{Account, CredentialRef, OAuthProviderId};
/// JWT-shaped token carrying `account_id` as the ChatGPT account claim.
fn fake_codex_token_for(account_id: &str) -> String {
let payload = serde_json::json!({
"https://api.openai.com/auth": { "chatgpt_account_id": account_id }
});
format!("h.{}.s", URL_SAFE_NO_PAD.encode(payload.to_string()))
}
fn pinned(label: &str, account_id: &str, auto: bool) -> PinnedAccount {
PinnedAccount {
credential: CredentialRef::new(
OAuthProviderId::OpenAiCodex,
Account::parse(label).unwrap(),
),
token: fake_codex_token_for(account_id),
auto,
}
}
/// Scripted router: `pin` returns `first`; `failover` returns the next
/// scripted candidate (or `None`). Records every call.
struct ScriptedRouter {
first: PinnedAccount,
candidates: std::sync::Mutex<std::collections::VecDeque<Option<PinnedAccount>>>,
failover_calls: AtomicUsize,
reported: std::sync::Mutex<Vec<(String, CodexQuotaEvidence)>>,
}
impl ScriptedRouter {
fn new(first: PinnedAccount, candidates: Vec<Option<PinnedAccount>>) -> Self {
Self {
first,
candidates: std::sync::Mutex::new(candidates.into()),
failover_calls: AtomicUsize::new(0),
reported: std::sync::Mutex::new(Vec::new()),
}
}
fn reports(&self) -> Vec<String> {
self.reported
.lock()
.unwrap()
.iter()
.map(|(k, _)| k.clone())
.collect()
}
}
#[async_trait]
impl CodexAccountRouter for ScriptedRouter {
async fn pin(&self, _model: &str) -> Result<PinnedAccount, String> {
Ok(self.first.clone())
}
async fn failover(
&self,
_model: &str,
_failed: &PinnedAccount,
_evidence: &CodexQuotaEvidence,
_failovers_so_far: u32,
) -> Result<Option<PinnedAccount>, String> {
self.failover_calls.fetch_add(1, Ordering::SeqCst);
Ok(self.candidates.lock().unwrap().pop_front().flatten())
}
async fn report_exhausted(&self, failed: &PinnedAccount, evidence: &CodexQuotaEvidence) {
self.reported
.lock()
.unwrap()
.push((failed.credential.storage_key(), *evidence));
}
}
/// One observed request: `chatgpt-account-id` header, bearer token, body.
#[derive(Clone, Debug)]
struct Seen {
account: String,
bearer: String,
body: bytes::Bytes,
}
/// Mock Codex endpoint scripted per request: each entry is
/// `(status, headers, body)`; the last entry repeats. Records the
/// `chatgpt-account-id` header, the bearer and the body of every request.
type Script = Vec<(StatusCode, Vec<(&'static str, &'static str)>, &'static str)>;
async fn spawn_scripted_codex(script: Script) -> (String, Arc<std::sync::Mutex<Vec<Seen>>>) {
let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
let counter = Arc::new(AtomicUsize::new(0));
let script = Arc::new(script);
let app = Router::new().route(
"/codex/responses",
axum_post({
let seen = Arc::clone(&seen);
let counter = Arc::clone(&counter);
let script = Arc::clone(&script);
move |headers: axum::http::HeaderMap, body: bytes::Bytes| {
let seen = Arc::clone(&seen);
let counter = Arc::clone(&counter);
let script = Arc::clone(&script);
async move {
let header = |name: &str| {
headers
.get(name)
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_string()
};
let bearer = header("authorization")
.strip_prefix("Bearer ")
.unwrap_or("")
.to_string();
seen.lock().unwrap().push(Seen {
account: header("chatgpt-account-id"),
bearer,
body,
});
let n = counter.fetch_add(1, Ordering::SeqCst);
let (status, hdrs, body) = &script[n.min(script.len() - 1)];
let mut resp = axum::response::Response::builder().status(*status);
for (k, v) in hdrs {
resp = resp.header(*k, *v);
}
let ct = if status.is_success() {
"text/event-stream"
} else {
"application/json"
};
resp.header("content-type", ct)
.body(axum::body::Body::from(*body))
.unwrap()
}
}
}),
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
(format!("http://{addr}"), seen)
}
async fn run_codex_with_router(
base_url: &str,
router: &dyn CodexAccountRouter,
max_retries: u32,
tx: &mpsc::UnboundedSender<StreamEvent>,
) -> Result<Value, Box<dyn std::error::Error + Send + Sync>> {
let cfg = ProviderConfig {
base_url: base_url.to_string(),
model: "gpt-5.6-sol".to_string(),
provider: "openai-codex".to_string(),
};
call_codex_stream_with_router(
&cfg,
&reqwest::Client::new(),
router,
&[],
&Some("test".to_string()),
&[],
tx,
None,
None,
ReasoningLevel::Medium,
crate::runtime::openai::catalog::CodexRequestRole::Foreground,
&tokio_util::sync::CancellationToken::new(),
max_retries,
&crate::runtime::trace::TraceContext::disabled(),
)
.await
}
const QUOTA_429: &str = r#"{"error":{"type":"usage_limit_reached","message":"ECHOED prompt","plan_type":"plus","resets_in_seconds":3600}}"#;
const GENERIC_429: &str = r#"{"error":{"type":"rate_limit_exceeded","message":"slow down"}}"#;
fn notices(rx: &mut mpsc::UnboundedReceiver<StreamEvent>) -> Vec<String> {
let mut out = Vec::new();
while let Ok(ev) = rx.try_recv() {
if let StreamEvent::Session(
crate::SessionEvent::Notice(n) | crate::SessionEvent::Retry(n),
) = ev
{
out.push(n);
}
}
out
}
/// Auto policy, exhausted 429 on seat A before any output, seat B fresh:
/// exactly one switch, the retry carries B's header (derived from B's
/// token), identical body bytes, and the cooldown was reported for A.
#[tokio::test]
async fn quota_429_fails_over_once_to_distinct_account_with_paired_header() {
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::TOO_MANY_REQUESTS, vec![], QUOTA_429),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, mut rx) = mpsc::unbounded_channel();
let result = run_codex_with_router(&url, &router, 3, &tx).await.unwrap();
assert_eq!(result["content"][0]["text"], "hello");
let seen = seen.lock().unwrap();
assert_eq!(seen.len(), 2, "one failed attempt on A, one success on B");
assert_eq!(seen[0].account, "acct_A");
assert_eq!(seen[1].account, "acct_B", "header must follow the pinned token");
// Bearer and header are derived from the SAME credential on every
// attempt: A's token with A's id, then B's token with B's id — never
// a mixed pair.
assert_eq!(seen[0].bearer, fake_codex_token_for("acct_A"));
assert_eq!(seen[1].bearer, fake_codex_token_for("acct_B"));
assert_ne!(seen[0].bearer, seen[1].bearer);
assert_eq!(seen[0].body, seen[1].body, "same request bytes, no rebuild");
assert_eq!(router.failover_calls.load(Ordering::SeqCst), 1);
assert_eq!(router.reports(), vec!["openai-codex@astra1".to_string()]);
let notes = notices(&mut rx);
assert!(
notes.iter().any(|n| n.contains("'astra1'") && n.contains("'astra2'")),
"operator must see the switch: {notes:?}"
);
assert!(
notes.iter().all(|n| !n.contains("acct_") && !n.contains("h.")),
"notices carry labels only, never ids/tokens: {notes:?}"
);
}
/// Explicitly selected account: never switched, even with an eligible
/// alternative. The turn ends with the static quota message and no
/// transport retries are burned on a provably exhausted seat.
#[tokio::test]
async fn quota_429_on_explicit_account_never_fails_over() {
let (url, seen) = spawn_scripted_codex(vec![(
StatusCode::TOO_MANY_REQUESTS,
vec![],
QUOTA_429,
)])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", false),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 5, &tx)
.await
.expect_err("explicit account must fail, not switch");
assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
assert!(!err.to_string().contains("ECHOED"));
assert_eq!(seen.lock().unwrap().len(), 1, "no retry on exhausted seat");
assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
// The cooldown is still reported so the broker stops advertising A.
assert_eq!(router.reports(), vec!["openai-codex@astra1".to_string()]);
}
/// A generic 429 (no exhaustion evidence) keeps the ordinary transport
/// retry on the SAME account — it is not proof of a spent window.
#[tokio::test]
async fn generic_429_retries_same_account_and_never_switches() {
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::TOO_MANY_REQUESTS, vec![], GENERIC_429),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
run_codex_with_router(&url, &router, 2, &tx).await.unwrap();
let seen = seen.lock().unwrap();
assert_eq!(seen.len(), 2);
assert!(seen.iter().all(|r| r.account == "acct_A"));
assert!(seen.iter().all(|r| r.bearer == fake_codex_token_for("acct_A")));
assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
assert!(router.reports().is_empty());
}
/// Exhausted-window headers on a 429 are provider-declared evidence too.
#[tokio::test]
async fn quota_429_by_headers_fails_over() {
let (url, seen) = spawn_scripted_codex(vec![
(
StatusCode::TOO_MANY_REQUESTS,
vec![
("x-codex-primary-used-percent", "100"),
("x-codex-primary-reset-after-seconds", "500"),
],
"{}",
),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
run_codex_with_router(&url, &router, 2, &tx).await.unwrap();
assert_eq!(seen.lock().unwrap()[1].account, "acct_B");
let reported = router.reported.lock().unwrap();
assert_eq!(reported[0].1.reset_after_secs, Some(500));
}
/// No eligible seat: one quota failure ends the turn — no loop, no
/// second attempt on the exhausted account.
#[tokio::test]
async fn quota_429_with_no_eligible_candidate_fails_without_looping() {
let (url, seen) = spawn_scripted_codex(vec![(
StatusCode::TOO_MANY_REQUESTS,
vec![],
QUOTA_429,
)])
.await;
let router = ScriptedRouter::new(pinned("astra1", "acct_A", true), vec![None]);
let (tx, _rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 10, &tx)
.await
.expect_err("must fail");
assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
assert_eq!(seen.lock().unwrap().len(), 1);
assert_eq!(router.failover_calls.load(Ordering::SeqCst), 1);
}
/// The router hands back the very account that failed (e.g. a broker
/// whose cooldown sink is a no-op): refused, never retried.
#[tokio::test]
async fn failover_to_same_account_is_refused() {
let (url, seen) = spawn_scripted_codex(vec![(
StatusCode::TOO_MANY_REQUESTS,
vec![],
QUOTA_429,
)])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra1", "acct_A", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 10, &tx)
.await
.expect_err("must fail");
assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
assert_eq!(seen.lock().unwrap().len(), 1);
}
/// Both seats exhausted: exactly one switch, then stop (bounded to
/// `MAX_CODEX_ACCOUNT_FAILOVERS`), even with a third candidate scripted.
#[tokio::test]
async fn all_accounts_exhausted_stops_after_one_failover() {
let (url, seen) = spawn_scripted_codex(vec![(
StatusCode::TOO_MANY_REQUESTS,
vec![],
QUOTA_429,
)])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![
Some(pinned("astra2", "acct_B", true)),
Some(pinned("astra3", "acct_C", true)),
],
);
let (tx, _rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 10, &tx)
.await
.expect_err("must fail");
assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
let seen = seen.lock().unwrap();
assert_eq!(seen.len(), 2, "A then B, never C");
assert_eq!(seen[0].account, "acct_A");
assert_eq!(seen[1].account, "acct_B");
assert_eq!(router.failover_calls.load(Ordering::SeqCst), 1);
assert_eq!(
router.reports(),
vec![
"openai-codex@astra1".to_string(),
"openai-codex@astra2".to_string()
],
"both exhausted seats reported for cooldown"
);
}
/// In-stream quota failure (HTTP 200 + terminal `insufficient_quota`)
/// with NO output produced: one failover.
#[tokio::test]
async fn in_stream_quota_before_output_fails_over() {
const QUOTA_SSE: &str = concat!(
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"r\"}}\n\n",
"data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"insufficient_quota\",\"message\":\"ECHOED\"}}}\n\n",
"data: [DONE]\n\n",
);
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::OK, vec![], QUOTA_SSE),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
let result = run_codex_with_router(&url, &router, 2, &tx).await.unwrap();
assert_eq!(result["content"][0]["text"], "hello");
let seen = seen.lock().unwrap();
assert_eq!(seen.len(), 2);
assert_eq!(seen[1].account, "acct_B");
assert_eq!(seen[1].bearer, fake_codex_token_for("acct_B"));
}
/// In-stream quota failure AFTER the model produced content — here only
/// a reasoning-summary delta, which this decoder does not even forward —
/// is terminal: no replay on another account.
#[tokio::test]
async fn in_stream_quota_after_any_output_never_replays() {
const QUOTA_AFTER_REASONING: &str = concat!(
"data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"thinking…\"}\n\n",
"data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"code\":\"quota_exceeded\",\"message\":\"x\"}}}\n\n",
"data: [DONE]\n\n",
);
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::OK, vec![], QUOTA_AFTER_REASONING),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 5, &tx)
.await
.expect_err("must not replay after output");
assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
assert_eq!(seen.lock().unwrap().len(), 1);
assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
// Still reported: the seat IS exhausted.
assert_eq!(router.reports(), vec!["openai-codex@astra1".to_string()]);
}
/// Text already streamed to the UI, then quota: terminal.
#[tokio::test]
async fn in_stream_quota_after_text_never_replays() {
const QUOTA_AFTER_TEXT: &str = concat!(
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n",
"data: {\"type\":\"error\",\"code\":\"usage_limit_reached\",\"message\":\"x\"}\n\n",
);
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::OK, vec![], QUOTA_AFTER_TEXT),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 5, &tx)
.await
.expect_err("must not replay after output");
assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
assert_eq!(seen.lock().unwrap().len(), 1);
assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
}
/// `usage_not_included` is deterministic for the account: no transport
/// retry, no failover, status error surfaced.
#[tokio::test]
async fn usage_not_included_fails_fast_without_retry_or_failover() {
let (url, seen) = spawn_scripted_codex(vec![(
StatusCode::TOO_MANY_REQUESTS,
vec![],
r#"{"error":{"type":"usage_not_included","message":"no codex here"}}"#,
)])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 5, &tx)
.await
.expect_err("must fail");
assert!(err.to_string().starts_with("codex request failed: 429"), "{err}");
assert!(!err.to_string().contains("no codex here"));
assert_eq!(seen.lock().unwrap().len(), 1);
assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
assert!(router.reports().is_empty());
}
/// A transient 503 (transport retry on A) followed by a quota 429 on A:
/// still pre-output, so one failover to B is allowed; B then needs its
/// own transient retry, which the shared budget still covers — a
/// failover neither consumes nor resets transport retries.
#[tokio::test]
async fn transport_retry_then_quota_then_failover_keeps_one_retry_budget() {
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::SERVICE_UNAVAILABLE, vec![], "{}"),
(StatusCode::TOO_MANY_REQUESTS, vec![], QUOTA_429),
(StatusCode::SERVICE_UNAVAILABLE, vec![], "{}"),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
run_codex_with_router(&url, &router, 2, &tx).await.unwrap();
let seen = seen.lock().unwrap();
let accounts: Vec<&str> = seen.iter().map(|r| r.account.as_str()).collect();
assert_eq!(accounts, ["acct_A", "acct_A", "acct_B", "acct_B"]);
assert_eq!(router.failover_calls.load(Ordering::SeqCst), 1);
}
/// Transport budget already spent on A when the quota 429 arrives: the
/// failover is still permitted (it is not a transport retry), but B gets
/// no further transport retries — bounded on both axes.
#[tokio::test]
async fn failover_does_not_reset_transport_budget() {
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::SERVICE_UNAVAILABLE, vec![], "{}"),
(StatusCode::TOO_MANY_REQUESTS, vec![], QUOTA_429),
(StatusCode::SERVICE_UNAVAILABLE, vec![], "{}"),
])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 1, &tx)
.await
.expect_err("B's 503 exceeds the shared budget");
assert!(err.to_string().starts_with("codex request failed: 503"), "{err}");
let accounts: Vec<String> = seen.lock().unwrap().iter().map(|r| r.account.clone()).collect();
assert_eq!(accounts, ["acct_A", "acct_A", "acct_B"]);
}
/// Explicit NAMED account (`openai-codex@astra1`) with an in-stream quota
/// failure and an eligible alternative scripted: never rotates.
#[tokio::test]
async fn explicit_named_account_in_stream_quota_never_rotates() {
const QUOTA_SSE: &str = concat!(
"data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"insufficient_quota\",\"message\":\"x\"}}}\n\n",
"data: [DONE]\n\n",
);
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::OK, vec![], QUOTA_SSE),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", false),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, mut rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 5, &tx)
.await
.expect_err("explicit named account must not rotate");
assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
assert_eq!(seen.lock().unwrap().len(), 1);
assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
assert!(notices(&mut rx).iter().all(|n| !n.contains("switched")));
}
/// Explicit DEFAULT slot behaves the same as a named one: no rotation.
#[tokio::test]
async fn explicit_default_account_quota_429_never_rotates() {
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::TOO_MANY_REQUESTS, vec![], QUOTA_429),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let router = ScriptedRouter::new(
pinned("default", "acct_D", false),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 5, &tx)
.await
.expect_err("explicit default must not rotate");
assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
assert_eq!(seen.lock().unwrap().len(), 1);
assert_eq!(router.reports(), vec!["openai-codex".to_string()]);
}
/// Reasoning-only output as an output ITEM (no text delta, nothing
/// forwarded to the UI) followed by a quota failure: terminal.
#[tokio::test]
async fn in_stream_quota_after_reasoning_item_never_replays() {
const REASONING_ITEM_THEN_QUOTA: &str = concat!(
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"r\"}}\n\n",
"data: {\"type\":\"response.in_progress\",\"response\":{\"id\":\"r\"}}\n\n",
"data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"type\":\"reasoning\",\"id\":\"rs_1\",\"summary\":[]}}\n\n",
"data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"type\":\"insufficient_quota\",\"message\":\"x\"}}}\n\n",
"data: [DONE]\n\n",
);
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::OK, vec![], REASONING_ITEM_THEN_QUOTA),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, mut rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 5, &tx)
.await
.expect_err("reasoning output already produced");
assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
assert_eq!(seen.lock().unwrap().len(), 1);
assert_eq!(router.failover_calls.load(Ordering::SeqCst), 0);
// Nothing reached the UI — the gate is stricter than "was forwarded".
while let Ok(ev) = rx.try_recv() {
assert!(
!matches!(ev, StreamEvent::Llm(crate::LlmEvent::Text(_))),
"no text was forwarded"
);
}
}
/// A metadata-only prelude (`response.created` / `in_progress`) does not
/// count as output: failover stays permitted.
#[tokio::test]
async fn metadata_only_prelude_does_not_block_failover() {
const METADATA_THEN_QUOTA: &str = concat!(
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"r\"}}\n\n",
"data: {\"type\":\"response.in_progress\",\"response\":{\"id\":\"r\"}}\n\n",
"data: {\"type\":\"error\",\"code\":\"usage_limit_reached\",\"message\":\"x\"}\n\n",
);
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::OK, vec![], METADATA_THEN_QUOTA),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let router = ScriptedRouter::new(
pinned("astra1", "acct_A", true),
vec![Some(pinned("astra2", "acct_B", true))],
);
let (tx, _rx) = mpsc::unbounded_channel();
run_codex_with_router(&url, &router, 2, &tx).await.unwrap();
assert_eq!(seen.lock().unwrap()[1].account, "acct_B");
}
/// A failover candidate whose token carries no ChatGPT account claim
/// cannot be paired with a header: the turn fails at the runtime, and
/// no request is ever sent with a header borrowed from the failed seat.
#[tokio::test]
async fn failover_candidate_without_account_claim_never_sends_mixed_headers() {
let (url, seen) = spawn_scripted_codex(vec![
(StatusCode::TOO_MANY_REQUESTS, vec![], QUOTA_429),
(StatusCode::OK, vec![], CODEX_SSE_SUCCESS),
])
.await;
let claimless = PinnedAccount {
credential: CredentialRef::new(
OAuthProviderId::OpenAiCodex,
Account::parse("astra2").unwrap(),
),
token: format!(
"h.{}.s",
URL_SAFE_NO_PAD.encode(serde_json::json!({"sub": "nobody"}).to_string())
),
auto: true,
};
let router = ScriptedRouter::new(pinned("astra1", "acct_A", true), vec![Some(claimless)]);
let (tx, _rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 5, &tx)
.await
.expect_err("claimless token cannot be paired");
assert!(err.to_string().contains("account id"), "{err}");
let seen = seen.lock().unwrap();
assert_eq!(seen.len(), 1, "only A's attempt was sent");
assert_eq!(seen[0].account, "acct_A");
}
/// The router itself failing during failover is not a loop: report,
/// fail the turn with the quota message.
#[tokio::test]
async fn router_failover_error_ends_turn() {
struct FailingRouter(PinnedAccount);
#[async_trait]
impl CodexAccountRouter for FailingRouter {
async fn pin(&self, _m: &str) -> Result<PinnedAccount, String> {
Ok(self.0.clone())
}
async fn failover(
&self,
_m: &str,
_f: &PinnedAccount,
_e: &CodexQuotaEvidence,
_n: u32,
) -> Result<Option<PinnedAccount>, String> {
Err("broker unreachable".into())
}
async fn report_exhausted(&self, _f: &PinnedAccount, _e: &CodexQuotaEvidence) {}
}
let (url, seen) = spawn_scripted_codex(vec![(
StatusCode::TOO_MANY_REQUESTS,
vec![],
QUOTA_429,
)])
.await;
let router = FailingRouter(pinned("astra1", "acct_A", true));
let (tx, _rx) = mpsc::unbounded_channel();
let err = run_codex_with_router(&url, &router, 5, &tx)
.await
.expect_err("must fail");
assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
assert_eq!(seen.lock().unwrap().len(), 1);
}
/// The broker adapter with the default trait surface pins the policy
/// account as an explicit selection and reports no candidates: the old
/// `TokenOnlyBroker` mock keeps working and a quota 429 simply fails.
#[tokio::test]
async fn broker_adapter_default_trait_surface_is_explicit_and_fails_closed() {
let (url, seen) = spawn_scripted_codex(vec![(
StatusCode::TOO_MANY_REQUESTS,
vec![],
QUOTA_429,
)])
.await;
let err = run_codex(&url, 5).await.expect_err("must fail");
assert_eq!(err.to_string(), format!("Codex{RESPONSES_QUOTA_SUFFIX}"));
assert_eq!(seen.lock().unwrap().len(), 1, "no retry, no failover");
}
/// The dispatch seam lifts the generic three-attempt budget to the
/// persistent posture adopted for Anthropic OAuth overloads (10 retries)
/// — incident: 2026-07-16 chatgpt.com 503/520/timeout bursts.
#[test]
fn codex_retry_budget_lifts_configured_default_to_persistent_floor() {
assert_eq!(codex_retry_budget(3), CODEX_PERSISTENT_RETRIES);
assert_eq!(codex_retry_budget(0), CODEX_PERSISTENT_RETRIES);
assert_eq!(codex_retry_budget(10), 10);
}
/// A user who explicitly configured an even larger budget keeps it —
/// the floor only ever raises, never lowers.
#[test]
fn codex_retry_budget_honors_larger_user_configuration() {
assert_eq!(codex_retry_budget(15), 15);
}
/// Backoff doubles per attempt but caps at 2^6 — a 10-deep budget must
/// never sleep 512s on its final attempt.
/// 04-tui I4: the incomplete-response retry is a typed retry (the
/// header's "retrying"), not a notice the client has to recognise.
#[tokio::test]
async fn a_stream_retry_is_a_typed_retry_event() {
let (tx, mut rx) = mpsc::unbounded_channel();
let cancel = tokio_util::sync::CancellationToken::new();
let waiting = cancel.clone();
let task = tokio::spawn(async move { wait_stream_retry(&tx, &waiting, 1, 3, true).await.is_ok() });
let ev = rx.recv().await.expect("the retry is announced");
cancel.cancel();
assert!(!task.await.unwrap(), "cancel ends the wait");
assert!(
matches!(&ev, StreamEvent::Session(crate::SessionEvent::Retry(t)) if t.starts_with("Response incomplete — retrying")),
"{ev:?}"
);
}
#[test]
fn retry_delay_is_exponential_with_a_capped_tail() {
assert_eq!(retry_delay(1).as_secs(), 1);
assert_eq!(retry_delay(2).as_secs(), 2);
assert_eq!(retry_delay(4).as_secs(), 8);
assert_eq!(retry_delay(7).as_secs(), 64);
assert_eq!(retry_delay(10).as_secs(), 64);
}
// ─── Phase 1 privacy: hostile provider echoes the request ───────────────
// spec §5.1: provider error bodies are untrusted and may echo the full
// request (prompts, system text, tool schemas, credentials). Neither the
// surfaced error nor ANY log line may contain response-body content —
// status + provider label only.
/// Unique raw-content sentinel placed in the outgoing request body.
const ECHO_SENTINEL: &str = "PH1-OAI-SENTINEL-2b8c4d-RAW-CONTENT";
/// `std::io::Write` sink capturing formatted tracing output in-process.
struct CaptureWriter(Arc<std::sync::Mutex<Vec<u8>>>);
impl std::io::Write for CaptureWriter {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
fn capture_subscriber(
buf: &Arc<std::sync::Mutex<Vec<u8>>>,
) -> impl tracing::Subscriber + Send + Sync {
let sink = buf.clone();
tracing_subscriber::fmt()
.with_max_level(tracing::Level::TRACE)
.with_writer(move || CaptureWriter(sink.clone()))
.with_ansi(false)
.finish()
}
/// Hostile loopback provider: every request is answered with `status` and
/// a JSON error envelope whose `message` is `"ECHOED:" + the full request
/// body` — the preserved holdout probe shape, aimed at the OpenAI routes.
async fn spawn_hostile_echo_provider(status: StatusCode) -> String {
let app = Router::new().fallback(move |body: String| async move {
let envelope = serde_json::json!({
"error": { "message": format!("ECHOED:{body}") }
});
(
status,
[("content-type", "application/json")],
envelope.to_string(),
)
.into_response()
});
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
format!("http://{addr}")
}
/// Request body shaped like a real provider request: prompt sentinel plus
/// a distinctive tool schema. If the provider echo survives anywhere,
/// these markers betray it.
fn sentinel_request_body() -> Value {
json!({
"model": "gpt-test",
"messages": [
{"role": "system", "content": format!("system secret {ECHO_SENTINEL}")},
{"role": "user", "content": ECHO_SENTINEL},
],
"tools": [{
"type": "function",
"function": {
"name": "ph1_secret_tool_zz",
"description": "internal tool schema",
"parameters": {"type": "object", "properties": {}}
}
}],
})
}
/// Markers that must never surface in errors or logs.
const BANNED_MARKERS: &[&str] = &[
"ECHOED",
ECHO_SENTINEL,
"ph1_secret_tool_zz",
"\"messages\"",
"system secret",
];
fn assert_no_banned(haystack: &str, context: &str) {
for banned in BANNED_MARKERS {
assert!(
!haystack.contains(banned),
"provider body content `{banned}` leaked into {context}: {haystack}"
);
}
}
#[tokio::test]
async fn send_with_retries_non_retryable_error_never_surfaces_provider_body() {
let url = spawn_hostile_echo_provider(StatusCode::BAD_REQUEST).await;
let client = reqwest::Client::new();
let body = sentinel_request_body();
let buf = Arc::new(std::sync::Mutex::new(Vec::new()));
let _guard = tracing::subscriber::set_default(capture_subscriber(&buf));
let err = send_with_retries(
SendTarget {
label: "codex",
url: &url,
quota_probe: true,
},
|| client.post(&url).json(&body),
&tokio_util::sync::CancellationToken::new(),
3,
&mut 0,
&mut tr::StreamAttempt::new(None),
)
.await
.expect_err("400 must fail fast");
let msg = err.to_string();
assert!(
msg.starts_with("codex request failed: 400"),
"status + label must survive for classification/usability: {msg}"
);
assert_no_banned(&msg, "the surfaced error");
let logs = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
assert_no_banned(&logs, "tracing output");
}
/// Answer every request with `status` and a fixed body.
async fn spawn_fixed_body_provider(status: StatusCode, body: String) -> String {
let app = Router::new().fallback(move || {
let body = body.clone();
async move { (status, [("content-type", "application/json")], body).into_response() }
});
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
format!("http://{addr}")
}
/// Codex 4xx: identifiers and Synaps-authored text only. Every probe is
/// plain prose that a quote/length/charset sanitizer lets through; none
/// of it may reach the error string, the net.rs classification log, or
/// any captured tracing output.
#[tokio::test]
async fn codex_4xx_never_surfaces_provider_prose_in_error_or_logs() {
const PROSE: &str = "ZZPROSE";
let probes = [
// Apostrophe flips naive quote parsing; the echo follows it.
"The model can't accept 'ZZPROSE echo of the user prompt, it's long' here".to_string(),
// Short quoted span with a secret.
format!("Invalid 'input[0]': '{PROSE} AKIAIOSFODNN7EXAMPLE pls'"),
// Unquoted prose echo.
format!("Your request about {PROSE} my private quarterly notes was rejected"),
// Curly / guillemet quotes.
format!("Bad value \u{2018}{PROSE} one\u{2019} and \u{201c}{PROSE} two\u{201d} \u{ab}x\u{bb}"),
// Secret glued to punctuation.
format!("{PROSE} key:sk-proj-abc123 password=hunter2 https://h/x?token=abc"),
// Bidi override and zero-width characters.
format!("{PROSE}\u{202e}gnirts\u{202c} s\u{200b}k-\u{2066}live\u{2069}"),
];
for prose in probes {
let body = serde_json::json!({"error": {
"message": prose,
"type": "invalid_request_error",
"code": "unsupported_value",
"param": "reasoning.effort",
}})
.to_string();
let url = spawn_fixed_body_provider(StatusCode::BAD_REQUEST, body).await;
let client = reqwest::Client::new();
let buf = Arc::new(std::sync::Mutex::new(Vec::new()));
let _guard = tracing::subscriber::set_default(capture_subscriber(&buf));
let err = send_with_retries(
SendTarget {
label: "codex",
url: &url,
quota_probe: true,
},
|| client.post(&url).body("{}"),
&tokio_util::sync::CancellationToken::new(),
3,
&mut 0,
&mut tr::StreamAttempt::new(None),
)
.await
.expect_err("400 must fail fast");
// Run the error through the route classifier too: it logs.
let runtime =
super::super::net::provider_error_to_runtime_for("openai-codex/gpt-5.6-sol", err);
let msg = runtime.to_string();
let logs = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
for (where_, text) in [("error", &msg), ("logs", &logs)] {
for banned in [
PROSE, "can't", "AKIA", "sk-", "hunter2", "token=", "private", "gnirts",
"\u{202e}", "\u{200b}", "\u{2066}", "\u{2018}", "\u{201c}",
] {
assert!(
!text.contains(banned),
"provider prose `{banned}` leaked into {where_} for probe {prose:?}: {text}"
);
}
}
// Identifiers and our own explanation do surface.
assert!(
msg.contains("[invalid_request_error/unsupported_value, param reasoning.effort]"),
"{msg}"
);
assert!(msg.contains("/effort"), "static explanation missing: {msg}");
}
}
#[tokio::test(start_paused = true)]
async fn send_with_retries_retry_logs_never_contain_provider_body() {
let url = spawn_hostile_echo_provider(StatusCode::SERVICE_UNAVAILABLE).await;
let client = reqwest::Client::new();
let body = sentinel_request_body();
let buf = Arc::new(std::sync::Mutex::new(Vec::new()));
let _guard = tracing::subscriber::set_default(capture_subscriber(&buf));
let err = send_with_retries(
SendTarget {
label: "codex",
url: &url,
quota_probe: true,
},
|| client.post(&url).json(&body),
&tokio_util::sync::CancellationToken::new(),
1,
&mut 0,
&mut tr::StreamAttempt::new(None),
)
.await
.expect_err("persistent 503 must exhaust the budget");
let msg = err.to_string();
assert!(
msg.starts_with("codex request failed: 503"),
"status + label must survive after exhaustion: {msg}"
);
assert_no_banned(&msg, "the surfaced error");
let logs = String::from_utf8(buf.lock().unwrap().clone()).unwrap();
assert!(
logs.contains("retry 1/1"),
"retry accounting must stay observable: {logs}"
);
assert!(
logs.contains("503"),
"retry log must keep the status for diagnosis: {logs}"
);
assert_no_banned(&logs, "retry tracing output");
}
}
#[cfg(test)]
mod interrupted_broker_tests {
use super::*;
use crate::auth::{
AccessToken, BrokerError, CredentialBroker, ProxyByteStream, ProxyRequest, ProxyResponse,
};
use std::sync::{Arc, Mutex};
use tokio_util::sync::CancellationToken;
struct Broker {
bodies: Vec<(&'static str, Option<BrokerError>)>,
seen: Mutex<Vec<ProxyRequest>>,
}
#[async_trait::async_trait]
impl CredentialBroker for Broker {
async fn access_token(
&self,
_: crate::auth::OAuthProviderId,
) -> Result<AccessToken, BrokerError> {
unreachable!()
}
async fn proxy(&self, _: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
unreachable!()
}
async fn proxy_stream(
&self,
request: ProxyRequest,
) -> Result<ProxyByteStream, BrokerError> {
let mut seen = self.seen.lock().unwrap();
let (body, error) = &self.bodies[seen.len().min(self.bodies.len() - 1)];
seen.push(request);
let mut chunks = vec![Ok(bytes::Bytes::from_static(body.as_bytes()))];
if let Some(error) = error {
chunks.push(Err(error.clone()));
}
Ok(Box::pin(futures::stream::iter(chunks)))
}
async fn anthropic_usage(&self) -> Result<Value, BrokerError> {
unreachable!()
}
async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
Ok(vec![])
}
}
const CHAT_PARTIAL: &str = concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"discard\"}}]}\n\n",
"data: {\"choices\":[],\"usage\":{\"prompt_tokens\":7,\"completion_tokens\":1}}\n\n"
);
const CHAT_OK: &str =
"data: {\"choices\":[{\"delta\":{\"content\":\"accepted\"}}]}\n\ndata: [DONE]\n\n";
const XAI_PARTIAL: &str =
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"discard\"}\n\n";
const XAI_OK: &str = "data: {\"type\":\"response.output_text.delta\",\"delta\":\"accepted\"}\n\ndata: {\"type\":\"response.completed\"}\n\n";
async fn drive(
broker: Arc<Broker>,
xai: bool,
retries: u32,
suppressed: bool,
cancel: &CancellationToken,
) -> (
Result<Value, super::super::net::BoxedProviderError>,
Vec<StreamEvent>,
) {
let cfg = ProviderConfig {
base_url: "http://unused.invalid".into(),
model: if xai { "grok-4.5" } else { "test" }.into(),
provider: if xai { "xai-auth" } else { "local" }.into(),
};
let broker: Arc<dyn CredentialBroker> = broker;
let (tx, mut rx) = mpsc::unbounded_channel();
let trace = crate::runtime::trace::TraceContext::disabled();
let result = if xai {
call_xai_responses_stream_inner(
&cfg,
&broker,
&[],
&None,
&[],
&tx,
None,
agent_core::reasoning::ReasoningLevel::Adaptive,
cancel,
retries,
&trace,
true,
)
.await
} else {
call_oai_stream_inner(
&cfg,
&broker,
&[],
&None,
&[],
&tx,
None,
None,
0,
agent_core::reasoning::ReasoningLevel::Medium,
cancel,
&trace,
true,
suppressed,
retries,
)
.await
};
let mut events = vec![];
while let Ok(event) = rx.try_recv() {
events.push(event);
}
(result, events)
}
#[tokio::test(start_paused = true)]
async fn broker_transport_recovery_preserves_usage_and_identical_bytes() {
for xai in [false, true] {
let broker = Arc::new(Broker {
bodies: vec![
(
if xai { XAI_PARTIAL } else { CHAT_PARTIAL },
Some(BrokerError::Transport(
"unexpected EOF during chunk size line".into(),
)),
),
(if xai { XAI_OK } else { CHAT_OK }, None),
],
seen: Mutex::new(vec![]),
});
let (result, events) =
drive(broker.clone(), xai, 1, false, &CancellationToken::new()).await;
assert_eq!(result.unwrap()["content"][0]["text"], "accepted");
let seen = broker.seen.lock().unwrap();
assert_eq!(seen.len(), 2);
assert_eq!(seen[0].body_bytes, seen[1].body_bytes);
assert_eq!(
events
.iter()
.filter(|e| matches!(e, StreamEvent::Llm(crate::LlmEvent::ResponseReset)))
.count(),
1
);
if !xai {
assert_eq!(
events
.iter()
.filter_map(|e| match e {
StreamEvent::Session(crate::SessionEvent::Usage {
input_tokens,
..
}) => Some(*input_tokens),
_ => None,
})
.sum::<u64>(),
7
);
}
}
}
#[tokio::test(start_paused = true)]
async fn broker_nontransport_denial_never_retries_and_precancel_never_dispatches() {
for xai in [false, true] {
let broker = Arc::new(Broker {
bodies: vec![(
if xai { XAI_PARTIAL } else { CHAT_PARTIAL },
Some(BrokerError::Denied("blocked".into())),
)],
seen: Mutex::new(vec![]),
});
let (result, events) =
drive(broker.clone(), xai, 10, false, &CancellationToken::new()).await;
assert!(result.is_err());
assert_eq!(broker.seen.lock().unwrap().len(), 1);
assert!(!events
.iter()
.any(|e| matches!(e, StreamEvent::Llm(crate::LlmEvent::ResponseReset))));
let canceled = CancellationToken::new();
canceled.cancel();
assert!(drive(broker.clone(), xai, 10, false, &canceled)
.await
.0
.is_err());
assert_eq!(broker.seen.lock().unwrap().len(), 1);
}
}
#[tokio::test(start_paused = true)]
async fn suppressed_retry_emits_no_llm_previews() {
let broker = Arc::new(Broker {
bodies: vec![
(CHAT_PARTIAL, Some(BrokerError::Transport("eof".into()))),
(CHAT_OK, None),
],
seen: Mutex::new(vec![]),
});
let (result, events) = drive(broker, false, 1, true, &CancellationToken::new()).await;
assert!(result.is_ok());
assert!(!events.iter().any(|e| matches!(e, StreamEvent::Llm(_))));
}
#[test]
fn exhaustion_is_transient_for_auto_driver_but_not_arbitrary_error_prose() {
use crate::extensions::session_driver::{classify_error, Outcome};
assert_eq!(
classify_error(STREAM_INTERRUPTED),
(Outcome::ProviderError, "transient".into())
);
assert_eq!(
classify_error(&format!("API error: {STREAM_INTERRUPTED}")),
(Outcome::ProviderError, "transient".into())
);
assert_eq!(
classify_error(&format!("denied: {STREAM_INTERRUPTED}")).0,
Outcome::Blocked
);
}
}
#[cfg(test)]
mod muse_tests {
//! Meta Muse Responses route: exact body shape (mirrors the official
//! Muse Code client) and an end-to-end turn through a capturing broker.
use super::*;
use crate::auth::{
AccessToken, BrokerError, CredentialBroker, ProxyByteStream, ProxyMethod, ProxyRequest,
ProxyResponse,
};
use agent_core::reasoning::ReasoningLevel;
use std::sync::{Arc, Mutex};
#[test]
fn body_mirrors_official_client_and_adaptive_omits_reasoning() {
let input = json!([{"role":"user","content":[{"type":"input_text","text":"hi"}]}]);
let tool = json!({"type":"function","name":"bash","parameters":{"type":"object"}});
let key = muse_prompt_cache_key("sys", input.as_array().unwrap().first());
let body = build_muse_body(
"muse-wonder-1.0",
ReasoningLevel::Adaptive,
input.clone(),
"sys",
vec![tool.clone()],
Some(2048),
&key,
)
.unwrap();
assert_eq!(body["model"], "muse-wonder-1.0");
assert_eq!(body["instructions"], "sys");
assert_eq!(body["input"], input);
assert_eq!(body["store"], false);
assert_eq!(body["stream"], true);
assert_eq!(body["tools"], json!([tool]));
assert_eq!(body["max_output_tokens"], 2048);
assert_eq!(body["prompt_cache_key"], key.as_str());
assert!(
body.get("reasoning").is_none(),
"adaptive = provider default"
);
assert!(body.get("messages").is_none());
assert!(body.get("temperature").is_none());
// Empty prompt/tools/limit are omitted, not sent empty.
let bare = build_muse_body(
"muse-wonder-1.0",
ReasoningLevel::Adaptive,
json!([]),
"",
Vec::new(),
None,
&key,
)
.unwrap();
for absent in ["instructions", "tools", "max_output_tokens", "reasoning"] {
assert!(bare.get(absent).is_none(), "{absent}");
}
}
#[test]
fn named_effort_requires_exact_live_evidence_before_any_io() {
// Held for the whole test: the seed assertions read the shared
// process-global `muse/*` cache rows too.
let _guard = crate::runtime::openai::catalog::muse::test_cache_guard();
crate::runtime::openai::catalog::capability_cache::replace_provider("muse", &[]);
// The static seed advertises no effort set: named levels are refused
// in the pure builder, i.e. before credentials or network.
for level in [
ReasoningLevel::Off,
ReasoningLevel::High,
ReasoningLevel::Max,
] {
let err = build_muse_body(
"muse-wonder-1.0",
level,
json!([]),
"",
Vec::new(),
None,
"k",
)
.unwrap_err();
assert!(err.contains("adaptive"), "{level}: {err}");
}
// Uncataloged-but-routable id: provider default only; malformed: none.
assert!(build_muse_body(
"never-cataloged-zz9",
ReasoningLevel::Adaptive,
json!([]),
"",
Vec::new(),
None,
"k"
)
.is_ok());
assert!(build_muse_body(
"Bad Id",
ReasoningLevel::Adaptive,
json!([]),
"",
Vec::new(),
None,
"k"
)
.is_err());
let models = crate::runtime::openai::catalog::parse_muse_catalog_models(
r#"{"models":[{"id":"muse-body-probe-zz2","reasoning_effort_variants":
[{"effort":"none"},{"effort":"xhigh"}]}]}"#,
)
.unwrap();
crate::runtime::openai::catalog::capability_cache::replace_provider("muse", &models);
for (level, effort) in [
(ReasoningLevel::Off, "none"),
(ReasoningLevel::XHigh, "xhigh"),
] {
let body = build_muse_body(
"muse-body-probe-zz2",
level,
json!([]),
"",
Vec::new(),
None,
"k",
)
.unwrap();
assert_eq!(body["reasoning"], json!({"effort": effort}));
}
assert!(build_muse_body(
"muse-body-probe-zz2",
ReasoningLevel::High,
json!([]),
"",
Vec::new(),
None,
"k"
)
.is_err());
crate::runtime::openai::catalog::capability_cache::replace_provider("muse", &[]);
}
#[test]
fn cache_key_is_deterministic_namespaced_and_prefix_sensitive() {
let first = json!({"role":"user","content":"a"});
let k1 = muse_prompt_cache_key("sys", Some(&first));
assert_eq!(k1, muse_prompt_cache_key("sys", Some(&first)));
assert!(k1.starts_with("synaps-muse-"), "{k1}");
assert_ne!(k1, codex_prompt_cache_key("sys", Some(&first)));
assert_ne!(k1, muse_prompt_cache_key("sys2", Some(&first)));
assert_ne!(
k1,
muse_prompt_cache_key("sys", Some(&json!({"role":"user","content":"b"})))
);
}
struct CaptureBroker {
sse: &'static str,
seen: Mutex<Vec<ProxyRequest>>,
}
#[async_trait::async_trait]
impl CredentialBroker for CaptureBroker {
async fn access_token(
&self,
_: crate::auth::OAuthProviderId,
) -> Result<AccessToken, BrokerError> {
unreachable!("the runtime never holds the Muse key")
}
async fn proxy(&self, _: ProxyRequest) -> Result<ProxyResponse, BrokerError> {
unreachable!()
}
async fn proxy_stream(
&self,
request: ProxyRequest,
) -> Result<ProxyByteStream, BrokerError> {
self.seen.lock().unwrap().push(request);
Ok(Box::pin(futures::stream::iter(vec![Ok(
bytes::Bytes::from_static(self.sse.as_bytes()),
)])))
}
async fn anthropic_usage(&self) -> Result<Value, BrokerError> {
unreachable!()
}
async fn capabilities(&self) -> Result<Vec<agent_core::auth::ProviderStatus>, BrokerError> {
Ok(vec![])
}
}
const SSE: &str = concat!(
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"Hel\"}\n\n",
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"lo\"}\n\n",
"data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",",
"\"call_id\":\"call_1\",\"name\":\"bash\",\"arguments\":\"{\\\"command\\\":\\\"ls\\\"}\"}}\n\n",
"data: {\"type\":\"response.completed\",\"response\":{\"usage\":",
"{\"input_tokens\":12,\"output_tokens\":3}}}\n\n",
);
#[tokio::test]
async fn turn_streams_through_broker_with_system_prompt_as_instructions() {
let broker = Arc::new(CaptureBroker {
sse: SSE,
seen: Mutex::new(Vec::new()),
});
let dyn_broker: Arc<dyn CredentialBroker> = broker.clone();
let cfg = ProviderConfig {
base_url: crate::auth::muse::API_BASE_URL.into(),
model: "muse-wonder-1.0".into(),
provider: "muse".into(),
};
let tools = vec![json!({
"name":"bash","description":"run","input_schema":{"type":"object","properties":{}}
})];
let messages: Vec<crate::SharedMessage> =
vec![Arc::new(json!({"role":"user","content":"hi muse"}))];
let (tx, mut rx) = mpsc::unbounded_channel::<StreamEvent>();
let cancel = tokio_util::sync::CancellationToken::new();
let result = call_responses_route_inner(
"muse",
&cfg,
&dyn_broker,
&tools,
&Some("You are Synaps.".into()),
&messages,
&tx,
Some(4096),
ReasoningLevel::Adaptive,
&cancel,
0,
&crate::runtime::trace::TraceContext::disabled(),
true,
)
.await
.expect("muse turn completes");
// Assistant text + translated tool call come back Anthropic-shaped.
let content = result["content"].as_array().unwrap();
assert_eq!(content[0]["text"], "Hello");
let tool_use = content
.iter()
.find(|block| block["type"] == "tool_use")
.expect("tool call surfaced");
assert_eq!(tool_use["name"], "bash");
assert_eq!(tool_use["input"]["command"], "ls");
// Exactly one broker-proxied POST to the pinned Responses path.
let seen = broker.seen.lock().unwrap();
assert_eq!(seen.len(), 1);
let request = &seen[0];
assert_eq!(request.provider, "muse");
assert_eq!(request.method, ProxyMethod::Post);
assert_eq!(request.path, "/v1/responses");
assert!(request.stream);
let body = request.body.as_ref().unwrap();
assert_eq!(body["model"], "muse-wonder-1.0");
assert_eq!(body["instructions"], "You are Synaps.");
assert_eq!(body["store"], false);
assert_eq!(body["max_output_tokens"], 4096);
assert!(body["prompt_cache_key"]
.as_str()
.unwrap()
.starts_with("synaps-muse-"));
// System prompt is NOT duplicated into the input items.
let wire = body["input"].to_string();
assert!(!wire.contains("You are Synaps."), "{wire}");
assert!(wire.contains("hi muse"), "{wire}");
assert_eq!(body["tools"][0]["name"], "bash");
// Exact-bytes handoff matches the structured body.
let bytes = request.body_bytes.as_ref().expect("exact wire bytes");
assert_eq!(
serde_json::from_slice::<Value>(bytes).unwrap(),
body.clone()
);
drop(tx);
let mut streamed = String::new();
while let Ok(event) = rx.try_recv() {
if let StreamEvent::Llm(crate::runtime::types::LlmEvent::Text(text)) = event {
streamed.push_str(&text);
}
}
assert_eq!(streamed, "Hello");
}
#[test]
fn dangling_preamble_trigger_cases() {
// (i) Muse's own commentary label with no tool call.
assert_eq!(
muse_dangling_preamble(false, Some("commentary"), "Now checking prod logs."),
Some("commentary_phase")
);
// (ii) short final text ending in ':' (the observed failures).
assert_eq!(
muse_dangling_preamble(
false,
None,
"Key facts assembled. Blocking question before I build anything:\n"
),
Some("colon_preamble")
);
assert_eq!(
muse_dangling_preamble(
false,
Some("final_answer"),
"Now checking whether prod logs show the error:"
),
Some("colon_preamble")
);
// Never when the response acted.
assert_eq!(
muse_dangling_preamble(true, Some("commentary"), "Running:"),
None
);
// Real answers: no colon, empty, or long content that ends in ':'.
assert_eq!(muse_dangling_preamble(false, None, "All done."), None);
assert_eq!(muse_dangling_preamble(false, None, " "), None);
assert_eq!(
muse_dangling_preamble(false, Some("final_answer"), "Which format?"),
None
);
let long = format!("{}:", "x".repeat(PREAMBLE_MAX_CHARS));
assert_eq!(muse_dangling_preamble(false, None, &long), None);
let edge = format!("{}:", "x".repeat(PREAMBLE_MAX_CHARS - 1));
assert_eq!(
muse_dangling_preamble(false, None, &edge),
Some("colon_preamble")
);
}
#[test]
fn preamble_nudge_fires_at_most_once_per_turn() {
let flagged =
json!({"role":"assistant","content":[],DANGLING_PREAMBLE_KEY:"colon_preamble"});
let clean = json!({"role":"assistant","content":[]});
let mut nudged = false;
assert_eq!(take_preamble_nudge(&clean, &mut nudged, false, true), None);
assert!(!nudged, "a clean response must not consume the one nudge");
assert_eq!(
take_preamble_nudge(&flagged, &mut nudged, false, true).as_deref(),
Some("colon_preamble")
);
assert!(nudged);
// Second dangling preamble in the same turn: the turn ends normally.
assert_eq!(
take_preamble_nudge(&flagged, &mut nudged, false, true),
None
);
// Cancelled rounds and the disable switch never nudge.
let mut fresh = false;
assert_eq!(take_preamble_nudge(&flagged, &mut fresh, true, true), None);
assert_eq!(
take_preamble_nudge(&flagged, &mut fresh, false, false),
None
);
assert!(!fresh);
}
#[test]
fn preamble_nudge_env_switch_parses_off_values_only() {
assert!(preamble_nudge_enabled_value(None));
assert!(preamble_nudge_enabled_value(Some("1")));
assert!(preamble_nudge_enabled_value(Some("")));
for off in ["0", "false", "OFF", " no "] {
assert!(!preamble_nudge_enabled_value(Some(off)), "{off}");
}
}
async fn muse_turn(sse: &'static str, messages: Vec<Value>) -> (Value, Value) {
let broker = Arc::new(CaptureBroker {
sse,
seen: Mutex::new(Vec::new()),
});
let dyn_broker: Arc<dyn CredentialBroker> = broker.clone();
let cfg = ProviderConfig {
base_url: crate::auth::muse::API_BASE_URL.into(),
model: "muse-wonder-1.0".into(),
provider: "muse".into(),
};
let tools = vec![json!({
"name":"bash","description":"run","input_schema":{"type":"object","properties":{}}
})];
let messages: Vec<crate::SharedMessage> = messages.into_iter().map(Arc::new).collect();
let (tx, _rx) = mpsc::unbounded_channel::<StreamEvent>();
let result = call_responses_route_inner(
"muse",
&cfg,
&dyn_broker,
&tools,
&None,
&messages,
&tx,
None,
ReasoningLevel::Adaptive,
&tokio_util::sync::CancellationToken::new(),
0,
&crate::runtime::trace::TraceContext::disabled(),
true,
)
.await
.expect("muse turn completes");
let body = broker.seen.lock().unwrap()[0].body.clone().unwrap();
(result, body)
}
const PREAMBLE_SSE: &str = concat!(
"data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":",
"{\"type\":\"message\",\"role\":\"assistant\",\"content\":[]}}\n\n",
"data: {\"type\":\"response.output_text.delta\",\"output_index\":0,",
"\"delta\":\"Blocking question before I build anything:\"}\n\n",
"data: {\"type\":\"response.completed\",\"response\":{\"usage\":",
"{\"input_tokens\":12,\"output_tokens\":3}}}\n\n",
);
#[tokio::test]
async fn muse_route_flags_dangling_preamble_outside_content() {
let (result, _) =
muse_turn(PREAMBLE_SSE, vec![json!({"role":"user","content":"go"})]).await;
assert_eq!(result[DANGLING_PREAMBLE_KEY], "colon_preamble");
// Stored content stays Anthropic-shaped and carries no sidecar/phase.
assert_eq!(
result["content"],
json!([{"type":"text","text":"Blocking question before I build anything:"}])
);
// A response that acted is not flagged.
let (result, _) = muse_turn(SSE, vec![json!({"role":"user","content":"go"})]).await;
assert!(result.get(DANGLING_PREAMBLE_KEY).is_none(), "{result}");
}
#[tokio::test]
async fn muse_request_replays_assistant_text_with_phase() {
let (_, body) = muse_turn(
SSE,
vec![
json!({"role":"user","content":"go"}),
json!({"role":"assistant","content":[
{"type":"text","text":"Checking:"},
{"type":"tool_use","id":"call_a","name":"bash","input":{}},
]}),
json!({"role":"user","content":[
{"type":"tool_result","tool_use_id":"call_a","content":"ok"},
]}),
json!({"role":"assistant","content":[{"type":"text","text":"Done."}]}),
json!({"role":"user","content":"next"}),
],
)
.await;
let input = body["input"].as_array().unwrap();
assert_eq!(input[1]["phase"], "commentary");
assert_eq!(input[1]["content"][0]["text"], "Checking:");
assert_eq!(input[2]["type"], "function_call");
assert_eq!(input[4]["phase"], "final_answer");
assert_eq!(input[4]["content"][0]["type"], "output_text");
}
}