diff --git a/CHANGELOG.md b/CHANGELOG.md index 71f63f6287..d572648c7b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ ## [v0.15.2] - 2026-11-05 -BiBCode v0.15.2 fixes chat and diff interactions, provider compatibility, and diagnostics. This release includes the fixes from PR #308 since v0.15.1. +BiBCode v0.15.2 fixes chat and diff interactions, provider compatibility, and diagnostics. This release includes the fixes from PR #308, PR #314, and issue #316 since v0.15.1. ### Chats and diffs @@ -12,6 +12,7 @@ BiBCode v0.15.2 fixes chat and diff interactions, provider compatibility, and di ### Providers +- Codex no longer requests the MCP catalogue that can exceed its chat transport limit on new or resumed threads. Oversized stdout failures during launch stop automatic retries and preserve Retry/Dismiss recovery. MCP status now reflects startup notifications, so servers without notifications may be absent. (#316) - The Codex skill picker includes skills installed through plugins, including Superpowers. (#298) - Codex stderr logs no longer appear as raw **Codex warning** rows with ANSI escapes and incorrect severity. (#299) - The Claude model picker shows model names rather than taglines with Claude Code 2.1.318. (#300) @@ -24,6 +25,7 @@ BiBCode v0.15.2 fixes chat and diff interactions, provider compatibility, and di ### Regression coverage +- The Unix PTY setup-failure test waits for a complete descendant PID before checking process-group cleanup. The native macOS arm64 CI build has a four-hour job budget, matching x64. (#314) - Changed files coverage now clicks **View diff** through the chat and verifies that the selected diff stays open through same-turn plan updates. (#275) - The packaged resume-at-end check follows the active timeline through turn completion, avoiding stale scrollers and replies from history. (#215) diff --git a/apps/server/src/diagnostics/profiler/provider_turns.rs b/apps/server/src/diagnostics/profiler/provider_turns.rs index 3e8141bd80..40a34667c4 100644 --- a/apps/server/src/diagnostics/profiler/provider_turns.rs +++ b/apps/server/src/diagnostics/profiler/provider_turns.rs @@ -96,7 +96,8 @@ fn turn_failure_finding( /// Records a message delivery that failed for good without reaching its provider as a /// `delivery.failing` finding: the provider, the delivery's `attempts` and the `error_class` -/// (`provider-configuration` or `attempts-exhausted`). Never the error text, a path or a title. +/// (`provider-configuration`, `transport-line-too-long` or `attempts-exhausted`). Never error text, +/// a path or a title. /// Does nothing while the profiler is off. pub fn record_delivery_failing(provider: &str, attempts: i64, error_class: &str) { #[cfg(test)] diff --git a/apps/server/src/production/provider_runtime.rs b/apps/server/src/production/provider_runtime.rs index ba16f9d2f6..95c196e789 100644 --- a/apps/server/src/production/provider_runtime.rs +++ b/apps/server/src/production/provider_runtime.rs @@ -362,6 +362,10 @@ pub enum ProviderReconciliationOutcome { ProviderConfiguration { detail: String, }, + /// An oversized transport close prevents reconciliation; the original send stays uncertain. + TransportLineTooLong { + detail: String, + }, /// The reconciliation launch failed before it reached the provider, such as an app-server /// that exits before `initialize`. Restart recovery tries again with backoff, and counts /// these against its launch budget so a failure that never clears cannot relaunch forever. @@ -612,6 +616,9 @@ pub enum ProviderRuntimeError { MissingExecutable { provider: String, binary: String }, #[error("{provider} provider operation failed: {detail}")] Provider { provider: String, detail: String }, + /// Retrying the same launch cannot make an oversized transport line fit. + #[error("{provider} provider operation failed: {detail}")] + TransportLineTooLong { provider: String, detail: String }, /// Only absent/cursorless matching runtime state; conflicting identities remain rejected. #[error("{provider} provider operation failed: {detail}")] FrozenSessionUnavailable { provider: String, detail: String }, @@ -2099,6 +2106,14 @@ async fn deliver_orchestration_turn_with_identity( Err(error) => { let attempts = frozen_delivery.as_ref().map_or(0, |row| row.attempts); note_provider_configuration_refusal(&error, attempts); + if let ProviderRuntimeError::TransportLineTooLong { provider, .. } = &error { + tracing::warn!(%provider, attempts, "transport line too long; failing the message delivery"); + crate::diagnostics::profiler::provider_turns::record_delivery_failing( + provider, + attempts, + "transport-line-too-long", + ); + } return launch_failure_outcome(&error, &label); } } @@ -2172,15 +2187,15 @@ fn frozen_failure_outcome( /// A durable turn is replayed unchanged, so a launch refused for the turn's own options fails the /// delivery once, and so does a provider this BiBCode cannot drive, whose executable is missing, -/// or that refuses its own configuration (each is refused again until its settings, installation -/// or configuration change). Every other launch failure, such as another spawn error, is retried -/// with backoff. +/// refuses its own configuration, or exceeds its transport line bound. Every other launch +/// failure, such as another spawn error, is retried with backoff. fn launch_failure_outcome(error: &ProviderRuntimeError, label: &str) -> ProviderDeliveryOutcome { let detail = delivery_detail(error, Some(label)); match error { ProviderRuntimeError::InvalidOption { .. } => ProviderDeliveryOutcome::Refused { detail }, ProviderRuntimeError::IncompatibleProvider { .. } | ProviderRuntimeError::MissingExecutable { .. } + | ProviderRuntimeError::TransportLineTooLong { .. } | ProviderRuntimeError::ProviderConfiguration { .. } => { ProviderDeliveryOutcome::Rejected { detail } } @@ -2192,6 +2207,75 @@ fn launch_failure_outcome(error: &ProviderRuntimeError, label: &str) -> Provider mod delivery_failure_tests { use super::*; + #[test] + fn transport_line_too_long_rejects_direct_and_nested_launch_errors() { + use crate::provider::{codex::protocol::ProtocolError, cursor::acp::AcpProtocolError}; + + for provider in ["codex", "cursor", "grok"] { + let detail = if provider == "codex" { + ProtocolError::LineTooLong { + stream: "stdout", + maximum: 8 * 1024 * 1024, + } + .to_string() + } else { + AcpProtocolError::LineTooLong { + stream: "stdout", + maximum: 128 * 1024, + } + .to_string() + }; + let closed = if provider == "codex" { + ProtocolError::Closed { + reason: detail.clone(), + } + .to_string() + } else { + AcpProtocolError::Closed { + reason: detail.clone(), + } + .to_string() + }; + for detail in [detail, closed] { + let error = provider_error(provider)(detail.clone()); + assert_eq!( + launch_failure_outcome(&error, "Provider"), + ProviderDeliveryOutcome::Rejected { + detail: detail.clone() + } + ); + assert_eq!( + frozen_failure_outcome(&error, None), + ProviderDeliveryOutcome::Rejected { detail } + ); + } + } + } + + #[test] + fn transport_size_classification_does_not_scrape_remote_request_messages() { + for (provider, detail) in [ + ("codex", "Codex JSON-RPC connection is closed: stdout ended"), + ( + "codex", + "Codex JSON-RPC request thread/resume failed (-32000): read line exceeded a remote limit", + ), + ( + "claudeAgent", + "Codex JSON-RPC read line exceeded 64 bytes on stdout", + ), + ( + "opencode", + "ACP JSON-RPC read line exceeded 64 bytes on stdout", + ), + ] { + assert!(matches!( + launch_failure_outcome(&provider_error(provider)(detail), "Provider"), + ProviderDeliveryOutcome::DefinitelyNotSent { .. } + )); + } + } + #[test] fn invalid_options_are_refused_with_the_same_delivery_detail() { let error = ProviderRuntimeError::InvalidOption { @@ -2278,6 +2362,7 @@ pub(crate) fn delivery_detail( // Drivers and BiBCode's own delivery ordering both raise this one, so its detail, without // the "{driver id} provider operation failed:" prefix, is what it says. ProviderRuntimeError::Provider { detail, .. } + | ProviderRuntimeError::TransportLineTooLong { detail, .. } | ProviderRuntimeError::FrozenSessionUnavailable { detail, .. } | ProviderRuntimeError::IncompatibleProvider { detail, .. } => detail.clone(), ProviderRuntimeError::Spawn { provider, detail } => { @@ -2531,6 +2616,15 @@ pub async fn reconcile_orchestration_turn( detail: delivery_detail(&error, Some(&label)), }; } + Err(ProviderRuntimeError::TransportLineTooLong { provider, detail }) => { + tracing::warn!(%provider, attempts = row.attempts, "transport line too long; stopping delivery recovery"); + crate::diagnostics::profiler::provider_turns::record_delivery_failing( + &provider, + row.attempts, + "transport-line-too-long", + ); + return ProviderReconciliationOutcome::TransportLineTooLong { detail }; + } // A launch that fails, such as an app-server that exits before `initialize`, // counts against restart recovery's launch budget. Err(error) => { @@ -3281,6 +3375,172 @@ mod workspace_loss_tests { .unwrap() } + #[tokio::test] + async fn oversized_stdout_delivery_stops_after_one_launch() { + assert_oversized_stdout_delivery_stops_after_one_launch(TurnDeliveryState::Failed).await; + } + + #[tokio::test] + async fn oversized_stdout_restart_recovery_stops_after_one_launch() { + assert_oversized_stdout_delivery_stops_after_one_launch(TurnDeliveryState::Uncertain).await; + } + + async fn assert_oversized_stdout_delivery_stops_after_one_launch( + expected_state: TurnDeliveryState, + ) { + use crate::production::turn_delivery::TurnDeliveryService; + use crate::provider::codex::protocol::{ + ConnectionConfig, IncomingEvent, JsonRpcConnection, + }; + + struct OversizedLaunch(AtomicUsize); + impl ProviderDriverFactory for OversizedLaunch { + fn create( + &self, + _: ProviderLaunchRequest, + ) -> BoxRuntimeFuture<'_, Result, ProviderRuntimeError>> + { + Box::pin(async move { + self.0.fetch_add(1, Ordering::SeqCst); + let (connection, mut events) = JsonRpcConnection::spawn( + std::io::Cursor::new(vec![b'x'; 65]), + tokio::io::sink(), + tokio::io::empty(), + ConnectionConfig { + max_stdout_line_bytes: 64, + ..ConnectionConfig::default() + }, + ); + assert!(matches!( + events.recv().await, + Some(IncomingEvent::Closed { .. }) + )); + let error = connection + .request("initialize", json!({})) + .await + .unwrap_err(); + Err(codex_start_error(error.into(), "unused-config")) + }) + } + } + let f = Fixture::new(false).await; + let factory = Arc::new(OversizedLaunch(AtomicUsize::new(0))); + let supervisor = Arc::new(ProviderRuntimeSupervisor::start( + f.engine.clone(), + factory.clone(), + ActivityProjection::new(ActivityRepository::new( + f.engine.repositories().database().clone(), + )), + SupervisorOptions::default(), + )); + let command = turn("oversized", false); + let mut payload = serde_json::to_value(&command).unwrap(); + freeze_delivery_route( + &f.engine, + &f._root.path().to_path_buf(), + &command, + &mut payload, + ) + .await + .unwrap(); + f.engine + .dispatch_with_admission( + command.clone(), + CommandAdmission { + payload_digest: canonical_command_digest(&command).unwrap(), + attachment_refs: Vec::new(), + provider_turn: Some(NewProviderTurnDelivery { + command_id: "oversized".into(), + thread_id: "t1".into(), + message_id: "oversized".into(), + provider_instance_id: "codex".into(), + provider_kind: "codex".into(), + provider_session_id: None, + delivery_key: "oversized-key".into(), + payload, + state: TurnDeliveryState::Pending, + mode: TurnDeliveryMode::Start, + created_at: BEFORE_LOSS.into(), + }), + }, + || {}, + ) + .await + .unwrap(); + if expected_state == TurnDeliveryState::Uncertain { + f.engine + .repositories() + .claim_provider_turn("oversized".into(), BEFORE_LOSS.into()) + .await + .unwrap() + .expect("leave the delivery sending before restart recovery"); + } + crate::diagnostics::profiler::provider_turns::take_recorded(); + let service = TurnDeliveryService::start( + f.engine.clone(), + supervisor.clone(), + f._root.path().to_path_buf(), + ); + timeout(Duration::from_secs(5), async { + loop { + let row = f + .engine + .repositories() + .get_provider_turn_delivery("oversized".into()) + .await + .unwrap() + .unwrap(); + if row.state == expected_state { + assert_eq!(row.attempts, 1); + let detail = row.last_error.unwrap(); + assert!(detail.contains( + "connection is closed: Codex JSON-RPC read line exceeded 64 bytes on stdout" + ), "{detail}"); + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("deterministic transport failure rejects on its first attempt"); + service.wake(); + tokio::time::sleep(Duration::from_millis(200)).await; + service.shutdown().await; + if expected_state == TurnDeliveryState::Uncertain { + let recovered = TurnDeliveryService::start( + f.engine.clone(), + supervisor.clone(), + f._root.path().to_path_buf(), + ); + tokio::time::sleep(Duration::from_millis(200)).await; + recovered.shutdown().await; + assert_eq!( + f.engine + .repositories() + .get_provider_turn_delivery("oversized".into()) + .await + .unwrap() + .unwrap() + .state, + TurnDeliveryState::Uncertain, + ); + } + assert_eq!(factory.0.load(Ordering::SeqCst), 1); + let findings: Vec<_> = crate::diagnostics::profiler::provider_turns::take_recorded() + .into_iter() + .filter(|(kind, _)| *kind == "delivery.failing") + .collect(); + assert_eq!( + findings, + [( + "delivery.failing", + json!({"provider":"codex", "attempts":1, "errorClass":"transport-line-too-long"}) + )] + ); + supervisor.shutdown().await.unwrap(); + f.shutdown().await; + } + /// A manual compaction of the fixture's live Codex conversation. async fn request_compaction(f: &Fixture) -> String { f.project("ready").await; @@ -8818,6 +9078,7 @@ fn normalize_agent_activity_transition_error(error: ProviderRuntimeError) -> Pro } ProviderRuntimeError::Provider { .. } | ProviderRuntimeError::InvalidOption { .. } + | ProviderRuntimeError::TransportLineTooLong { .. } | ProviderRuntimeError::FrozenSessionUnavailable { .. } | ProviderRuntimeError::IncompatibleProvider { .. } | ProviderRuntimeError::ProviderConfiguration { .. } => "provider operation failure", @@ -19872,9 +20133,34 @@ fn pipe_error(provider: &str, stream: &str) -> ProviderRuntimeError { fn provider_error( provider: &str, ) -> impl FnOnce(E) -> ProviderRuntimeError + '_ { - move |error| ProviderRuntimeError::Provider { - provider: provider.to_owned(), - detail: error.to_string(), + move |error| { + let detail = error.to_string(); + // The protocol nests LineTooLong's Display in Closed before returning a pending request. + let line_too_long = + match provider { + "codex" => detail.starts_with("Codex JSON-RPC read line exceeded ") + || detail.starts_with( + "Codex JSON-RPC connection is closed: Codex JSON-RPC read line exceeded ", + ), + "cursor" | "grok" => { + detail.starts_with("ACP JSON-RPC read line exceeded ") + || detail.starts_with( + "ACP JSON-RPC connection is closed: ACP JSON-RPC read line exceeded ", + ) + } + _ => false, + }; + if line_too_long { + ProviderRuntimeError::TransportLineTooLong { + provider: provider.to_owned(), + detail, + } + } else { + ProviderRuntimeError::Provider { + provider: provider.to_owned(), + detail, + } + } } } diff --git a/apps/server/src/production/turn_delivery.rs b/apps/server/src/production/turn_delivery.rs index 6f9a6a8f82..061153cf8d 100644 --- a/apps/server/src/production/turn_delivery.rs +++ b/apps/server/src/production/turn_delivery.rs @@ -790,10 +790,11 @@ async fn settle_dismissed_once( | ProviderReconciliationOutcome::LaunchFailed { .. } => Ok(true), // A turn this runtime still holds, or a server that cannot report its status: // the marker stays until a later pass finds the session idle. - // A relaunch refused for the provider's configuration is refused again: stop asking. + // Configuration refusals and oversized transport closes repeat: stop asking. ProviderReconciliationOutcome::Found | ProviderReconciliationOutcome::ConversationLost { .. } - | ProviderReconciliationOutcome::ProviderConfiguration { .. } => Ok(false), + | ProviderReconciliationOutcome::ProviderConfiguration { .. } + | ProviderReconciliationOutcome::TransportLineTooLong { .. } => Ok(false), } }) .buffer_unordered(MAX_CONCURRENT_THREADS) @@ -961,7 +962,8 @@ async fn recover_sending( // Whether the original send arrived is unknown, and every relaunch is refused the same // way: the user's Retry or Dismiss decides. ProviderReconciliationOutcome::ConversationLost { detail } - | ProviderReconciliationOutcome::ProviderConfiguration { detail } => { + | ProviderReconciliationOutcome::ProviderConfiguration { detail } + | ProviderReconciliationOutcome::TransportLineTooLong { detail } => { (TurnDeliveryState::Uncertain, Some(detail)) } // A launch that never reached the provider is tried again, at most diff --git a/apps/server/src/provider/codex/mcp_status.rs b/apps/server/src/provider/codex/mcp_status.rs index 92a2b0899f..4bdc4ea8ad 100644 --- a/apps/server/src/provider/codex/mcp_status.rs +++ b/apps/server/src/provider/codex/mcp_status.rs @@ -1,10 +1,9 @@ use std::{ - collections::{BTreeMap, HashSet, VecDeque}, + collections::{BTreeMap, VecDeque}, future::Future, pin::Pin, sync::{Arc, Weak}, task::{Context, Poll}, - time::Duration, }; use serde::Serialize; @@ -12,12 +11,7 @@ use serde_json::Value; use tokio::sync::{mpsc, oneshot}; use tokio_util::sync::CancellationToken; -use super::protocol::JsonRpcConnection; - const PRE_ROOT_LIMIT: usize = 64; -pub(crate) const MCP_STATUS_PAGE_LIMIT: usize = 8; -pub(crate) const MCP_STATUS_PAGE_SIZE: u32 = 50; -pub(crate) const MCP_STATUS_REQUEST_TIMEOUT: Duration = Duration::from_secs(5); #[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] #[serde(rename_all = "kebab-case")] @@ -38,8 +32,6 @@ pub(crate) struct McpServerStatus { pub(crate) detail: Option, } -pub(crate) type McpLoadResult = Result, String>; - #[derive(Clone)] pub(crate) struct McpStatusHandle { sender: mpsc::Sender, @@ -94,13 +86,7 @@ impl McpOpenReservation { } pub(crate) enum McpStatusEffect { - Load { - epoch: u64, - generation: u64, - root: String, - }, Snapshot(Vec), - Warning(String), Complete(Vec>>), } @@ -119,11 +105,6 @@ pub(crate) enum McpStatusCommand { root: Option, server: McpServerStatus, }, - LoadFinished { - epoch: u64, - generation: u64, - result: McpLoadResult, - }, #[cfg(test)] SnapshotForTest { done: oneshot::Sender>, @@ -135,20 +116,13 @@ pub(crate) enum McpStatusCommand { struct McpStatusState { root: Option, - epoch: u64, - generation: u64, - in_flight: Option, + opening: Option, servers: BTreeMap, pre_root: VecDeque<(Option, McpServerStatus)>, } -struct InFlightRefresh { - epoch: u64, - generation: u64, - opening: bool, - opening_ownership: Option, - root_changed: bool, - overlay: BTreeMap, +struct Opening { + ownership: OpeningOwnership, waiters: Vec>>, } @@ -212,20 +186,6 @@ impl McpStatusHandle { .await } - pub(crate) async fn load_finished( - &self, - epoch: u64, - generation: u64, - result: McpLoadResult, - ) -> Result<(), String> { - self.send(McpStatusCommand::LoadFinished { - epoch, - generation, - result, - }) - .await - } - #[cfg(test)] pub(crate) async fn snapshot_for_test(&self) -> Result, String> { let (done_tx, done_rx) = oneshot::channel(); @@ -259,19 +219,16 @@ pub(crate) async fn run_actor( ) { let mut state = McpStatusState { root: None, - epoch: 0, - generation: 0, - in_flight: None, + opening: None, servers: BTreeMap::new(), pre_root: VecDeque::new(), }; loop { let opening_cancellation = state - .in_flight + .opening .as_ref() - .and_then(|in_flight| in_flight.opening_ownership.as_ref()) - .map(|ownership| ownership.cancellation.clone()); + .map(|opening| opening.ownership.cancellation.clone()); let command = if let Some(cancellation) = opening_cancellation { tokio::select! { biased; @@ -296,18 +253,13 @@ pub(crate) async fn run_actor( McpStatusCommand::Notification { root, server } => { notification(&mut state, root, server, &effects_tx); } - McpStatusCommand::LoadFinished { - epoch, - generation, - result, - } => finish_load(&mut state, epoch, generation, result, &effects_tx), #[cfg(test)] McpStatusCommand::SnapshotForTest { done } => { let _ = done.send(snapshot(&state.servers)); } McpStatusCommand::Shutdown { done } => { - if let Some(in_flight) = state.in_flight.take() { - for waiter in in_flight.waiters { + if let Some(opening) = state.opening.take() { + for waiter in opening.waiters { let _ = waiter.send(Err("MCP status actor shut down".to_owned())); } } @@ -323,62 +275,35 @@ fn begin_open( done: oneshot::Sender>, ownership: oneshot::Sender>, ) { - if let Some(owner) = state - .in_flight - .as_ref() - .filter(|in_flight| in_flight.opening) - .and_then(|in_flight| in_flight.opening_ownership.as_ref()) - .and_then(|ownership| ownership.owner.upgrade()) + if let Some(opening) = state.opening.as_mut() + && let Some(owner) = opening.ownership.owner.upgrade() { - state - .in_flight - .as_mut() - .expect("checked above") - .waiters - .push(done); + opening.waiters.push(done); let _ = ownership.send(owner); return; } - if state - .in_flight - .as_ref() - .is_some_and(|in_flight| in_flight.opening) - { - cancel_open(state, "MCP status opening was cancelled".to_owned()); - } - - let mut waiters = state - .in_flight - .take() - .map_or_else(Vec::new, |in_flight| in_flight.waiters); - waiters.push(done); - state.generation = state.generation.wrapping_add(1); + cancel_open(state, "MCP status opening was cancelled".to_owned()); state.pre_root.clear(); let cancellation = CancellationToken::new(); let owner = Arc::new(McpOpenOwnership { cancellation: cancellation.clone(), }); - state.in_flight = Some(InFlightRefresh { - epoch: state.epoch, - generation: state.generation, - opening: true, - opening_ownership: Some(OpeningOwnership { + state.opening = Some(Opening { + ownership: OpeningOwnership { owner: Arc::downgrade(&owner), cancellation, - }), - root_changed: false, - overlay: BTreeMap::new(), - waiters, + }, + waiters: vec![done], }); let _ = ownership.send(owner); } fn cancel_open(state: &mut McpStatusState, error: String) { - let Some(in_flight) = state.in_flight.take_if(|in_flight| in_flight.opening) else { + let Some(opening) = state.opening.take() else { return; }; state.pre_root.clear(); - for waiter in in_flight.waiters { + for waiter in opening.waiters { let _ = waiter.send(Err(error.clone())); } } @@ -388,72 +313,22 @@ fn bind_root( root: String, effects_tx: &mpsc::UnboundedSender, ) { - let changed = state.root.as_deref() != Some(root.as_str()); - if changed { - state.root = Some(root.clone()); - state.epoch = state.epoch.wrapping_add(1); + if state.opening.is_some() || state.root.as_deref() != Some(root.as_str()) { state.servers.clear(); } - - if changed - && state - .in_flight - .as_ref() - .is_some_and(|in_flight| !in_flight.opening) - { - let waiters = state.in_flight.take().expect("checked above").waiters; - state.generation = state.generation.wrapping_add(1); - state.pre_root.clear(); - let epoch = state.epoch; - let generation = state.generation; - state.in_flight = Some(InFlightRefresh { - epoch, - generation, - opening: false, - opening_ownership: None, - root_changed: true, - overlay: BTreeMap::new(), - waiters, - }); - dispatch_load(state, effects_tx, epoch, generation, root); - return; - } - - if state.in_flight.is_none() { - state.generation = state.generation.wrapping_add(1); - state.in_flight = Some(InFlightRefresh { - epoch: state.epoch, - generation: state.generation, - opening: true, - opening_ownership: None, - root_changed: changed, - overlay: BTreeMap::new(), - waiters: Vec::new(), - }); - } - - let staged = state + for (_, server) in state .pre_root .drain(..) .filter(|(notification_root, _)| matches_root(notification_root.as_deref(), &root)) - .collect::>(); - let (epoch, generation) = { - let Some(in_flight) = state.in_flight.as_mut() else { - return; - }; - if !in_flight.opening { - return; - } - in_flight.epoch = state.epoch; - in_flight.root_changed |= changed; - for (_, server) in staged { - in_flight.overlay.insert(server.name.clone(), server); - } - in_flight.opening = false; - in_flight.opening_ownership = None; - (in_flight.epoch, in_flight.generation) - }; - dispatch_load(state, effects_tx, epoch, generation, root); + { + state.servers.insert(server.name.clone(), server); + } + state.root = Some(root); + let waiters = state + .opening + .take() + .map_or_else(Vec::new, |opening| opening.waiters); + publish(state, waiters, effects_tx); } fn refresh( @@ -461,47 +336,28 @@ fn refresh( done: oneshot::Sender>, effects_tx: &mpsc::UnboundedSender, ) { - if let Some(in_flight) = state.in_flight.as_mut() { - in_flight.waiters.push(done); + if let Some(opening) = state.opening.as_mut() { + opening.waiters.push(done); return; } - let Some(root) = state.root.clone() else { + if state.root.is_none() { let _ = done.send(Err("MCP status root is not bound".to_owned())); return; - }; - state.generation = state.generation.wrapping_add(1); - let in_flight = InFlightRefresh { - epoch: state.epoch, - generation: state.generation, - opening: false, - opening_ownership: None, - root_changed: false, - overlay: BTreeMap::new(), - waiters: vec![done], - }; - let epoch = in_flight.epoch; - let generation = in_flight.generation; - state.in_flight = Some(in_flight); - dispatch_load(state, effects_tx, epoch, generation, root); + } + publish(state, vec![done], effects_tx); } -fn dispatch_load( - state: &mut McpStatusState, +fn publish( + state: &McpStatusState, + waiters: Vec>>, effects_tx: &mpsc::UnboundedSender, - epoch: u64, - generation: u64, - root: String, ) { - if effects_tx - .send(McpStatusEffect::Load { - epoch, - generation, - root, - }) - .is_err() - && let Some(in_flight) = state.in_flight.take() + // shortcut: status is notification-only; use discovery when Codex offers a bounded auth-only response. + let _ = effects_tx.send(McpStatusEffect::Snapshot(snapshot(&state.servers))); + if let Err(error) = effects_tx.send(McpStatusEffect::Complete(waiters)) + && let McpStatusEffect::Complete(waiters) = error.0 { - for waiter in in_flight.waiters { + for waiter in waiters { let _ = waiter.send(Err("MCP status effect receiver is unavailable".to_owned())); } } @@ -513,11 +369,7 @@ fn notification( server: McpServerStatus, effects_tx: &mpsc::UnboundedSender, ) { - if state - .in_flight - .as_ref() - .is_some_and(|in_flight| in_flight.opening) - { + if state.opening.is_some() { if state.pre_root.len() == PRE_ROOT_LIMIT { state.pre_root.pop_front(); } @@ -527,62 +379,15 @@ fn notification( let Some(bound_root) = state.root.as_deref() else { return; }; - if !matches_root(root.as_deref(), bound_root) { - return; - } - if let Some(in_flight) = state.in_flight.as_mut() { - in_flight.overlay.insert(server.name.clone(), server); - return; - } - if state.servers.get(&server.name) == Some(&server) { + if !matches_root(root.as_deref(), bound_root) + || state.servers.get(&server.name) == Some(&server) + { return; } state.servers.insert(server.name.clone(), server); let _ = effects_tx.send(McpStatusEffect::Snapshot(snapshot(&state.servers))); } -fn finish_load( - state: &mut McpStatusState, - epoch: u64, - generation: u64, - result: McpLoadResult, - effects_tx: &mpsc::UnboundedSender, -) { - if state - .in_flight - .as_ref() - .is_none_or(|in_flight| in_flight.epoch != epoch || in_flight.generation != generation) - { - return; - } - let in_flight = state.in_flight.take().expect("checked above"); - state.pre_root.clear(); - match result { - Ok(mut baseline) => { - baseline.extend(in_flight.overlay); - state.servers = baseline; - let _ = effects_tx.send(McpStatusEffect::Snapshot(snapshot(&state.servers))); - } - Err(error) => { - let mut visible = if in_flight.root_changed { - BTreeMap::new() - } else { - state.servers.clone() - }; - visible.extend(in_flight.overlay); - let changed = in_flight.root_changed || visible != state.servers; - state.servers = visible; - if changed { - let _ = effects_tx.send(McpStatusEffect::Snapshot(snapshot(&state.servers))); - } - let _ = effects_tx.send(McpStatusEffect::Warning(format!( - "MCP status discovery failed: {error}" - ))); - } - } - let _ = effects_tx.send(McpStatusEffect::Complete(in_flight.waiters)); -} - fn matches_root(notification_root: Option<&str>, root: &str) -> bool { notification_root.is_none_or(|notification_root| notification_root == root) } @@ -591,82 +396,6 @@ fn snapshot(servers: &BTreeMap) -> Vec servers.values().cloned().collect() } -pub(crate) async fn refresh_mcp_status_snapshot( - connection: &JsonRpcConnection, - root: &str, -) -> McpLoadResult { - let mut servers = BTreeMap::new(); - let mut cursor = None; - let mut seen_cursors = HashSet::new(); - for _ in 0..MCP_STATUS_PAGE_LIMIT { - let mut params = serde_json::json!({ - "threadId": root, - "limit": MCP_STATUS_PAGE_SIZE, - "detail": "toolsAndAuthOnly", - }); - if let Some(cursor) = cursor.as_ref() { - params["cursor"] = serde_json::json!(cursor); - } - let cancellation = CancellationToken::new(); - let request = connection.request_cancellable("mcpServerStatus/list", params, &cancellation); - tokio::pin!(request); - let response = match tokio::time::timeout(MCP_STATUS_REQUEST_TIMEOUT, &mut request).await { - Ok(result) => result.map_err(|error| error.to_string())?, - Err(_) => { - cancellation.cancel(); - let _ = request.await; - return Err("mcpServerStatus/list request timed out".to_owned()); - } - }; - let data = response - .get("data") - .and_then(Value::as_array) - .ok_or_else(|| "mcpServerStatus/list response missing data array".to_owned())?; - for server in data.iter().map(mcp_server_status_from_snapshot) { - let server = server?; - servers.insert(server.name.clone(), server); - } - match response.get("nextCursor") { - None | Some(Value::Null) => return Ok(servers), - Some(Value::String(next_cursor)) if !next_cursor.trim().is_empty() => { - if !seen_cursors.insert(next_cursor.clone()) { - return Err("mcpServerStatus/list response repeated nextCursor".to_owned()); - } - cursor = Some(next_cursor.clone()); - } - _ => return Err("mcpServerStatus/list response has invalid nextCursor".to_owned()), - } - } - Err("mcpServerStatus/list response exceeded page limit".to_owned()) -} - -pub(crate) fn mcp_server_status_from_snapshot(server: &Value) -> Result { - let name = server - .get("name") - .and_then(Value::as_str) - .map(str::trim) - .filter(|name| !name.is_empty()) - .ok_or_else(|| "mcpServerStatus/list row missing a non-empty name".to_owned())?; - let (state, detail) = match server.get("authStatus").and_then(Value::as_str) { - Some("unsupported" | "bearerToken" | "oAuth") => (McpServerState::Starting, None), - Some("notLoggedIn") => ( - McpServerState::NeedsAuth, - Some("Authentication required.".to_owned()), - ), - Some(other) => { - return Err(format!( - "mcpServerStatus/list row has unsupported authStatus {other}" - )); - } - None => return Err("mcpServerStatus/list row missing authStatus".to_owned()), - }; - Ok(McpServerStatus { - name: name.to_owned(), - state, - detail, - }) -} - pub(crate) fn mcp_server_status_from_notification(params: &Value) -> Option { mcp_server_status( params.get("name")?.as_str()?, @@ -705,82 +434,27 @@ pub(crate) fn mcp_server_status( #[cfg(test)] mod tests { - use std::collections::BTreeMap; - - use tokio::sync::mpsc; - use super::*; - fn server(name: &str, state: McpServerState) -> McpServerStatus { + fn server(name: &str) -> McpServerStatus { McpServerStatus { name: name.to_owned(), - state, + state: McpServerState::Connected, detail: None, } } - fn map(servers: [McpServerStatus; N]) -> BTreeMap { - servers - .into_iter() - .map(|server| (server.name.clone(), server)) - .collect() - } - - fn names(servers: &[McpServerStatus]) -> Vec<&str> { - servers.iter().map(|server| server.name.as_str()).collect() - } - - #[tokio::test] - async fn stages_notifications_until_the_opening_load_completes() { - // Mutation caught: committing an in-flight notification emits an incomplete snapshot. - let (handle, receiver) = McpStatusHandle::channel(64); - let (effects_tx, mut effects_rx) = mpsc::unbounded_channel(); - let actor = tokio::spawn(run_actor(receiver, effects_tx)); - - let opening = handle.begin_open().await.unwrap(); - handle - .notification( - Some("root-1".into()), - server("before", McpServerState::Connected), - ) - .await - .unwrap(); - handle.bind_root("root-1".into()).await.unwrap(); - - let McpStatusEffect::Load { - epoch, - generation, - root, - } = effects_rx.recv().await.unwrap() - else { - panic!("expected list load"); + async fn published(effects: &mut mpsc::UnboundedReceiver) -> Vec { + let McpStatusEffect::Snapshot(servers) = effects.recv().await.unwrap() else { + panic!("snapshot must precede completion"); }; - assert_eq!(root, "root-1"); - - handle - .notification( - Some("root-1".into()), - server("during", McpServerState::Starting), - ) - .await - .unwrap(); - assert!(effects_rx.try_recv().is_err()); - - handle - .load_finished( - epoch, - generation, - Ok(map([server("seed", McpServerState::Connected)])), - ) - .await - .unwrap(); - let McpStatusEffect::Snapshot(snapshot) = effects_rx.recv().await.unwrap() else { - panic!("expected complete snapshot"); + let McpStatusEffect::Complete(waiters) = effects.recv().await.unwrap() else { + panic!("expected completion"); }; - assert_eq!(names(&snapshot), vec!["before", "during", "seed"]); - drop(opening); - handle.shutdown().await.unwrap(); - actor.await.unwrap(); + for waiter in waiters { + waiter.send(Ok(())).unwrap(); + } + servers.into_iter().map(|server| server.name).collect() } #[tokio::test] @@ -788,344 +462,113 @@ mod tests { let (handle, receiver) = McpStatusHandle::channel(64); let (effects_tx, mut effects_rx) = mpsc::unbounded_channel(); let actor = tokio::spawn(run_actor(receiver, effects_tx)); - let opening = handle.begin_open().await.unwrap(); for index in 0..PRE_ROOT_LIMIT + 2 { handle .notification( - Some(if index % 2 == 0 { - "matching-root".to_owned() - } else { - "foreign-root".to_owned() - }), - server(&format!("server-{index:02}"), McpServerState::Connected), + Some( + if index % 2 == 0 { + "matching-root" + } else { + "foreign-root" + } + .into(), + ), + server(&format!("server-{index:02}")), ) .await .unwrap(); } handle.bind_root("matching-root".into()).await.unwrap(); - let McpStatusEffect::Load { - epoch, generation, .. - } = effects_rx.recv().await.unwrap() - else { - panic!("expected list load"); - }; - handle - .load_finished(epoch, generation, Ok(BTreeMap::new())) - .await - .unwrap(); - let McpStatusEffect::Snapshot(snapshot) = effects_rx.recv().await.unwrap() else { - panic!("expected complete snapshot"); - }; assert_eq!( - snapshot - .iter() - .map(|server| server.name.clone()) - .collect::>(), + published(&mut effects_rx).await, (2..=PRE_ROOT_LIMIT) .step_by(2) .map(|index| format!("server-{index:02}")) .collect::>() ); - - drop(opening); + assert_eq!(opening.await.unwrap(), Ok(())); handle.shutdown().await.unwrap(); actor.await.unwrap(); } #[tokio::test] - async fn changed_root_failure_only_publishes_new_root_observations() { - // Mutation caught: retaining old-root committed servers after a failed new-root load. + async fn refresh_retains_observations_and_changed_root_drops_old_status() { let (handle, receiver) = McpStatusHandle::channel(64); let (effects_tx, mut effects_rx) = mpsc::unbounded_channel(); let actor = tokio::spawn(run_actor(receiver, effects_tx)); - - let old_opening = handle.begin_open().await.unwrap(); - handle.bind_root("old-root".into()).await.unwrap(); - let McpStatusEffect::Load { - epoch, generation, .. - } = effects_rx.recv().await.unwrap() - else { - panic!("expected old-root load"); - }; - handle - .load_finished( - epoch, - generation, - Ok(map([server("old-only", McpServerState::Connected)])), - ) - .await - .unwrap(); - let _ = effects_rx.recv().await; - let _ = effects_rx.recv().await; - drop(old_opening); - - let new_opening = handle.begin_open().await.unwrap(); - handle - .notification( - Some("old-root".into()), - server("late-old", McpServerState::Connected), - ) - .await - .unwrap(); + for root in ["old-root", "old-root", "new-root"] { + let opening = handle.begin_open().await.unwrap(); + let first_refresh = handle.refresh().await.unwrap(); + let second_refresh = handle.refresh().await.unwrap(); + handle + .notification(Some(root.into()), server(root)) + .await + .unwrap(); + handle + .notification(Some("foreign".into()), server("foreign")) + .await + .unwrap(); + handle.bind_root(root.into()).await.unwrap(); + assert_eq!(published(&mut effects_rx).await, [root]); + assert_eq!(opening.await.unwrap(), Ok(())); + assert_eq!(first_refresh.await.unwrap(), Ok(())); + assert_eq!(second_refresh.await.unwrap(), Ok(())); + let refresh = handle.refresh().await.unwrap(); + assert_eq!(published(&mut effects_rx).await, [root]); + assert_eq!(refresh.await.unwrap(), Ok(())); + handle + .notification(Some(root.into()), server(root)) + .await + .unwrap(); + // Synchronize with the mailbox before testing duplicate suppression. + assert_eq!(handle.snapshot_for_test().await.unwrap(), [server(root)]); + assert!(effects_rx.try_recv().is_err()); + } + let reopened = handle.begin_open().await.unwrap(); handle.bind_root("new-root".into()).await.unwrap(); - let McpStatusEffect::Load { - epoch, generation, .. - } = effects_rx.recv().await.unwrap() - else { - panic!("expected new-root load"); - }; - handle - .notification( - Some("new-root".into()), - server("new-only", McpServerState::Starting), - ) - .await - .unwrap(); - handle - .load_finished(epoch, generation, Err("list unavailable".to_owned())) - .await - .unwrap(); - - let McpStatusEffect::Snapshot(new_root_failure_snapshot) = effects_rx.recv().await.unwrap() - else { - panic!("expected new-root snapshot"); - }; - assert_eq!(names(&new_root_failure_snapshot), vec!["new-only"]); - assert!(!names(&new_root_failure_snapshot).contains(&"old-only")); - assert!(!names(&new_root_failure_snapshot).contains(&"late-old")); - assert!(matches!( - effects_rx.recv().await, - Some(McpStatusEffect::Warning(_)) - )); - let _ = effects_rx.recv().await; - drop(new_opening); + assert!(published(&mut effects_rx).await.is_empty()); + assert_eq!(reopened.await.unwrap(), Ok(())); handle.shutdown().await.unwrap(); actor.await.unwrap(); } #[tokio::test] - async fn overlapping_refreshes_coalesce_and_late_old_epoch_success_is_ignored() { - // Mutation caught: a second refresh starts duplicate I/O or applies a stale root result. + async fn cancelled_opening_releases_joined_refresh_and_discards_staged_notifications() { let (handle, receiver) = McpStatusHandle::channel(64); let (effects_tx, mut effects_rx) = mpsc::unbounded_channel(); let actor = tokio::spawn(run_actor(receiver, effects_tx)); - let opening = handle.begin_open().await.unwrap(); let refresh = handle.refresh().await.unwrap(); - handle.bind_root("root-1".into()).await.unwrap(); - let McpStatusEffect::Load { - epoch: old_epoch, - generation: old_generation, - .. - } = effects_rx.recv().await.unwrap() - else { - panic!("expected first load"); - }; - assert!(effects_rx.try_recv().is_err()); - - let replacement_opening = handle.begin_open().await.unwrap(); - handle.bind_root("root-2".into()).await.unwrap(); - let McpStatusEffect::Load { - epoch, generation, .. - } = effects_rx.recv().await.unwrap() - else { - panic!("expected replacement load"); - }; - handle - .load_finished( - old_epoch, - old_generation, - Ok(map([server("old-only", McpServerState::Connected)])), - ) - .await - .unwrap(); - assert!(effects_rx.try_recv().is_err()); - handle - .load_finished( - epoch, - generation, - Ok(map([server("new-only", McpServerState::Connected)])), - ) + .notification(None, server("abandoned")) .await .unwrap(); - let McpStatusEffect::Snapshot(snapshot) = effects_rx.recv().await.unwrap() else { - panic!("expected replacement snapshot"); - }; - assert_eq!(names(&snapshot), vec!["new-only"]); - drop((opening, refresh, replacement_opening)); - handle.shutdown().await.unwrap(); - actor.await.unwrap(); - } - - #[tokio::test] - async fn root_change_during_refresh_replaces_the_old_load() { - // Mutation caught: accepting an old-root result after the root has changed. - let (handle, receiver) = McpStatusHandle::channel(64); - let (effects_tx, mut effects_rx) = mpsc::unbounded_channel(); - let actor = tokio::spawn(run_actor(receiver, effects_tx)); - - let opening = handle.begin_open().await.unwrap(); - handle.bind_root("old-root".into()).await.unwrap(); - let McpStatusEffect::Load { - epoch, generation, .. - } = effects_rx.recv().await.unwrap() - else { - panic!("expected initial load"); - }; - handle - .load_finished( - epoch, - generation, - Ok(map([server("old-only", McpServerState::Connected)])), - ) - .await - .unwrap(); - let _ = effects_rx.recv().await; - let McpStatusEffect::Complete(waiters) = effects_rx.recv().await.unwrap() else { - panic!("expected initial completion"); - }; - for waiter in waiters { - waiter.send(Ok(())).unwrap(); - } - assert_eq!(opening.await.unwrap(), Ok(())); - - let refresh = handle.refresh().await.unwrap(); - let McpStatusEffect::Load { - epoch: old_epoch, - generation: old_generation, - root, - } = effects_rx.recv().await.unwrap() - else { - panic!("expected old-root refresh load"); - }; - assert_eq!(root, "old-root"); - handle.bind_root("new-root".into()).await.unwrap(); - let McpStatusEffect::Load { - epoch, - generation, - root, - } = tokio::time::timeout(std::time::Duration::from_millis(100), effects_rx.recv()) - .await - .expect("expected replacement load") - .unwrap() - else { - panic!("expected replacement load"); - }; - assert_eq!(root, "new-root"); - - handle - .load_finished( - old_epoch, - old_generation, - Ok(map([server("old-only", McpServerState::Connected)])), - ) - .await - .unwrap(); - assert!(effects_rx.try_recv().is_err()); - - handle - .load_finished( - epoch, - generation, - Ok(map([server("new-only", McpServerState::Connected)])), - ) - .await - .unwrap(); - let McpStatusEffect::Snapshot(snapshot) = effects_rx.recv().await.unwrap() else { - panic!("expected new-root snapshot"); - }; - assert_eq!(names(&snapshot), vec!["new-only"]); - let McpStatusEffect::Complete(waiters) = effects_rx.recv().await.unwrap() else { - panic!("expected refresh completion"); - }; - for waiter in waiters { - waiter.send(Ok(())).unwrap(); - } - assert_eq!(refresh.await.unwrap(), Ok(())); + opening.cancel().await; + assert!(refresh.await.unwrap().is_err()); + let retry = handle.begin_open().await.unwrap(); + handle.bind_root("root".into()).await.unwrap(); + assert!(published(&mut effects_rx).await.is_empty()); + assert_eq!(retry.await.unwrap(), Ok(())); handle.shutdown().await.unwrap(); actor.await.unwrap(); } #[tokio::test] async fn closed_effect_receiver_fails_opening_waiters() { - // Mutation caught: retaining a completion waiter after the effect worker has stopped. let (handle, receiver) = McpStatusHandle::channel(64); let (effects_tx, effects_rx) = mpsc::unbounded_channel(); drop(effects_rx); let actor = tokio::spawn(run_actor(receiver, effects_tx)); - let opening = handle.begin_open().await.unwrap(); - handle.bind_root("root-1".into()).await.unwrap(); - - let completion = tokio::time::timeout(std::time::Duration::from_millis(100), opening) - .await - .expect("effect failure must resolve the opening waiter") - .unwrap(); - assert!(completion.is_err()); - handle.shutdown().await.unwrap(); - actor.await.unwrap(); - } - - #[tokio::test] - async fn concurrent_same_root_refreshes_emit_one_load_and_complete_both_waiters() { - // Mutation caught: a second post-bind refresh starts I/O or loses a completion waiter. - let (handle, receiver) = McpStatusHandle::channel(64); - let (effects_tx, mut effects_rx) = mpsc::unbounded_channel(); - let actor = tokio::spawn(run_actor(receiver, effects_tx)); - - let opening = handle.begin_open().await.unwrap(); - handle.bind_root("root-1".into()).await.unwrap(); - let McpStatusEffect::Load { - epoch, generation, .. - } = effects_rx.recv().await.unwrap() - else { - panic!("expected opening load"); - }; - handle - .load_finished(epoch, generation, Ok(BTreeMap::new())) - .await - .unwrap(); - let _ = effects_rx.recv().await; - let McpStatusEffect::Complete(waiters) = effects_rx.recv().await.unwrap() else { - panic!("expected opening completion"); - }; - for waiter in waiters { - waiter.send(Ok(())).unwrap(); - } - assert_eq!(opening.await.unwrap(), Ok(())); - - let first = handle.refresh().await.unwrap(); - let second = handle.refresh().await.unwrap(); - let McpStatusEffect::Load { - epoch, - generation, - root, - } = effects_rx.recv().await.unwrap() - else { - panic!("expected one refresh load"); - }; - assert_eq!(root, "root-1"); - assert!(effects_rx.try_recv().is_err()); - handle - .load_finished( - epoch, - generation, - Ok(map([server("same-root", McpServerState::Connected)])), - ) - .await - .unwrap(); - let _ = effects_rx.recv().await; - let McpStatusEffect::Complete(waiters) = effects_rx.recv().await.unwrap() else { - panic!("expected refresh completions"); - }; - assert_eq!(waiters.len(), 2); - for waiter in waiters { - waiter.send(Ok(())).unwrap(); - } - assert_eq!(first.await.unwrap(), Ok(())); - assert_eq!(second.await.unwrap(), Ok(())); + handle.bind_root("root".into()).await.unwrap(); + assert!( + tokio::time::timeout(std::time::Duration::from_secs(1), opening) + .await + .unwrap() + .unwrap() + .is_err() + ); handle.shutdown().await.unwrap(); actor.await.unwrap(); } diff --git a/apps/server/src/provider/codex/protocol.rs b/apps/server/src/provider/codex/protocol.rs index 521d590a2d..7afc8b96ad 100644 --- a/apps/server/src/provider/codex/protocol.rs +++ b/apps/server/src/provider/codex/protocol.rs @@ -735,6 +735,23 @@ async fn fail_connection( } *inner.close_reason.lock().await = Some(reason.clone()); let mut pending = inner.pending.lock().await; + if let ProtocolError::LineTooLong { stream, maximum } = &error { + let mut pending_methods: Vec<_> = pending + .values() + .map(|entry| entry.method.as_str()) + .collect(); + pending_methods.sort_unstable(); + pending_methods.dedup(); + // The bounded reader fails on the first non-LF byte beyond maximum. + tracing::warn!( + stream, + maximum, + bytes_seen = maximum.saturating_add(1), + lf_observed = false, + ?pending_methods, + "Codex JSON-RPC line too long" + ); + } for (_, entry) in pending.drain() { let _ = entry.responder.send(Err(ProtocolError::Closed { reason: reason.clone(), @@ -1086,6 +1103,93 @@ mod tests { )); } + #[tokio::test] + async fn oversized_stdout_closes_pending_requests_with_private_diagnostics() { + let Some(isolated) = crate::test_support::in_own_process( + "provider::codex::protocol::tests::oversized_stdout_closes_pending_requests_with_private_diagnostics", + ) else { + return; + }; + #[derive(Clone, Default)] + struct Captured(Arc>>); + impl io::Write for Captured { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + let log = Captured::default(); + let _logging = tracing::subscriber::set_default( + tracing_subscriber::fmt() + .with_ansi(false) + .without_time() + .with_writer({ + let log = log.clone(); + move || log.clone() + }) + .finish(), + ); + for trailing_lf in [false, true] { + let (stdout, mut peer_stdout) = duplex(16 * 1024); + let (stdin, peer_stdin) = duplex(1024); + let (connection, mut incoming) = JsonRpcConnection::spawn( + stdout, + stdin, + tokio::io::empty(), + ConnectionConfig::default(), + ); + let peer = async move { + let mut request = String::new(); + BufReader::new(peer_stdin) + .read_line(&mut request) + .await + .unwrap(); + let mut payload = vec![b'x'; CODEX_TRANSPORT_MESSAGE_MAX_BYTES + 1]; + payload[..16].copy_from_slice(b"payload-sentinel"); + if trailing_lf { + payload.push(b'\n'); + } + peer_stdout.write_all(&payload).await.unwrap(); + }; + let (result, ()) = tokio::join!( + connection.request("thread/resume", json!({"threadId":"request-sentinel"})), + peer, + ); + let error = result.unwrap_err(); + assert_eq!(connection.pending_request_count().await, 0); + assert!(matches!(error, ProtocolError::Closed { .. })); + assert_eq!( + error.to_string(), + "Codex JSON-RPC connection is closed: Codex JSON-RPC read line exceeded 8388608 bytes on stdout" + ); + assert!(matches!( + incoming.recv().await, + Some(IncomingEvent::Closed { .. }) + )); + } + let log = String::from_utf8(log.0.lock().unwrap().clone()).unwrap(); + assert_eq!( + log.matches("Codex JSON-RPC line too long").count(), + 2, + "{log}" + ); + for field in [ + "stream=\"stdout\"", + "maximum=8388608", + "bytes_seen=8388609", + "lf_observed=false", + "pending_methods=[\"thread/resume\"]", + ] { + assert!(log.contains(field), "missing {field}: {log}"); + } + assert!(!log.contains("payload-sentinel")); + assert!(!log.contains("request-sentinel")); + isolated.complete(); + } + #[tokio::test] async fn dropped_writer_and_response_waiter_report_closed_contracts() { let (writer, writer_rx) = mpsc::unbounded_channel(); diff --git a/apps/server/src/provider/codex/runtime.rs b/apps/server/src/provider/codex/runtime.rs index 5a139d9d2a..6d355d1c29 100644 --- a/apps/server/src/provider/codex/runtime.rs +++ b/apps/server/src/provider/codex/runtime.rs @@ -10,7 +10,7 @@ use thiserror::Error; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::{ sync::{Mutex, mpsc, oneshot}, - task::{JoinHandle, JoinSet}, + task::JoinHandle, }; use tokio_util::sync::CancellationToken; @@ -28,7 +28,7 @@ use super::{ }, mcp_status::{ McpOpenCompletion, McpOpenReservation, McpStatusEffect, McpStatusHandle, - mcp_server_status_from_notification, refresh_mcp_status_snapshot, run_actor, + mcp_server_status_from_notification, run_actor, }, model::{ BuildTurnStartInput, CODEX_HISTORY_TURNS_PAGE_LIMIT, CODEX_REMOTE_JSON_TOO_LARGE, @@ -697,11 +697,9 @@ pub async fn probe_provider( async fn run_mcp_status_effects( runtime: Weak, - handle: McpStatusHandle, mut effects_rx: mpsc::UnboundedReceiver, cancellation: CancellationToken, ) { - let mut loads = JoinSet::new(); loop { let effect = tokio::select! { biased; @@ -710,25 +708,8 @@ async fn run_mcp_status_effects( Some(effect) => effect, None => break, }, - Some(_) = loads.join_next(), if !loads.is_empty() => continue, }; match effect { - McpStatusEffect::Load { - epoch, - generation, - root, - } => { - let Some(inner) = runtime.upgrade() else { - break; - }; - let connection = inner.connection.lock().await.clone(); - drop(inner); - let handle = handle.clone(); - loads.spawn(async move { - let result = refresh_mcp_status_snapshot(&connection, &root).await; - let _ = handle.load_finished(epoch, generation, result).await; - }); - } McpStatusEffect::Snapshot(servers) => { let Some(inner) = runtime.upgrade() else { break; @@ -749,14 +730,6 @@ async fn run_mcp_status_effects( ) .await; } - McpStatusEffect::Warning(detail) => { - let Some(inner) = runtime.upgrade() else { - break; - }; - CodexSessionRuntime { inner } - .emit("runtime.warning", None, None, json!({ "message": detail })) - .await; - } McpStatusEffect::Complete(waiters) => { #[cfg(test)] if let Some(inner) = runtime.upgrade() @@ -772,8 +745,6 @@ async fn run_mcp_status_effects( } } } - loads.abort_all(); - while loads.join_next().await.is_some() {} } fn mcp_status_error(message: String) -> RuntimeError { @@ -893,7 +864,6 @@ impl CodexSessionRuntime { let actor_task = tokio::spawn(run_actor(mcp_status_rx, mcp_status_effects_tx)); let effect_task = tokio::spawn(run_mcp_status_effects( Arc::downgrade(&runtime.inner), - mcp_status, mcp_status_effects_rx, reconciliation_cancellation, )); @@ -4279,10 +4249,7 @@ mod tests { }, persistence::{Database, run_migrations}, provider::codex::{ - mcp_status::{ - MCP_STATUS_PAGE_LIMIT, MCP_STATUS_PAGE_SIZE, MCP_STATUS_REQUEST_TIMEOUT, - McpServerState, McpServerStatus, - }, + mcp_status::{McpServerState, McpServerStatus}, model::ReconciliationThread, protocol::ConnectionConfig, }, @@ -4382,19 +4349,6 @@ mod tests { writer.flush().await.expect("JSON-RPC flush"); } - fn assert_mcp_status_list_request(request: &Value, thread_id: &str, cursor: Option<&str>) { - let mut expected_params = json!({ - "threadId": thread_id, - "limit": MCP_STATUS_PAGE_SIZE, - "detail": "toolsAndAuthOnly", - }); - if let Some(cursor) = cursor { - expected_params["cursor"] = json!(cursor); - } - assert_eq!(request["method"], "mcpServerStatus/list"); - assert_eq!(request["params"], expected_params); - } - fn reconciliation_test_options() -> CodexSessionOptions { CodexSessionOptions { version: "0.1.1".to_owned(), @@ -5383,147 +5337,137 @@ mod tests { assert_ne!(first.request_id, second.request_id); } - async fn assert_mcp_status_discovery_failure(responses: Vec, expected_warning: &str) { - let (connection, incoming, peer_stdout, peer_stdin, _peer_stderr) = - runtime_test_connection(); - let runtime = CodexSessionRuntime::new(reconciliation_test_options(), connection, incoming); - let peer = tokio::spawn(async move { - let mut reader = BufReader::new(peer_stdin); - let mut writer = peer_stdout; - let initialize = read_runtime_test_json(&mut reader).await; - write_runtime_test_json(&mut writer, json!({ "id": initialize["id"], "result": {} })) - .await; - assert_eq!( - read_runtime_test_json(&mut reader).await["method"], - "initialized" + #[tokio::test] + async fn mcp_catalogue_oversize_thread_resumes_and_completes_followup_turns() { + // The peer would kill this transport if asked for its synthetic tool catalogue. + for resume in [true, false] { + let (connection, incoming, peer_stdout, peer_stdin, _peer_stderr) = + runtime_test_connection(); + let mut options = reconciliation_test_options(); + if resume { + options.resume_cursor = Some("provider-root".into()); + } + let runtime = CodexSessionRuntime::new_with_agent_activity_enabled( + options, connection, incoming, false, ); - let start = read_runtime_test_json(&mut reader).await; - write_runtime_test_json( - &mut writer, - json!({ "id": start["id"], "result": { "thread": { "id": "provider-root" } } }), - ) - .await; - let mut expected_cursor = None; - for response in responses { - let request = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request( - &request, - "provider-root", - expected_cursor.as_deref(), + let (completed_tx, mut completed_rx) = mpsc::channel(1); + let peer = tokio::spawn(async move { + let mut reader = BufReader::new(peer_stdin); + let mut writer = peer_stdout; + let initialize = read_runtime_test_json(&mut reader).await; + assert_eq!(initialize["method"], "initialize"); + write_runtime_test_json(&mut writer, json!({"id":initialize["id"], "result":{}})) + .await; + assert_eq!( + read_runtime_test_json(&mut reader).await["method"], + "initialized" ); - expected_cursor = response - .get("nextCursor") - .and_then(Value::as_str) - .map(str::to_owned); - let response = if response.get("error").is_some() { - json!({ "id": request["id"], "error": response["error"] }) - } else { - json!({ "id": request["id"], "result": response }) - }; - write_runtime_test_json(&mut writer, response).await; - } - let shutdown = read_runtime_test_json(&mut reader).await; - assert_eq!(shutdown["method"], "shutdown"); - write_runtime_test_json(&mut writer, json!({ "id": shutdown["id"], "result": null })) + let open = read_runtime_test_json(&mut reader).await; + assert_eq!( + open["method"], + if resume { + "thread/resume" + } else { + "thread/start" + } + ); + if resume { + assert_eq!(open["params"]["threadId"], "provider-root"); + } + write_runtime_test_json( + &mut writer, + json!({"method":"mcpServer/startupStatus/updated", + "params":{"threadId":"provider-root", "name":"docs", "status":"ready"}}), + ) .await; - }); - - runtime - .start() - .await - .expect("status discovery is best effort"); - let events = timeout(Duration::from_secs(1), async { - let mut events = Vec::new(); - loop { - let event = runtime.next_event().await.expect("runtime event"); - let ready = event.event_type == "session.ready"; - events.push(event); - if ready { - return events; + write_runtime_test_json( + &mut writer, + json!({"id":open["id"], "result":{"thread":{"id":"provider-root"}}}), + ) + .await; + let mut turns = 0; + loop { + let request = read_runtime_test_json(&mut reader).await; + match request["method"].as_str().unwrap() { + "mcpServerStatus/list" => { + let response = json!({"id":request["id"], "result":{"data":[{ + "name":"docs", "authStatus":"unsupported", "tools":{ + "large":{"name":"large", "description":"x".repeat(9_000_000), "inputSchema":{}} + } + }], "nextCursor":null}}); + let mut bytes = serde_json::to_vec(&response).unwrap(); + bytes.push(b'\n'); + assert!(bytes.len() > crate::provider::codex::protocol::CODEX_TRANSPORT_MESSAGE_MAX_BYTES); + let _ = writer.write_all(&bytes).await; + return turns; + } + "turn/start" => { + assert_eq!(request["params"]["threadId"], "provider-root"); + turns += 1; + let id = format!("turn-{turns}"); + write_runtime_test_json( + &mut writer, + json!({"id":request["id"], "result":{"turn":{"id":id}}}), + ) + .await; + completed_rx.recv().await.unwrap(); + write_runtime_test_json(&mut writer, json!({"method":"turn/completed", "params":{ + "threadId":"provider-root", "turn":{"id":id, "status":"completed"} + }})).await; + } + "shutdown" => { + write_runtime_test_json( + &mut writer, + json!({"id":request["id"], "result":null}), + ) + .await; + return turns; + } + method => panic!("unexpected method: {method}"), + } } - } - }) - .await - .expect("session becomes ready"); - assert_eq!( - events - .iter() - .map(|event| event.event_type.as_str()) - .collect::>(), - vec![ - "session.connecting", - "mcp.status.updated", - "runtime.warning", - "session.ready" - ] - ); - assert_eq!(events[0].payload, json!({})); - assert_eq!(events[1].payload, json!({ "servers": [] })); - assert!( - events[2].payload["message"] - .as_str() - .is_some_and(|message| message.contains(expected_warning)) - ); - assert_eq!(events[3].payload, json!({})); - - runtime.shutdown().await.expect("runtime shuts down"); - peer.await.expect("peer"); - } - - #[tokio::test] - async fn mcp_status_list_remote_errors_do_not_block_session_start() { - assert_mcp_status_discovery_failure( - vec![json!({ - "error": { "code": -32000, "message": "MCP unavailable" } - })], - "MCP unavailable", - ) - .await; - } - - #[tokio::test] - async fn mcp_status_list_malformed_pages_do_not_block_session_start() { - assert_mcp_status_discovery_failure( - vec![json!({ "data": "invalid", "nextCursor": null })], - "missing data array", - ) - .await; - } - - #[tokio::test] - async fn mcp_status_list_repeated_cursors_do_not_block_session_start() { - assert_mcp_status_discovery_failure( - vec![ - json!({ "data": [], "nextCursor": "repeat" }), - json!({ "data": [], "nextCursor": "repeat" }), - ], - "repeated nextCursor", - ) - .await; - } - - #[tokio::test] - async fn mcp_status_list_blank_and_non_string_cursors_warn_then_ready() { - for next_cursor in [json!(" "), json!(7)] { - assert_mcp_status_discovery_failure( - vec![json!({ "data": [], "nextCursor": next_cursor })], - "response has invalid nextCursor", - ) - .await; + }); + timeout(Duration::from_secs(5), async { + let session = runtime + .start() + .await + .expect("start/resume survives the MCP catalogue"); + assert_eq!(session.resume_cursor.as_deref(), Some("provider-root")); + let opening_events = drain_runtime_events(&runtime).await; + assert!( + opening_events + .iter() + .any(|event| event.event_type == "session.ready") + ); + assert!(opening_events.iter().any(|event| event.event_type + == "mcp.status.updated" + && event.payload == json!({"servers":[{"name":"docs", "state":"connected"}]}))); + for index in 1..=2 { + runtime.refresh_mcp_status().await; + let result = runtime + .send_turn(Some("synthetic followup".into()), vec![], None, None) + .await + .expect("same-thread followup accepted"); + assert_eq!(result.resume_cursor.as_deref(), Some("provider-root")); + assert_eq!(result.turn_id, format!("turn-{index}")); + completed_tx.send(()).await.unwrap(); + loop { + let event = runtime.next_event().await.unwrap(); + if event.event_type == "turn.completed" { + assert_eq!(event.turn_id.as_deref(), Some(result.turn_id.as_str())); + assert_eq!(event.payload["state"], "completed"); + break; + } + } + } + runtime.shutdown().await.unwrap(); + assert_eq!(peer.await.unwrap(), 2); + }) + .await + .expect("same-thread recovery and continued delivery complete"); } } - #[tokio::test] - async fn mcp_status_list_ninth_distinct_cursor_is_bounded_before_session_ready() { - assert_mcp_status_discovery_failure( - (0..MCP_STATUS_PAGE_LIMIT) - .map(|index| json!({ "data": [], "nextCursor": format!("cursor-{index}") })) - .collect(), - "exceeded page limit", - ) - .await; - } - #[test] fn mcp_status_official_startup_notifications_map_exactly() { let cases = [ @@ -5632,26 +5576,6 @@ mod tests { json!({ "id": start["id"], "result": { "thread": { "id": "provider-root" } } }), ) .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "provider-root", None); - write_runtime_test_json( - &mut writer, - json!({ - "id": list["id"], - "result": { - "data": [{ - "name": "baseline", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - }], - "nextCursor": null - } - }), - ) - .await; let shutdown = read_runtime_test_json(&mut reader).await; assert_eq!(shutdown["method"], "shutdown"); @@ -5674,7 +5598,6 @@ mod tests { "servers": [ { "name": "app-missing", "state": "connected" }, { "name": "app-null", "state": "connected" }, - { "name": "baseline", "state": "starting" }, { "name": "matching", "state": "connected" } ] }), @@ -5711,27 +5634,14 @@ mod tests { let start = read_runtime_test_json(&mut reader).await; write_runtime_test_json( &mut writer, - json!({ "id": start["id"], "result": { "thread": { "id": "old-root" } } }), + json!({"method":"mcpServer/startupStatus/updated", "params":{ + "threadId":"old-root", "name":"old-only", "status":"ready" + }}), ) .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "old-root", None); write_runtime_test_json( &mut writer, - json!({ - "id": list["id"], - "result": { - "data": [{ - "name": "old-only", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - }], - "nextCursor": null - } - }), + json!({ "id": start["id"], "result": { "thread": { "id": "old-root" } } }), ) .await; let _ = hold_old_peer.await; @@ -5777,26 +5687,6 @@ mod tests { json!({ "id": resume["id"], "result": { "thread": { "id": "new-root" } } }), ) .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "new-root", None); - write_runtime_test_json( - &mut writer, - json!({ - "id": list["id"], - "result": { - "data": [{ - "name": "baseline", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - }], - "nextCursor": null - } - }), - ) - .await; let shutdown = read_runtime_test_json(&mut reader).await; assert_eq!(shutdown["method"], "shutdown"); @@ -5822,7 +5712,6 @@ mod tests { "servers": [ { "name": "app-missing", "state": "connected" }, { "name": "app-null", "state": "connected" }, - { "name": "baseline", "state": "starting" }, { "name": "new-root-update", "state": "connected" } ] }), @@ -5838,7 +5727,7 @@ mod tests { } #[tokio::test] - async fn mcp_status_notifications_on_both_open_boundaries_win_and_foreign_root_is_excluded() { + async fn mcp_status_open_and_two_public_refreshes_share_one_snapshot_without_requests() { let (connection, incoming, peer_stdout, peer_stdin, _peer_stderr) = runtime_test_connection(); let runtime = CodexSessionRuntime::new_with_agent_activity_enabled( @@ -5847,7 +5736,8 @@ mod tests { incoming, false, ); - let peer_runtime = runtime.clone(); + let (opening_seen_tx, opening_seen_rx) = oneshot::channel(); + let (bind_root_tx, bind_root_rx) = oneshot::channel(); let peer = tokio::spawn(async move { let mut reader = BufReader::new(peer_stdin); let mut writer = peer_stdout; @@ -5860,185 +5750,40 @@ mod tests { ); let start = read_runtime_test_json(&mut reader).await; assert_eq!(start["method"], "thread/start"); - - for params in [ - json!({ - "threadId": "provider-root", - "name": "before-response", - "status": "ready" - }), - json!({ - "threadId": "foreign-root", - "name": "foreign", - "status": "ready" - }), - ] { - peer_runtime - .handle_notification("mcpServer/startupStatus/updated".to_owned(), params, 0) - .await; - } - + opening_seen_tx.send(()).expect("report opening request"); + bind_root_rx.await.expect("release root binding"); write_runtime_test_json( &mut writer, - json!({ "id": start["id"], "result": { "thread": { "id": "provider-root" } } }), + json!({"method":"mcpServer/startupStatus/updated", "params":{ + "threadId":"provider-root", "name":"shared", "status":"starting" + }}), ) .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "provider-root", None); - peer_runtime - .handle_notification( - "mcpServer/startupStatus/updated".to_owned(), - json!({ - "threadId": "provider-root", - "name": "after-response", - "status": "ready" - }), - 0, - ) - .await; write_runtime_test_json( &mut writer, - json!({ - "id": list["id"], - "result": { - "data": [ - { - "name": "before-response", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - }, - { - "name": "after-response", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - } - ], - "nextCursor": null - } - }), + json!({ "id": start["id"], "result": { "thread": { "id": "provider-root" } } }), ) .await; - let shutdown = read_runtime_test_json(&mut reader).await; - assert_eq!(shutdown["method"], "shutdown"); - write_runtime_test_json(&mut writer, json!({ "id": shutdown["id"], "result": null })) - .await; + let request = read_runtime_test_json(&mut reader).await; + assert_eq!(request["method"], "shutdown"); + write_runtime_test_json(&mut writer, json!({"id":request["id"], "result":null})).await; }); - runtime.start().await.expect("runtime starts"); - let events = drain_runtime_events(&runtime).await; - assert_eq!( - events - .iter() - .map(|event| (event.event_type.as_str(), event.payload.clone())) - .collect::>(), - vec![ - ("session.connecting", json!({})), - ( - "mcp.status.updated", - json!({ - "servers": [ - { "name": "after-response", "state": "connected" }, - { "name": "before-response", "state": "connected" } - ] - }), - ), - ("session.ready", json!({})), - ] + let start_runtime = runtime.clone(); + let start = tokio::spawn(async move { start_runtime.start().await }); + opening_seen_rx.await.expect("opening request arrives"); + let mut first_refresh = Box::pin(runtime.refresh_mcp_status()); + let mut second_refresh = Box::pin(runtime.refresh_mcp_status()); + assert!( + std::future::poll_fn(|cx| Poll::Ready(first_refresh.as_mut().poll(cx))) + .await + .is_pending() ); - - runtime.shutdown().await.expect("runtime shuts down"); - peer.await.expect("peer"); - } - - #[tokio::test] - async fn mcp_status_open_and_two_public_refreshes_share_one_request_and_snapshot() { - let (connection, incoming, peer_stdout, peer_stdin, _peer_stderr) = - runtime_test_connection(); - let runtime = CodexSessionRuntime::new_with_agent_activity_enabled( - reconciliation_test_options(), - connection, - incoming, - false, - ); - let (opening_seen_tx, opening_seen_rx) = oneshot::channel(); - let (bind_root_tx, bind_root_rx) = oneshot::channel(); - let peer = tokio::spawn(async move { - let mut reader = BufReader::new(peer_stdin); - let mut writer = peer_stdout; - let initialize = read_runtime_test_json(&mut reader).await; - write_runtime_test_json(&mut writer, json!({ "id": initialize["id"], "result": {} })) - .await; - assert_eq!( - read_runtime_test_json(&mut reader).await["method"], - "initialized" - ); - let start = read_runtime_test_json(&mut reader).await; - assert_eq!(start["method"], "thread/start"); - opening_seen_tx.send(()).expect("report opening request"); - bind_root_rx.await.expect("release root binding"); - write_runtime_test_json( - &mut writer, - json!({ "id": start["id"], "result": { "thread": { "id": "provider-root" } } }), - ) - .await; - - let mut first_page_requests = 0; - loop { - let request = read_runtime_test_json(&mut reader).await; - if request["method"] == "mcpServerStatus/list" { - first_page_requests += 1; - assert_mcp_status_list_request(&request, "provider-root", None); - write_runtime_test_json( - &mut writer, - json!({ - "id": request["id"], - "result": { - "data": [{ - "name": "shared", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - }], - "nextCursor": null - } - }), - ) - .await; - continue; - } - assert_eq!(request["method"], "shutdown"); - write_runtime_test_json( - &mut writer, - json!({ "id": request["id"], "result": null }), - ) - .await; - return first_page_requests; - } - }); - - let start_runtime = runtime.clone(); - let start = tokio::spawn(async move { start_runtime.start().await }); - opening_seen_rx.await.expect("opening request arrives"); - let mut first_refresh = Box::pin(runtime.refresh_mcp_status()); - let mut second_refresh = Box::pin(runtime.refresh_mcp_status()); - assert!( - std::future::poll_fn(|cx| Poll::Ready(first_refresh.as_mut().poll(cx))) - .await - .is_pending() - ); - assert!( - std::future::poll_fn(|cx| Poll::Ready(second_refresh.as_mut().poll(cx))) - .await - .is_pending() + assert!( + std::future::poll_fn(|cx| Poll::Ready(second_refresh.as_mut().poll(cx))) + .await + .is_pending() ); let (completion_blocked_tx, completion_blocked_rx) = oneshot::channel(); let (release_completion_tx, release_completion_rx) = oneshot::channel(); @@ -6085,7 +5830,7 @@ mod tests { assert!(drain_runtime_events(&runtime).await.is_empty()); runtime.shutdown().await.expect("runtime shuts down"); - assert_eq!(peer.await.expect("peer"), 1); + peer.await.expect("peer"); } #[tokio::test] @@ -6115,13 +5860,6 @@ mod tests { json!({ "id": start["id"], "result": { "thread": { "id": "provider-root" } } }), ) .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "provider-root", None); - write_runtime_test_json( - &mut writer, - json!({ "id": list["id"], "result": { "data": [], "nextCursor": null } }), - ) - .await; receive_updates .await @@ -6181,6 +5919,11 @@ mod tests { tokio::task::yield_now().await; assert!(drain_runtime_events(&runtime).await.is_empty()); + runtime.refresh_mcp_status().await; + let refreshed = drain_runtime_events(&runtime).await; + assert_eq!(refreshed.len(), 1); + assert_eq!(refreshed[0].payload, lifecycle.payload); + runtime.shutdown().await.expect("runtime shuts down"); peer.await.expect("peer"); } @@ -6211,13 +5954,6 @@ mod tests { json!({ "id": start["id"], "result": { "thread": { "id": "provider-root" } } }), ) .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "provider-root", None); - write_runtime_test_json( - &mut writer, - json!({ "id": list["id"], "result": { "data": [], "nextCursor": null } }), - ) - .await; let shutdown = read_runtime_test_json(&mut reader).await; assert_eq!(shutdown["method"], "shutdown"); @@ -6296,171 +6032,6 @@ mod tests { peer.await.expect("peer"); } - #[tokio::test] - async fn mcp_status_actor_orders_reconnect_failure_before_ready_and_drops_old_root() { - let (connection, incoming, peer_stdout, peer_stdin, _peer_stderr) = - runtime_test_connection(); - let runtime = CodexSessionRuntime::new_with_agent_activity_enabled( - reconciliation_test_options(), - connection, - incoming, - false, - ); - let (release_old_peer, hold_old_peer) = oneshot::channel(); - let old_peer = tokio::spawn(async move { - let mut reader = BufReader::new(peer_stdin); - let mut writer = peer_stdout; - let initialize = read_runtime_test_json(&mut reader).await; - write_runtime_test_json(&mut writer, json!({ "id": initialize["id"], "result": {} })) - .await; - assert_eq!( - read_runtime_test_json(&mut reader).await["method"], - "initialized" - ); - let start = read_runtime_test_json(&mut reader).await; - write_runtime_test_json( - &mut writer, - json!({ "id": start["id"], "result": { "thread": { "id": "old-root" } } }), - ) - .await; - let list = read_runtime_test_json(&mut reader).await; - write_runtime_test_json( - &mut writer, - json!({ - "id": list["id"], - "result": { - "data": [{ - "name": "old-only", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - }], - "nextCursor": null - } - }), - ) - .await; - let _ = hold_old_peer.await; - }); - - runtime.start().await.expect("initial runtime starts"); - let initial_events = drain_runtime_events(&runtime).await; - assert_eq!( - initial_events - .iter() - .map(|event| (event.event_type.as_str(), event.payload.clone())) - .collect::>(), - vec![ - ("session.connecting", json!({})), - ( - "mcp.status.updated", - json!({ - "servers": [{ "name": "old-only", "state": "starting" }] - }), - ), - ("session.ready", json!({})), - ] - ); - - let (replacement, replacement_incoming, new_stdout, new_stdin, _new_stderr) = - runtime_test_connection(); - let peer_runtime = runtime.clone(); - let new_peer = tokio::spawn(async move { - let mut reader = BufReader::new(new_stdin); - let mut writer = new_stdout; - let initialize = read_runtime_test_json(&mut reader).await; - write_runtime_test_json(&mut writer, json!({ "id": initialize["id"], "result": {} })) - .await; - assert_eq!( - read_runtime_test_json(&mut reader).await["method"], - "initialized" - ); - let resume = read_runtime_test_json(&mut reader).await; - assert_eq!(resume["method"], "thread/resume"); - assert_eq!(resume["params"]["threadId"], "old-root"); - let incoming_gate = peer_runtime.inner.explicit_close.lock().await; - write_runtime_test_json( - &mut writer, - json!({ - "method": "mcpServer/startupStatus/updated", - "params": { - "threadId": "new-root", - "name": "new-only", - "status": "ready" - } - }), - ) - .await; - write_runtime_test_json( - &mut writer, - json!({ "id": resume["id"], "result": { "thread": { "id": "new-root" } } }), - ) - .await; - assert!( - timeout( - Duration::from_millis(50), - read_runtime_test_json(&mut reader) - ) - .await - .is_err(), - "the thread response must not overtake the earlier wire notification" - ); - drop(incoming_gate); - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "new-root", None); - write_runtime_test_json( - &mut writer, - json!({ - "id": list["id"], - "error": { "code": -32000, "message": "MCP unavailable on reconnect" } - }), - ) - .await; - - let shutdown = read_runtime_test_json(&mut reader).await; - assert_eq!(shutdown["method"], "shutdown"); - write_runtime_test_json(&mut writer, json!({ "id": shutdown["id"], "result": null })) - .await; - }); - - runtime - .reconnect(replacement, replacement_incoming) - .await - .expect("runtime reconnects despite MCP discovery failure"); - let reconnect_events = drain_runtime_events(&runtime).await; - let event_types = reconnect_events - .iter() - .skip(1) - .map(|event| event.event_type.as_str()) - .collect::>(); - assert_eq!(reconnect_events[0].event_type, "session.connecting"); - assert_eq!( - event_types, - vec!["mcp.status.updated", "runtime.warning", "session.ready"] - ); - assert_eq!(reconnect_events[0].payload, json!({})); - let snapshot_names = reconnect_events[1].payload["servers"] - .as_array() - .expect("complete MCP snapshot") - .iter() - .map(|server| server["name"].as_str().expect("server name")) - .collect::>(); - assert_eq!(snapshot_names, vec!["new-only"]); - assert!( - reconnect_events[2].payload["message"] - .as_str() - .is_some_and(|message| message.contains("MCP unavailable on reconnect")) - ); - assert_eq!(reconnect_events[3].payload, json!({})); - - runtime.shutdown().await.expect("runtime shuts down"); - new_peer.await.expect("replacement peer"); - release_old_peer.send(()).expect("release old peer"); - old_peer.await.expect("old peer"); - } - #[tokio::test] async fn mcp_status_failed_open_does_not_strand_later_refresh() { let (connection, incoming, peer_stdout, peer_stdin, _peer_stderr) = @@ -6551,13 +6122,6 @@ mod tests { json!({ "id": retry_start["id"], "result": { "thread": { "id": "provider-root" } } }), ) .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "provider-root", None); - write_runtime_test_json( - &mut writer, - json!({ "id": list["id"], "result": { "data": [], "nextCursor": null } }), - ) - .await; let shutdown = read_runtime_test_json(&mut reader).await; assert_eq!(shutdown["method"], "shutdown"); @@ -6611,27 +6175,14 @@ mod tests { let start = read_runtime_test_json(&mut reader).await; write_runtime_test_json( &mut writer, - json!({ "id": start["id"], "result": { "thread": { "id": "old-root" } } }), + json!({"method":"mcpServer/startupStatus/updated", "params":{ + "threadId":"old-root", "name":"old-only", "status":"ready" + }}), ) .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "old-root", None); write_runtime_test_json( &mut writer, - json!({ - "id": list["id"], - "result": { - "data": [{ - "name": "old-only", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - }], - "nextCursor": null - } - }), + json!({ "id": start["id"], "result": { "thread": { "id": "old-root" } } }), ) .await; let _ = hold_old_peer.await; @@ -6657,19 +6208,7 @@ mod tests { assert_eq!(abandoned_resume["params"]["threadId"], "old-root"); resume_seen_tx.send(()).expect("report abandoned resume"); - let mut retry_initialize = read_runtime_test_json(&mut reader).await; - if retry_initialize["method"] == "mcpServerStatus/list" { - assert_mcp_status_list_request(&retry_initialize, "old-root", None); - write_runtime_test_json( - &mut writer, - json!({ - "id": retry_initialize["id"], - "error": { "code": -32000, "message": "refresh unavailable" } - }), - ) - .await; - retry_initialize = read_runtime_test_json(&mut reader).await; - } + let retry_initialize = read_runtime_test_json(&mut reader).await; assert_eq!(retry_initialize["method"], "initialize"); write_runtime_test_json( &mut writer, @@ -6688,13 +6227,6 @@ mod tests { json!({ "id": retry_resume["id"], "result": { "thread": { "id": "old-root" } } }), ) .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "old-root", None); - write_runtime_test_json( - &mut writer, - json!({ "id": list["id"], "result": { "data": [], "nextCursor": null } }), - ) - .await; let shutdown = read_runtime_test_json(&mut reader).await; assert_eq!(shutdown["method"], "shutdown"); @@ -6767,13 +6299,6 @@ mod tests { json!({ "id": start["id"], "result": { "thread": { "id": "old-root" } } }), ) .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "old-root", None); - write_runtime_test_json( - &mut writer, - json!({ "id": list["id"], "result": { "data": [], "nextCursor": null } }), - ) - .await; let _ = hold_old_peer.await; }); runtime.start().await.expect("initial runtime starts"); @@ -6802,13 +6327,6 @@ mod tests { ) .await; - let refresh = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&refresh, "old-root", None); - write_runtime_test_json( - &mut writer, - json!({ "id": refresh["id"], "result": { "data": [], "nextCursor": null } }), - ) - .await; let shutdown = read_runtime_test_json(&mut reader).await; write_runtime_test_json(&mut writer, json!({ "id": shutdown["id"], "result": null })) .await; @@ -6830,563 +6348,34 @@ mod tests { old_peer.await.expect("old peer"); } - #[tokio::test] - async fn mcp_status_reconnect_pre_root_update_wins_successful_list_before_ready() { - let (connection, incoming, peer_stdout, peer_stdin, _peer_stderr) = + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn cancelled_old_epoch_cannot_mutate_or_emit_after_response_wins() { + let (connection_a, incoming_a, peer_stdout_a, peer_stdin_a, _peer_stderr_a) = runtime_test_connection(); - let runtime = CodexSessionRuntime::new_with_agent_activity_enabled( - reconciliation_test_options(), - connection, - incoming, - false, - ); - let (release_old_peer, hold_old_peer) = oneshot::channel(); + let runtime = + CodexSessionRuntime::new(reconciliation_test_options(), connection_a, incoming_a); + let (send_root_tx, send_root_rx) = oneshot::channel(); + let (background_seen_tx, background_seen_rx) = oneshot::channel(); + let (release_background_tx, release_background_rx) = oneshot::channel(); + let (background_replied_tx, background_replied_rx) = oneshot::channel(); let old_peer = tokio::spawn(async move { - let mut reader = BufReader::new(peer_stdin); - let mut writer = peer_stdout; + let mut reader = BufReader::new(peer_stdin_a); + let mut writer = peer_stdout_a; let initialize = read_runtime_test_json(&mut reader).await; - write_runtime_test_json(&mut writer, json!({ "id": initialize["id"], "result": {} })) - .await; + write_runtime_test_json( + &mut writer, + json!({"id": initialize["id"].clone(), "result": {}}), + ) + .await; assert_eq!( read_runtime_test_json(&mut reader).await["method"], "initialized" ); let start = read_runtime_test_json(&mut reader).await; - write_runtime_test_json( - &mut writer, - json!({ "id": start["id"], "result": { "thread": { "id": "old-root" } } }), - ) - .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "old-root", None); write_runtime_test_json( &mut writer, json!({ - "id": list["id"], - "result": { - "data": [{ - "name": "context7", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - }], - "nextCursor": null - } - }), - ) - .await; - let _ = hold_old_peer.await; - }); - - runtime.start().await.expect("initial runtime starts"); - let initial_events = drain_runtime_events(&runtime).await; - assert_eq!( - initial_events - .iter() - .map(|event| event.event_type.as_str()) - .collect::>(), - vec!["session.connecting", "mcp.status.updated", "session.ready"] - ); - assert_eq!( - initial_events[1].payload, - json!({ "servers": [{ "name": "context7", "state": "starting" }] }) - ); - - let (replacement, replacement_incoming, new_stdout, new_stdin, _new_stderr) = - runtime_test_connection(); - let peer_runtime = runtime.clone(); - let (send_later_updates, receive_later_updates) = oneshot::channel(); - let new_peer = tokio::spawn(async move { - let mut reader = BufReader::new(new_stdin); - let mut writer = new_stdout; - let initialize = read_runtime_test_json(&mut reader).await; - write_runtime_test_json(&mut writer, json!({ "id": initialize["id"], "result": {} })) - .await; - assert_eq!( - read_runtime_test_json(&mut reader).await["method"], - "initialized" - ); - let resume = read_runtime_test_json(&mut reader).await; - assert_eq!(resume["method"], "thread/resume"); - assert_eq!(resume["params"]["threadId"], "old-root"); - peer_runtime - .handle_notification( - "mcpServer/startupStatus/updated".to_owned(), - json!({ - "threadId": "new-root", - "name": "context7", - "status": "ready" - }), - 0, - ) - .await; - write_runtime_test_json( - &mut writer, - json!({ "id": resume["id"], "result": { "thread": { "id": "new-root" } } }), - ) - .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "new-root", None); - write_runtime_test_json( - &mut writer, - json!({ - "id": list["id"], - "result": { - "data": [{ - "name": "context7", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - }], - "nextCursor": null - } - }), - ) - .await; - - receive_later_updates - .await - .expect("release later lifecycle updates"); - for (name, status) in [("context7", "ready"), ("barrier", "ready")] { - write_runtime_test_json( - &mut writer, - json!({ - "method": "mcpServer/startupStatus/updated", - "params": { - "threadId": "new-root", - "name": name, - "status": status - } - }), - ) - .await; - } - - let shutdown = read_runtime_test_json(&mut reader).await; - assert_eq!(shutdown["method"], "shutdown"); - write_runtime_test_json(&mut writer, json!({ "id": shutdown["id"], "result": null })) - .await; - }); - - runtime - .reconnect(replacement, replacement_incoming) - .await - .expect("runtime reconnects after successful MCP discovery"); - let reconnect_events = drain_runtime_events(&runtime).await; - assert_eq!( - reconnect_events - .iter() - .map(|event| event.event_type.as_str()) - .collect::>(), - vec!["session.connecting", "mcp.status.updated", "session.ready"] - ); - assert_eq!( - reconnect_events[1].payload, - json!({ "servers": [{ "name": "context7", "state": "connected" }] }) - ); - - send_later_updates - .send(()) - .expect("send later lifecycle updates"); - let barrier = timeout(Duration::from_secs(1), runtime.next_event()) - .await - .expect("barrier lifecycle event arrives") - .expect("runtime event channel remains open"); - assert_eq!(barrier.event_type, "mcp.status.updated"); - assert_eq!( - barrier.payload, - json!({ - "servers": [ - { "name": "barrier", "state": "connected" }, - { "name": "context7", "state": "connected" } - ] - }) - ); - tokio::task::yield_now().await; - assert!(drain_runtime_events(&runtime).await.is_empty()); - - runtime.shutdown().await.expect("runtime shuts down"); - new_peer.await.expect("replacement peer"); - release_old_peer.send(()).expect("release old peer"); - old_peer.await.expect("old peer"); - } - - #[tokio::test(start_paused = true)] - async fn mcp_status_silent_list_times_out_without_leaking_or_blocking_ready() { - let (connection, incoming, peer_stdout, peer_stdin, _peer_stderr) = - runtime_test_connection(); - let pending_connection = connection.clone(); - let runtime = CodexSessionRuntime::new_with_agent_activity_enabled( - reconciliation_test_options(), - connection, - incoming, - false, - ); - let (list_seen, wait_for_list) = oneshot::channel(); - let peer = tokio::spawn(async move { - let mut reader = BufReader::new(peer_stdin); - let mut writer = peer_stdout; - let initialize = read_runtime_test_json(&mut reader).await; - write_runtime_test_json(&mut writer, json!({ "id": initialize["id"], "result": {} })) - .await; - assert_eq!( - read_runtime_test_json(&mut reader).await["method"], - "initialized" - ); - let start = read_runtime_test_json(&mut reader).await; - write_runtime_test_json( - &mut writer, - json!({ "id": start["id"], "result": { "thread": { "id": "provider-root" } } }), - ) - .await; - let list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&list, "provider-root", None); - list_seen.send(()).expect("record silent list request"); - - let shutdown = read_runtime_test_json(&mut reader).await; - assert_eq!(shutdown["method"], "shutdown"); - write_runtime_test_json(&mut writer, json!({ "id": shutdown["id"], "result": null })) - .await; - }); - - let start_runtime = runtime.clone(); - let start = tokio::spawn(async move { start_runtime.start().await }); - wait_for_list.await.expect("silent list request arrives"); - assert_eq!(pending_connection.pending_request_count().await, 1); - - tokio::time::advance(MCP_STATUS_REQUEST_TIMEOUT + Duration::from_millis(1)).await; - tokio::task::yield_now().await; - start - .await - .expect("start task") - .expect("MCP timeout is non-fatal"); - - assert_eq!(pending_connection.pending_request_count().await, 0); - let events = drain_runtime_events(&runtime).await; - assert_eq!( - events - .iter() - .map(|event| event.event_type.as_str()) - .collect::>(), - vec![ - "session.connecting", - "mcp.status.updated", - "runtime.warning", - "session.ready" - ] - ); - assert_eq!(events[1].payload, json!({ "servers": [] })); - assert!( - events[2].payload["message"] - .as_str() - .is_some_and(|message| message.contains("request timed out")) - ); - - runtime.shutdown().await.expect("runtime shuts down"); - peer.await.expect("peer"); - } - - #[tokio::test] - async fn mcp_status_authoritative_empty_and_terminal_eighth_page_succeed() { - let (connection, incoming, peer_stdout, peer_stdin, _peer_stderr) = - runtime_test_connection(); - let runtime = CodexSessionRuntime::new_with_agent_activity_enabled( - reconciliation_test_options(), - connection, - incoming, - false, - ); - let peer = tokio::spawn(async move { - let mut reader = BufReader::new(peer_stdin); - let mut writer = peer_stdout; - let initialize = read_runtime_test_json(&mut reader).await; - write_runtime_test_json(&mut writer, json!({ "id": initialize["id"], "result": {} })) - .await; - assert_eq!( - read_runtime_test_json(&mut reader).await["method"], - "initialized" - ); - let start = read_runtime_test_json(&mut reader).await; - write_runtime_test_json( - &mut writer, - json!({ "id": start["id"], "result": { "thread": { "id": "provider-root" } } }), - ) - .await; - let initial_list = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&initial_list, "provider-root", None); - write_runtime_test_json( - &mut writer, - json!({ "id": initial_list["id"], "result": { "data": [], "nextCursor": null } }), - ) - .await; - - for page in 0..MCP_STATUS_PAGE_LIMIT { - let request = read_runtime_test_json(&mut reader).await; - let expected_cursor = (page > 0).then(|| format!("cursor-{page}")); - assert_mcp_status_list_request( - &request, - "provider-root", - expected_cursor.as_deref(), - ); - let terminal = page + 1 == MCP_STATUS_PAGE_LIMIT; - let data = if terminal { - vec![json!({ - "name": "terminal", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - })] - } else { - Vec::new() - }; - write_runtime_test_json( - &mut writer, - json!({ - "id": request["id"], - "result": { - "data": data, - "nextCursor": if terminal { - Value::Null - } else { - json!(format!("cursor-{}", page + 1)) - } - } - }), - ) - .await; - } - - let shutdown = read_runtime_test_json(&mut reader).await; - assert_eq!(shutdown["method"], "shutdown"); - write_runtime_test_json(&mut writer, json!({ "id": shutdown["id"], "result": null })) - .await; - }); - - runtime.start().await.expect("runtime starts"); - let initial_events = drain_runtime_events(&runtime).await; - assert_eq!( - initial_events - .iter() - .map(|event| (event.event_type.as_str(), event.payload.clone())) - .collect::>(), - vec![ - ("session.connecting", json!({})), - ("mcp.status.updated", json!({ "servers": [] })), - ("session.ready", json!({})), - ] - ); - - runtime.refresh_mcp_status().await; - let terminal_events = drain_runtime_events(&runtime).await; - assert_eq!( - terminal_events - .iter() - .map(|event| (event.event_type.as_str(), event.payload.clone())) - .collect::>(), - vec![( - "mcp.status.updated", - json!({ - "servers": [{ "name": "terminal", "state": "starting" }] - }), - )] - ); - - runtime.shutdown().await.expect("runtime shuts down"); - peer.await.expect("peer"); - } - - #[tokio::test] - async fn mcp_status_official_pages_emit_ordered_replacing_snapshots_without_warnings() { - let (connection, incoming, peer_stdout, peer_stdin, _peer_stderr) = - runtime_test_connection(); - let runtime = CodexSessionRuntime::new_with_agent_activity_enabled( - reconciliation_test_options(), - connection, - incoming, - false, - ); - let peer = tokio::spawn(async move { - let mut reader = BufReader::new(peer_stdin); - let mut writer = peer_stdout; - let initialize = read_runtime_test_json(&mut reader).await; - write_runtime_test_json(&mut writer, json!({ "id": initialize["id"], "result": {} })) - .await; - assert_eq!( - read_runtime_test_json(&mut reader).await["method"], - "initialized" - ); - let start = read_runtime_test_json(&mut reader).await; - assert_eq!(start["method"], "thread/start"); - write_runtime_test_json( - &mut writer, - json!({ - "id": start["id"], - "result": { - "thread": { "id": "provider-root" }, - "cwd": "/tmp/project", - "model": "gpt-5.3-codex" - } - }), - ) - .await; - - let first_page = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&first_page, "provider-root", None); - write_runtime_test_json( - &mut writer, - json!({ - "id": first_page["id"], - "result": { - "data": [{ - "name": "context7", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "unsupported" - }], - "nextCursor": "next-page" - } - }), - ) - .await; - let second_page = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&second_page, "provider-root", Some("next-page")); - write_runtime_test_json( - &mut writer, - json!({ - "id": second_page["id"], - "result": { - "data": [ - { - "name": "atlassian", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "notLoggedIn" - }, - { - "name": "oauth", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "oAuth" - } - ], - "nextCursor": null - } - }), - ) - .await; - - let replacement = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&replacement, "provider-root", None); - write_runtime_test_json( - &mut writer, - json!({ - "id": replacement["id"], - "result": { - "data": [{ - "name": "oauth", - "serverInfo": null, - "tools": {}, - "resources": [], - "resourceTemplates": [], - "authStatus": "oAuth" - }], - "nextCursor": null - } - }), - ) - .await; - let shutdown = read_runtime_test_json(&mut reader).await; - assert_eq!(shutdown["method"], "shutdown"); - write_runtime_test_json(&mut writer, json!({ "id": shutdown["id"], "result": null })) - .await; - }); - - runtime.start().await.expect("runtime starts"); - let initial_events = drain_runtime_events(&runtime).await; - assert_eq!( - initial_events - .iter() - .map(|event| (event.event_type.as_str(), event.payload.clone())) - .collect::>(), - vec![ - ("session.connecting", json!({})), - ( - "mcp.status.updated", - json!({ - "servers": [ - { - "name": "atlassian", - "state": "needs-auth", - "detail": "Authentication required." - }, - { "name": "context7", "state": "starting" }, - { "name": "oauth", "state": "starting" } - ] - }), - ), - ("session.ready", json!({})), - ] - ); - - runtime.refresh_mcp_status().await; - let replacement_events = drain_runtime_events(&runtime).await; - assert_eq!( - replacement_events - .iter() - .map(|event| (event.event_type.as_str(), event.payload.clone())) - .collect::>(), - vec![( - "mcp.status.updated", - json!({ - "servers": [{ "name": "oauth", "state": "starting" }] - }), - )] - ); - - runtime.shutdown().await.expect("runtime shuts down"); - peer.await.expect("peer"); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 4)] - async fn cancelled_old_epoch_cannot_mutate_or_emit_after_response_wins() { - let (connection_a, incoming_a, peer_stdout_a, peer_stdin_a, _peer_stderr_a) = - runtime_test_connection(); - let runtime = - CodexSessionRuntime::new(reconciliation_test_options(), connection_a, incoming_a); - let (send_root_tx, send_root_rx) = oneshot::channel(); - let (background_seen_tx, background_seen_rx) = oneshot::channel(); - let (release_background_tx, release_background_rx) = oneshot::channel(); - let (background_replied_tx, background_replied_rx) = oneshot::channel(); - let old_peer = tokio::spawn(async move { - let mut reader = BufReader::new(peer_stdin_a); - let mut writer = peer_stdout_a; - let initialize = read_runtime_test_json(&mut reader).await; - write_runtime_test_json( - &mut writer, - json!({"id": initialize["id"].clone(), "result": {}}), - ) - .await; - assert_eq!( - read_runtime_test_json(&mut reader).await["method"], - "initialized" - ); - let start = read_runtime_test_json(&mut reader).await; - write_runtime_test_json( - &mut writer, - json!({ - "id": start["id"].clone(), + "id": start["id"].clone(), "result": { "thread": {"id": "old-root"}, "cwd": "/tmp/project", @@ -7395,16 +6384,6 @@ mod tests { }), ) .await; - let mcp_status = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&mcp_status, "old-root", None); - write_runtime_test_json( - &mut writer, - json!({ - "id": mcp_status["id"].clone(), - "result": { "data": [], "nextCursor": null } - }), - ) - .await; send_root_rx.await.expect("release old root notification"); write_runtime_test_json( &mut writer, @@ -7575,16 +6554,6 @@ mod tests { }), ) .await; - let mcp_status = read_runtime_test_json(&mut reader).await; - assert_mcp_status_list_request(&mcp_status, "new-root", None); - write_runtime_test_json( - &mut writer, - json!({ - "id": mcp_status["id"].clone(), - "result": { "data": [], "nextCursor": null } - }), - ) - .await; let root_read = read_runtime_test_json(&mut reader).await; assert_eq!(root_read["method"], "thread/read"); assert_eq!(root_read["params"]["threadId"], "new-root"); diff --git a/apps/server/tests/provider_codex.rs b/apps/server/tests/provider_codex.rs index 2c452480e6..a31ff13e12 100644 --- a/apps/server/tests/provider_codex.rs +++ b/apps/server/tests/provider_codex.rs @@ -9629,15 +9629,15 @@ async fn activity_runtime_reconnect_and_shutdown_drop_stale_incoming_work() { })); let peer_a_task = tokio::spawn(peer_a.run()); runtime.start().await.expect("initial runtime start"); - let startup_events = runtime.collect_events(4).await; + let startup_events = timeout(Duration::from_secs(2), runtime.collect_events(3)) + .await + .expect("startup publishes status and readiness without catalogue discovery"); assert_eq!( - startup_events.last().map(|event| event.event_type.as_str()), - Some("session.ready") - ); - assert!( startup_events .iter() - .any(|event| event.event_type == "mcp.status.updated") + .map(|event| event.event_type.as_str()) + .collect::>(), + ["session.connecting", "mcp.status.updated", "session.ready"] ); peer_a_task.await.expect("initial peer"); @@ -9726,7 +9726,16 @@ async fn activity_runtime_reconnect_and_shutdown_drop_stale_incoming_work() { .reconnect(connection_b, new_incoming_rx) .await .expect("runtime reconnects"); - runtime.collect_events(2).await; + let reconnect_events = timeout(Duration::from_secs(2), runtime.collect_events(3)) + .await + .expect("reconnect publishes status and readiness"); + assert_eq!( + reconnect_events + .iter() + .map(|event| event.event_type.as_str()) + .collect::>(), + ["session.connecting", "mcp.status.updated", "session.ready"] + ); tokio::task::yield_now().await; assert!( old_incoming_tx diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index 984f85841b..bf205f7be3 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -1752,7 +1752,9 @@ it. - `delivery.failing` (#273) is recorded once for a message delivery that fails for good without reaching its provider: the provider refused its own configuration (`provider-configuration`, also when restart recovery meets - it), or the twentieth not-sent attempt or failed restart-recovery launch + it), a launch closed by an oversized stdio line (`transport-line-too-long`, + #316, also during restart recovery), or the twentieth not-sent attempt or + failed restart-recovery launch (`attempts-exhausted`; the recovered message then stays uncertain). It keeps the provider, the attempt count and that class, never the error text, and cites no event; each such failure also logs a `tracing::warn!`. diff --git a/docs/architecture/providers.md b/docs/architecture/providers.md index 3b20e9dac4..8cada3751f 100644 --- a/docs/architecture/providers.md +++ b/docs/architecture/providers.md @@ -140,7 +140,16 @@ it again."): the delivery fails with the actionable detail instead of relaunching the provider in a retry loop. A retry keeps the turn's original route, so it launches once the executable exists at that path; a changed Binary path takes a new message, sent after dismissing the failed one, which -holds back the thread's later messages. Every other +holds back the thread's later messages. A Codex, Cursor, or Grok launch closed +because a stdout line exceeded its transport bound also fails once +(`ProviderRuntimeError::TransportLineTooLong`, #316), preserving the nested +protocol detail. It records `delivery.failing` with class +`transport-line-too-long`. Restart recovery also stops after one such launch, +leaving the original delivery uncertain for Retry or Dismiss. The transport +bounds stay unchanged. Codex logs only +the stream, maximum, bytes seen, whether LF was observed, and registered pending +request methods; these methods do not identify which response or notification +exceeded the bound. No payload or request parameters are logged. Every other launch failure, such as another spawn error, a port race, or a server that is slow to start, stays definitely not sent and is retried with backoff up to 30 seconds between attempts. @@ -724,6 +733,25 @@ waiters. The response is applied only while that turn remains current, so a late response cannot overwrite a newer turn; failures remain nonfatal and the last valid stream-derived snapshot stays visible. +Codex MCP status uses `mcpServer/startupStatus/updated` notifications only. +It never requests `mcpServerStatus/list`: even `toolsAndAuthOnly` embeds tool +schemas and can exceed the shared chat transport's 8 MiB stdout limit. This +avoids the lab-proven MCP catalogue startup failure on both new and resumed +threads without changing the model's access to MCP tools. + +The status actor stages at most 64 notifications during initialization and +thread opening, then binds the returned root and publishes matching-root and +app-scoped observations before `session.ready`. A successful reconnect starts +with that connection's observations, including when it resumes the same root; +a failed or cancelled opening retains the previous committed snapshot and +releases joined refresh waiters. Foreign roots and malformed root identifiers +are excluded. Status publication and caller completion remain ordered through +one effect worker. Refresh republishes observed state without provider I/O or +erasing it. Servers that emit no startup notification may be absent, and auth +state is available only when reported by a notification; the list-only auth +baseline is no longer available. The catalogue loader, paging limits, timeouts, +and stale-load generations are removed because there is no discovery request. + Claude MCP status uses the CLI's native status surfaces rather than filesystem configuration guesses. The adapter normalizes the `system:init` `mcp_servers` snapshot and, after each successful turn, sends the correlated `mcp_status` diff --git a/docs/plans/2026-11-06-codex-stdout-line-too-long-design.md b/docs/plans/2026-11-06-codex-stdout-line-too-long-design.md new file mode 100644 index 0000000000..5ac4cc9b6e --- /dev/null +++ b/docs/plans/2026-11-06-codex-stdout-line-too-long-design.md @@ -0,0 +1,130 @@ +# Codex stdout LineTooLong closes delivery (#316) + +Status: **Self-approved by Mauro** (explicit “approve yourself” for this defect, 2026-11-06 ART). Design for issue [#316](https://github.com/mubeda/BibCode/issues/316). Verified against `origin/main` at `2101b2cde9` (post v0.15.2 chore merge); transport path unchanged from v0.15.1 for this mechanism. + +## Problem + +Codex **chat** stdio JSON-RPC (`apps/server/src/provider/codex/protocol.rs`) hard-fails the whole connection when any single stdout span exceeds **`CODEX_TRANSPORT_MESSAGE_MAX_BYTES` (8 MiB)** before a newline (`read_bounded_line` → `ProtocolError::LineTooLong` → `fail_connection` nests Display into `Closed`). Turn delivery classifies the resulting launch failure as **`DefinitelyNotSent`** and retries up to **`DELIVERY_ATTEMPTS_MAX` (20)**, then Rejects with `delivery.failing` / `attempts-exhausted`. The UI shows Delivery failed with the nested string; later messages wait behind the failed head. + +Observed on BiBCode 0.15.1 + Codex CLI 0.167.0 (remote server). Outgoing user text was ~90 characters — **not** the oversized payload. Emitting JSON-RPC method is **unproven** (no method/byte diagnostics today). Synthetic tests show over-bound blobs with or without a trailing LF produce the **identical** nested detail. + +Related but different: #120 (512 KiB activity decoder), #273 (MCP key collision + attempt cap). This incident uses the #273 cap with a different error. + +## Non-goals + +- Do **not** only raise the 8 MiB cap. +- Do **not** blanket-refactor Claude / OpenCode / unrelated adapters. +- Do **not** claim Cursor/Grok are “safe” from absence of logs. +- Do **not** push, retag, or cut a Release from this worktree until the release watcher reconciles #316 (parent hold). Existing CI on other branches may finish. +- Do **not** touch PR #314 / release/v0.15.2 worktree contents except by landing a normal fix PR later. + +## Provider impact (evidence classes) + +| Provider | Transport | Bound | This exact mechanism | +| ------------------------ | ----------------------------------------------------- | ------------------------- | --------------------------------------- | +| Codex chat | stdio JSON-RPC | 8 MiB | **Proven** (incident + synthetic) | +| Codex terminal remote WS | same max on frames; 512 KiB activity decoder separate | related bound | **Code-only**; not this Delivery banner | +| Cursor ACP | stdio JSON-RPC (`cursor/acp.rs`) | **128 KiB** | **Code-only susceptible** (tighter) | +| Grok | re-exports Cursor ACP | **128 KiB** | **Code-only susceptible** | +| Claude | Agent SDK / stream-json | no LineTooLong in adapter | **Not this mechanism** | +| OpenCode | HTTP/session API | no stdout LineTooLong | **Not this mechanism** | + +Shared amplifier: durable delivery + attempt cap for `DefinitelyNotSent`. Fix classification may live in shared launch outcome mapping; adapter-specific framing stays local. + +## Options + +### A. Classify deterministic transport size closes as non-retryable + privacy-safe diagnostics (recommended P0) + +1. When Codex (and, if the same string/path is shared, ACP) fails a launch with a nested Closed/LineTooLong (“read line exceeded … bytes on stdout”), map **`launch_failure_outcome`** (and the frozen path if it currently Rejects already) so delivery **Rejects once** instead of burning 20 relaunches. Match on structured error / stable substring already present in Display — do not scrape payloads. +2. On `LineTooLong`, log **privacy-safe** fields only: stream, maximum, bytes accumulated since last LF (or “at least maximum+1”), whether an LF was seen before fail, and **pending request method** if one is registered (no body, no thread/project content). +3. Regression tests: + - protocol: nested Closed Display string for over-bound stdout (Codex); optional twin for ACP 128 KiB only if a shared helper is extracted or ACP diagnostics are touched. + - delivery: a launch failure whose detail is that nested string becomes **Rejected** on first attempt (no 20-storm), profiler still records failing with a distinct error class if one exists or `attempts-exhausted` equivalent for immediate reject — prefer a clear class such as `transport-line-too-long` if profiler already allows custom classes; otherwise keep existing Rejected path with the detail string and assert attempts stay at 1. + +Stops the automatic retry storm immediately. The failed head still holds later queued messages until Retry or Dismiss; this does not by itself prevent an oversized frame on a later resume of the same history. + +### B. Bound startup fetches (P1, same PR if small and proven) + +Where Codex app-server accepts it on 0.167.0: request **metadata-only / exclude-turns** (or equivalent) on `thread/resume` so history is not one giant stdout line. Verify against Codex docs + a unit/integration expectation on the outbound params. If the option is absent or behavior unverified on 0.167.0, leave a follow-up and do **not** guess params. + +Optional: tighten MCP status / turns/list paging only with evidence; no speculative catalogue rewrite. + +NOTE (#316, evidence update): `excludeTurns` is present in the installed 0.167.0 stable schema. Synthetic resume probes did not show oversized default resume lines; MCP `toolsAndAuthOnly` did. See Evidence update section. + +### C. Raise transport cap only + +Rejected as sole fix: hides the emitter, still can OOM, does not stop deterministic relaunch if history grows past any new cap. + +## Decision + +Ship **A** now. Include **B** only if outbound params are documented for this Codex version and tests assert them; otherwise open a follow-up comment on #316. No C alone. + +Cursor/Grok: if classification keys off the shared nested “read line exceeded” detail string at `launch_failure_outcome`, they inherit the non-retry behavior without an ACP rewrite. Add an ACP LineTooLong diagnostic/regression **only** when touching `cursor/acp.rs` for shared helper extraction; otherwise leave ACP framing alone (code-only susceptibility noted on the issue). + +## Acceptance + +- Failing-then-passing regression: over-bound Codex stdout → nested Closed string → delivery Rejected without 20 attempts. +- Privacy-safe LineTooLong log fields present in unit or tracing test hooks as the repo usually asserts. +- `cargo fmt`, `clippy -D warnings` on touched crates, and the focused server tests for protocol + delivery/launch failure pass. +- Design note lands in the same PR as the code. +- No push/tag/Release from the implementing agent until parent release hold lifts. + +## Historical pre-fix recovery advice (already on #316) + +Copy prompt → **new Codex chat** (`thread/start`). Avoid Retry on the stuck thread. Dismiss unblocks queue but same-thread resume remains risky until A/B ship. + +## Evidence update (2026-11-06 ART, isolated Codex 0.167.0 probes) + +Source: installed `codex-cli 0.167.0` `app-server generate-json-schema` + throwaway `CODEX_HOME` probes under `/tmp/codex-316-*` (byte lengths / ok codes only; no payloads; no live user threads). + +1. **`excludeTurns` is real on 0.167.0** (stable `ThreadResumeParams`). Earlier deferral that “docs don’t pin 0.167.0” is superseded by the installed schema bundle. +2. **Synthetic oversized rollout history did not inflate `thread/resume` stdout** (legacy/paginated, 3–9 MiB synthetic items → ~1.7 KiB resume, empty `turns`). So resume history is **not** proven as the offending frame; cursor presence alone is insufficient. +3. **`mcpServerStatus/list` with `toolsAndAuthOnly` can exceed 8 MiB** (synthetic MCP tool description 9e6 chars → **9000517**-byte JSON-RPC line). This is a proven LineTooLong-capable mechanism and matches BiBCode’s launch `mcp_status.rs` call shape. Live pending-method confirmation still required. +4. **Option A remains partial mitigation** (stop retry storm + diagnostics). **Option B** is schema-safe defense-in-depth, not a proven incident fix. **Causal same-thread fix** should target the confirmed oversized-response class (MCP catalogue bound/avoidance and/or transport handling) once live `pending_methods` confirms, without guessing resume. + +## Decision update — MCP catalogue (lab-proven startup frame) + +Self-approved for this defect. Installed Codex 0.167.0 schema offers only `full` | `toolsAndAuthOnly` for `mcpServerStatus/list`; both embed tool schemas. BiBCode’s pre-fix mapper (`mcp_server_status_from_snapshot`) kept only `name` + `authStatus`. Lab: `toolsAndAuthOnly` can return a **9_000_517**-byte stdout line and kill the shared chat transport. + +**Chosen smallest compatible fix:** remove `mcpServerStatus/list` discovery from the launch/refresh path. +Opening publishes notification observations; refresh republishes the committed +snapshot without provider I/O. Remove the loader and its paging, timeout, +overlay-load and generation machinery, including the draft's test-only copy. +Keep bounded pre-root staging, cancellation ownership, root filtering, and the +ordered publication/completion worker. A successful opening replaces status +with the new connection's observations; failed/aborted reconnect retains the +old snapshot. Tool availability for the model remains Codex MCP runtime. +Auth/status UI uses `mcpServer/startupStatus/updated` notifications. Servers +without notifications and auth state known only from the catalogue may be +absent: this is a deliberate status-discovery limitation, not full parity with +the former list-backed baseline. + +**Not chosen:** raise the 8 MiB cap; add `excludeTurns` as a causal fix; treat +the lab MCP frame as proof of the live emitter. `excludeTurns` is real in the +installed 0.167.0 stable schema, but the synthetic history probes did not +establish it as a fix for this failure. No resume parameter is added here. + +**Proof target:** `mcp_catalogue_oversize_thread_resumes_and_completes_followup_turns` +drives `CodexSessionRuntime::start` through new-thread startup and existing-thread +resume, refresh, two completed turns on the same provider thread, and shutdown. +The synthetic peer would emit an oversized catalogue response if requested. +The MCP actor/runtime tests cover refresh retention, root filtering, publication +ordering, failed/aborted opening, and successful reconnect. + +**Scope:** lab-proven MCP startup fix ≠ confirmed production pending method. Option A remains partial (retry storm). Same-thread continuity for the MCP mechanism is restored by keeping the transport alive through launch. + +## Migration implementation scope + +The initial Option A decision above is superseded for this combined draft: +Option A remains the fallback for deterministic transport-size failures, and +the notification-only MCP path addresses same-thread continuation for the +lab-proven startup mechanism. A failed durable message remains visible with +Retry/Dismiss; retry after applying this fix can resume the same provider +thread when MCP catalogue discovery was the cause. This patch does not +silently replay failed messages, discard history, change databases, or restart +production. Issue #316 stays open and the release hold stays in place until +production emitter attribution and the remaining validation are resolved. + +Validation results and host-specific commands belong in the handoff report, +not in the living runbook. No commit is permitted before a read-only Codex +review reports NO FINDINGS on the combined diff. diff --git a/docs/skills/bibcode-profiler/references/findings.md b/docs/skills/bibcode-profiler/references/findings.md index d01a20ed3b..9ab7b32c54 100644 --- a/docs/skills/bibcode-profiler/references/findings.md +++ b/docs/skills/bibcode-profiler/references/findings.md @@ -329,10 +329,13 @@ occurrence. - **Raised when:** a message BiBCode was delivering to a provider fails for good without reaching it, once per delivery: the provider refused its own configuration (`provider-configuration`; Codex answers `thread/start` or - `thread/resume` with `-32600` "failed to load configuration"), or the twentieth + `thread/resume` with `-32600` "failed to load configuration"), a Codex, Cursor, + or Grok launch closed by an oversized stdio line (`transport-line-too-long`, + #316), or the twentieth attempt in a row ended not sent (`attempts-exhausted`). After a restart, a message left sending whose check cannot start the provider raises the same: - `provider-configuration` when the provider refuses its configuration, and + `provider-configuration` when the provider refuses its configuration, + `transport-line-too-long` after one oversized transport close, and `attempts-exhausted` after the twentieth failed launch in that run; the message is then uncertain and is not sent again by itself. The thread shows the message as not delivered, with the provider's reason, until the person retries or dismisses diff --git a/docs/testing/cross-platform-validation.md b/docs/testing/cross-platform-validation.md index 8eead9aec1..69110c92b9 100644 --- a/docs/testing/cross-platform-validation.md +++ b/docs/testing/cross-platform-validation.md @@ -388,6 +388,42 @@ over the guard threshold shows "Waiting for the conversation to be compacted." then the compaction, then the answer; Stop during that compaction leaves the message failed with Retry. +For transport-size launch failures (#316), the Codex protocol regression feeds +stdout past the real 8 MiB bound with and without a trailing LF, checks the +nested Closed detail, and captures only framing counts and pending method names. +The runtime regression drives that close through the supervisor and durable +delivery worker: the row must fail with one attempt, one launch, and one +`delivery.failing` finding of class `transport-line-too-long`. Direct and nested +ACP errors share the launch classification for Cursor and Grok. Restart recovery +of a row left sending must stop after one oversized launch, keep the row +uncertain across delivery-service restart, and record the same profiler class. +Run: + +```sh +cargo test -p bibcode-server --lib oversized_stdout -j 2 +cargo test -p bibcode-server --lib transport_line_too_long -j 2 +cargo test -p bibcode-server --lib mcp_catalogue -j 2 +cargo test -p bibcode-server --lib mcp_status -j 2 +``` + +The MCP catalogue fixture answers any `mcpServerStatus/list` request with a +synthetic response above the stdout bound. Both new-thread startup and +same-thread resume must reach ready, publish notification-derived status, and +complete two followup turns on the same provider thread. Refresh must preserve +the observed status without requesting the catalogue. The actor/runtime tests +also cover notification filtering, reconnect replacement, cancellation, +concurrent refresh during opening, and ordered publication. These fixtures +prove the MCP mechanism; they do not identify a production incident's emitter. + +For a native UI check, use a disposable provider home and BiBCode data root +with synthetic threads and MCP fixtures. Confirm connected statuses and +actionable authentication failures from startup notifications (an unauthenticated +OAuth server may show Error with login instructions), refresh retention, and +continuation in the same thread. A server with no notification may be absent from the status popover; +model tool availability is independent of that popover. Do not resume live user +threads to reproduce this failure or collect provider payloads. Pending-method +logs provide attribution without conversation data. + Deliver a queued message at completion and let its turn run past the idle timeout; the session must stay live until one idle timeout after that turn completes. Likewise, end a turn while a provider subagent keeps running: the session must stay diff --git a/docs/user/workspace-ui.md b/docs/user/workspace-ui.md index 61894cbcc7..ff9f94b7ae 100644 --- a/docs/user/workspace-ui.md +++ b/docs/user/workspace-ui.md @@ -860,6 +860,11 @@ Disabled providers still show the corresponding control with an unavailable tooltip, but the control cannot open a status popover. Stale activity does not override the selected provider's capability. +For Codex, the MCP popover shows status notifications from the current provider +connection. Servers that have not reported a status may be absent, and an empty +popover shows **Awaiting MCP status**. The popover does not determine which MCP +tools the model can use. + For Claude, the MCP popover starts with the status reported during session initialization and refreshes after successful provider responses. It preserves the last valid snapshot if a refresh is unavailable.