diff --git a/src/codex_agent.rs b/src/codex_agent.rs index b9cb0c00..d2335dd9 100644 --- a/src/codex_agent.rs +++ b/src/codex_agent.rs @@ -34,7 +34,9 @@ use codex_thread_store::{ }; use std::{ collections::HashMap, + future::Future, path::{Path, PathBuf}, + pin::Pin, sync::{Arc, Mutex}, }; use tracing::{debug, info}; @@ -123,187 +125,7 @@ impl CodexAgent { self: Arc, transport: impl ConnectTo + 'static, ) -> acp::Result<()> { - let agent = self; - Agent - .builder() - .name("codex-acp") - .on_receive_request( - { - let agent = agent.clone(); - async move |request: InitializeRequest, responder, _cx| { - responder.respond_with_result(agent.initialize(request).await) - } - }, - acp::on_receive_request!(), - ) - .on_receive_request( - { - let agent = agent.clone(); - async move |request: AuthenticateRequest, - responder, - cx: ConnectionTo| { - let agent = agent.clone(); - cx.spawn(async move { - responder.respond_with_result(agent.authenticate(request).await) - })?; - Ok(()) - } - }, - acp::on_receive_request!(), - ) - .on_receive_request( - { - let agent = agent.clone(); - async move |request: LogoutRequest, responder, cx: ConnectionTo| { - let agent = agent.clone(); - cx.spawn(async move { - responder.respond_with_result(agent.logout(request).await) - })?; - Ok(()) - } - }, - acp::on_receive_request!(), - ) - .on_receive_request( - { - let agent = agent.clone(); - async move |request: NewSessionRequest, responder, cx: ConnectionTo| { - let agent = agent.clone(); - let session_cx = cx.clone(); - cx.spawn(async move { - responder - .respond_with_result(agent.new_session(request, session_cx).await) - })?; - Ok(()) - } - }, - acp::on_receive_request!(), - ) - .on_receive_request( - { - let agent = agent.clone(); - async move |request: LoadSessionRequest, responder, cx: ConnectionTo| { - let agent = agent.clone(); - let session_cx = cx.clone(); - cx.spawn(async move { - responder - .respond_with_result(agent.load_session(request, session_cx).await) - })?; - Ok(()) - } - }, - acp::on_receive_request!(), - ) - .on_receive_request( - { - let agent = agent.clone(); - async move |request: ResumeSessionRequest, - responder, - cx: ConnectionTo| { - let agent = agent.clone(); - let session_cx = cx.clone(); - cx.spawn(async move { - responder.respond_with_result( - agent.resume_session(request, session_cx).await, - ) - })?; - Ok(()) - } - }, - acp::on_receive_request!(), - ) - .on_receive_request( - { - let agent = agent.clone(); - async move |request: ListSessionsRequest, - responder, - cx: ConnectionTo| { - let agent = agent.clone(); - cx.spawn(async move { - responder.respond_with_result(agent.list_sessions(request).await) - })?; - Ok(()) - } - }, - acp::on_receive_request!(), - ) - .on_receive_request( - { - let agent = agent.clone(); - async move |request: CloseSessionRequest, - responder, - cx: ConnectionTo| { - let agent = agent.clone(); - cx.spawn(async move { - responder.respond_with_result(agent.close_session(request).await) - })?; - Ok(()) - } - }, - acp::on_receive_request!(), - ) - .on_receive_request( - { - let agent = agent.clone(); - async move |request: PromptRequest, responder, cx: ConnectionTo| { - let agent = agent.clone(); - cx.spawn(async move { - responder.respond_with_result(agent.prompt(request).await) - })?; - Ok(()) - } - }, - acp::on_receive_request!(), - ) - .on_receive_notification( - { - let agent = agent.clone(); - async move |notification: CancelNotification, cx: ConnectionTo| { - let agent = agent.clone(); - cx.spawn(async move { - if let Err(e) = agent.cancel(notification).await { - tracing::error!("Error handling cancel: {:?}", e); - } - Ok(()) - })?; - Ok(()) - } - }, - acp::on_receive_notification!(), - ) - .on_receive_request( - { - let agent = agent.clone(); - async move |request: SetSessionModeRequest, - responder, - cx: ConnectionTo| { - let agent = agent.clone(); - cx.spawn(async move { - responder.respond_with_result(agent.set_session_mode(request).await) - })?; - Ok(()) - } - }, - acp::on_receive_request!(), - ) - .on_receive_request( - { - let agent = agent.clone(); - async move |request: SetSessionConfigOptionRequest, - responder, - cx: ConnectionTo| { - let agent = agent.clone(); - cx.spawn(async move { - responder - .respond_with_result(agent.set_session_config_option(request).await) - })?; - Ok(()) - } - }, - acp::on_receive_request!(), - ) - .connect_to(transport) - .await + serve_agent_api(self, transport).await } fn session_id_from_thread_id(thread_id: ThreadId) -> SessionId { @@ -436,6 +258,311 @@ impl CodexAgent { } } +type AgentApiFuture<'a, T> = Pin> + Send + 'a>>; + +trait AgentApi: Send + Sync + 'static { + fn initialize(&self, request: InitializeRequest) -> AgentApiFuture<'_, InitializeResponse>; + + fn authenticate( + &self, + request: AuthenticateRequest, + ) -> AgentApiFuture<'_, AuthenticateResponse>; + + fn logout(&self, request: LogoutRequest) -> AgentApiFuture<'_, LogoutResponse>; + + fn new_session( + &self, + request: NewSessionRequest, + cx: ConnectionTo, + ) -> AgentApiFuture<'_, NewSessionResponse>; + + fn load_session( + &self, + request: LoadSessionRequest, + cx: ConnectionTo, + ) -> AgentApiFuture<'_, LoadSessionResponse>; + + fn resume_session( + &self, + request: ResumeSessionRequest, + cx: ConnectionTo, + ) -> AgentApiFuture<'_, ResumeSessionResponse>; + + fn list_sessions( + &self, + request: ListSessionsRequest, + ) -> AgentApiFuture<'_, ListSessionsResponse>; + + fn close_session( + &self, + request: CloseSessionRequest, + ) -> AgentApiFuture<'_, CloseSessionResponse>; + + fn prompt(&self, request: PromptRequest) -> AgentApiFuture<'_, PromptResponse>; + + fn cancel(&self, notification: CancelNotification) -> AgentApiFuture<'_, ()>; + + fn set_session_mode( + &self, + request: SetSessionModeRequest, + ) -> AgentApiFuture<'_, SetSessionModeResponse>; + + fn set_session_config_option( + &self, + request: SetSessionConfigOptionRequest, + ) -> AgentApiFuture<'_, SetSessionConfigOptionResponse>; +} + +impl AgentApi for CodexAgent { + fn initialize(&self, request: InitializeRequest) -> AgentApiFuture<'_, InitializeResponse> { + Box::pin(CodexAgent::initialize(self, request)) + } + + fn authenticate( + &self, + request: AuthenticateRequest, + ) -> AgentApiFuture<'_, AuthenticateResponse> { + Box::pin(CodexAgent::authenticate(self, request)) + } + + fn logout(&self, request: LogoutRequest) -> AgentApiFuture<'_, LogoutResponse> { + Box::pin(CodexAgent::logout(self, request)) + } + + fn new_session( + &self, + request: NewSessionRequest, + cx: ConnectionTo, + ) -> AgentApiFuture<'_, NewSessionResponse> { + Box::pin(CodexAgent::new_session(self, request, cx)) + } + + fn load_session( + &self, + request: LoadSessionRequest, + cx: ConnectionTo, + ) -> AgentApiFuture<'_, LoadSessionResponse> { + Box::pin(CodexAgent::load_session(self, request, cx)) + } + + fn resume_session( + &self, + request: ResumeSessionRequest, + cx: ConnectionTo, + ) -> AgentApiFuture<'_, ResumeSessionResponse> { + Box::pin(CodexAgent::resume_session(self, request, cx)) + } + + fn list_sessions( + &self, + request: ListSessionsRequest, + ) -> AgentApiFuture<'_, ListSessionsResponse> { + Box::pin(CodexAgent::list_sessions(self, request)) + } + + fn close_session( + &self, + request: CloseSessionRequest, + ) -> AgentApiFuture<'_, CloseSessionResponse> { + Box::pin(CodexAgent::close_session(self, request)) + } + + fn prompt(&self, request: PromptRequest) -> AgentApiFuture<'_, PromptResponse> { + Box::pin(CodexAgent::prompt(self, request)) + } + + fn cancel(&self, notification: CancelNotification) -> AgentApiFuture<'_, ()> { + Box::pin(CodexAgent::cancel(self, notification)) + } + + fn set_session_mode( + &self, + request: SetSessionModeRequest, + ) -> AgentApiFuture<'_, SetSessionModeResponse> { + Box::pin(CodexAgent::set_session_mode(self, request)) + } + + fn set_session_config_option( + &self, + request: SetSessionConfigOptionRequest, + ) -> AgentApiFuture<'_, SetSessionConfigOptionResponse> { + Box::pin(CodexAgent::set_session_config_option(self, request)) + } +} + +async fn serve_agent_api( + agent: Arc, + transport: impl ConnectTo + 'static, +) -> acp::Result<()> { + Agent + .builder() + .name("codex-acp") + .on_receive_request( + { + let agent = agent.clone(); + async move |request: InitializeRequest, responder, _cx| { + responder.respond_with_result(agent.initialize(request).await) + } + }, + acp::on_receive_request!(), + ) + .on_receive_request( + { + let agent = agent.clone(); + async move |request: AuthenticateRequest, responder, cx: ConnectionTo| { + let agent = agent.clone(); + cx.spawn(async move { + responder.respond_with_result(agent.authenticate(request).await) + })?; + Ok(()) + } + }, + acp::on_receive_request!(), + ) + .on_receive_request( + { + let agent = agent.clone(); + async move |request: LogoutRequest, responder, cx: ConnectionTo| { + let agent = agent.clone(); + cx.spawn( + async move { responder.respond_with_result(agent.logout(request).await) }, + )?; + Ok(()) + } + }, + acp::on_receive_request!(), + ) + .on_receive_request( + { + let agent = agent.clone(); + async move |request: NewSessionRequest, responder, cx: ConnectionTo| { + let agent = agent.clone(); + let session_cx = cx.clone(); + cx.spawn(async move { + responder.respond_with_result(agent.new_session(request, session_cx).await) + })?; + Ok(()) + } + }, + acp::on_receive_request!(), + ) + .on_receive_request( + { + let agent = agent.clone(); + async move |request: LoadSessionRequest, responder, cx: ConnectionTo| { + let agent = agent.clone(); + let session_cx = cx.clone(); + cx.spawn(async move { + responder.respond_with_result(agent.load_session(request, session_cx).await) + })?; + Ok(()) + } + }, + acp::on_receive_request!(), + ) + .on_receive_request( + { + let agent = agent.clone(); + async move |request: ResumeSessionRequest, responder, cx: ConnectionTo| { + let agent = agent.clone(); + let session_cx = cx.clone(); + cx.spawn(async move { + responder + .respond_with_result(agent.resume_session(request, session_cx).await) + })?; + Ok(()) + } + }, + acp::on_receive_request!(), + ) + .on_receive_request( + { + let agent = agent.clone(); + async move |request: ListSessionsRequest, responder, cx: ConnectionTo| { + let agent = agent.clone(); + cx.spawn(async move { + responder.respond_with_result(agent.list_sessions(request).await) + })?; + Ok(()) + } + }, + acp::on_receive_request!(), + ) + .on_receive_request( + { + let agent = agent.clone(); + async move |request: CloseSessionRequest, responder, cx: ConnectionTo| { + let agent = agent.clone(); + cx.spawn(async move { + responder.respond_with_result(agent.close_session(request).await) + })?; + Ok(()) + } + }, + acp::on_receive_request!(), + ) + .on_receive_request( + { + let agent = agent.clone(); + async move |request: PromptRequest, responder, cx: ConnectionTo| { + let agent = agent.clone(); + cx.spawn( + async move { responder.respond_with_result(agent.prompt(request).await) }, + )?; + Ok(()) + } + }, + acp::on_receive_request!(), + ) + .on_receive_notification( + { + let agent = agent.clone(); + async move |notification: CancelNotification, cx: ConnectionTo| { + let agent = agent.clone(); + cx.spawn(async move { + if let Err(e) = agent.cancel(notification).await { + tracing::error!("Error handling cancel: {:?}", e); + } + Ok(()) + })?; + Ok(()) + } + }, + acp::on_receive_notification!(), + ) + .on_receive_request( + { + let agent = agent.clone(); + async move |request: SetSessionModeRequest, responder, cx: ConnectionTo| { + let agent = agent.clone(); + cx.spawn(async move { + responder.respond_with_result(agent.set_session_mode(request).await) + })?; + Ok(()) + } + }, + acp::on_receive_request!(), + ) + .on_receive_request( + { + let agent = agent.clone(); + async move |request: SetSessionConfigOptionRequest, + responder, + cx: ConnectionTo| { + let agent = agent.clone(); + cx.spawn(async move { + responder + .respond_with_result(agent.set_session_config_option(request).await) + })?; + Ok(()) + } + }, + acp::on_receive_request!(), + ) + .connect_to(transport) + .await +} + impl CodexAgent { async fn initialize(&self, request: InitializeRequest) -> Result { let InitializeRequest { @@ -948,6 +1075,10 @@ fn stored_session_title(name: Option<&str>, preview: &str) -> Option { mod tests { use super::*; + mod api_command_tests { + include!("codex_agent/api_command_tests.rs"); + } + #[test] fn stored_session_title_prefers_thread_name() { assert_eq!( diff --git a/src/codex_agent/api_command_tests.rs b/src/codex_agent/api_command_tests.rs new file mode 100644 index 00000000..db6d2047 --- /dev/null +++ b/src/codex_agent/api_command_tests.rs @@ -0,0 +1,265 @@ +use super::super::*; +use acp::schema::{ + ContentBlock, SessionConfigOptionValue, SessionConfigValueId, StopReason, TextContent, +}; +use std::{ + sync::{Arc, Mutex}, + time::Duration, +}; + +#[derive(Default)] +struct RecordingAgentApi { + calls: Mutex>, +} + +impl RecordingAgentApi { + fn record(&self, name: &'static str) { + self.calls.lock().unwrap().push(name); + } + + fn calls(&self) -> Vec<&'static str> { + self.calls.lock().unwrap().clone() + } +} + +impl AgentApi for RecordingAgentApi { + fn initialize(&self, _request: InitializeRequest) -> AgentApiFuture<'_, InitializeResponse> { + Box::pin(async { + self.record("initialize"); + Ok(InitializeResponse::new(ProtocolVersion::V1)) + }) + } + + fn authenticate( + &self, + _request: AuthenticateRequest, + ) -> AgentApiFuture<'_, AuthenticateResponse> { + Box::pin(async { + self.record("authenticate"); + Ok(AuthenticateResponse::new()) + }) + } + + fn logout(&self, _request: LogoutRequest) -> AgentApiFuture<'_, LogoutResponse> { + Box::pin(async { + self.record("logout"); + Ok(LogoutResponse::new()) + }) + } + + fn new_session( + &self, + _request: NewSessionRequest, + _cx: ConnectionTo, + ) -> AgentApiFuture<'_, NewSessionResponse> { + Box::pin(async { + self.record("new_session"); + Ok(NewSessionResponse::new(SessionId::new("new-session"))) + }) + } + + fn load_session( + &self, + _request: LoadSessionRequest, + _cx: ConnectionTo, + ) -> AgentApiFuture<'_, LoadSessionResponse> { + Box::pin(async { + self.record("load_session"); + Ok(LoadSessionResponse::new()) + }) + } + + fn resume_session( + &self, + _request: ResumeSessionRequest, + _cx: ConnectionTo, + ) -> AgentApiFuture<'_, ResumeSessionResponse> { + Box::pin(async { + self.record("resume_session"); + Ok(ResumeSessionResponse::new()) + }) + } + + fn list_sessions( + &self, + _request: ListSessionsRequest, + ) -> AgentApiFuture<'_, ListSessionsResponse> { + Box::pin(async { + self.record("list_sessions"); + Ok(ListSessionsResponse::new(vec![])) + }) + } + + fn close_session( + &self, + _request: CloseSessionRequest, + ) -> AgentApiFuture<'_, CloseSessionResponse> { + Box::pin(async { + self.record("close_session"); + Ok(CloseSessionResponse::new()) + }) + } + + fn prompt(&self, _request: PromptRequest) -> AgentApiFuture<'_, PromptResponse> { + Box::pin(async { + self.record("prompt"); + Ok(PromptResponse::new(StopReason::EndTurn)) + }) + } + + fn cancel(&self, _notification: CancelNotification) -> AgentApiFuture<'_, ()> { + Box::pin(async { + self.record("cancel"); + Ok(()) + }) + } + + fn set_session_mode( + &self, + _request: SetSessionModeRequest, + ) -> AgentApiFuture<'_, SetSessionModeResponse> { + Box::pin(async { + self.record("set_session_mode"); + Ok(SetSessionModeResponse::new()) + }) + } + + fn set_session_config_option( + &self, + _request: SetSessionConfigOptionRequest, + ) -> AgentApiFuture<'_, SetSessionConfigOptionResponse> { + Box::pin(async { + self.record("set_session_config_option"); + Ok(SetSessionConfigOptionResponse::new(vec![])) + }) + } +} + +struct TestAgentComponent(Arc); + +impl ConnectTo for TestAgentComponent { + fn connect_to( + self, + client: impl ConnectTo, + ) -> impl std::future::Future> + Send { + serve_agent_api(self.0, client) + } +} + +async fn wait_for_call(api: &RecordingAgentApi, name: &'static str) -> Result<(), Error> { + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if api.calls().contains(&name) { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .map_err(|_| Error::internal_error().data(format!("timed out waiting for {name}")))?; + + Ok(()) +} + +#[tokio::test] +async fn serve_dispatches_all_acp_requests_and_notifications() -> anyhow::Result<()> { + let api = Arc::new(RecordingAgentApi::default()); + let cwd = std::env::current_dir()?; + let api_for_client = api.clone(); + + Client + .connect_with(TestAgentComponent(api.clone()), async |cx| { + cx.send_request(InitializeRequest::new(ProtocolVersion::V1)) + .block_task() + .await?; + cx.send_request(AuthenticateRequest::new(AuthMethodId::new("chatgpt"))) + .block_task() + .await?; + cx.send_request(LogoutRequest::new()).block_task().await?; + cx.send_request(NewSessionRequest::new(cwd.clone())) + .block_task() + .await?; + cx.send_request(LoadSessionRequest::new( + SessionId::new("load-session"), + cwd.clone(), + )) + .block_task() + .await?; + cx.send_request(ResumeSessionRequest::new( + SessionId::new("resume-session"), + cwd.clone(), + )) + .block_task() + .await?; + cx.send_request(ListSessionsRequest::new()) + .block_task() + .await?; + cx.send_request(CloseSessionRequest::new(SessionId::new("close-session"))) + .block_task() + .await?; + cx.send_request(PromptRequest::new( + SessionId::new("prompt-session"), + vec![ContentBlock::Text(TextContent::new("hello"))], + )) + .block_task() + .await?; + cx.send_notification(CancelNotification::new(SessionId::new("cancel-session")))?; + cx.send_request(SetSessionModeRequest::new( + SessionId::new("mode-session"), + "read-only", + )) + .block_task() + .await?; + cx.send_request(SetSessionConfigOptionRequest::new( + SessionId::new("config-session"), + "model", + SessionConfigOptionValue::value_id(SessionConfigValueId::new("gpt-5")), + )) + .block_task() + .await?; + + wait_for_call(&api_for_client, "cancel").await?; + + Ok(()) + }) + .await?; + + let mut calls = api.calls(); + calls.sort_unstable(); + assert_eq!( + calls, + vec![ + "authenticate", + "cancel", + "close_session", + "initialize", + "list_sessions", + "load_session", + "logout", + "new_session", + "prompt", + "resume_session", + "set_session_config_option", + "set_session_mode", + ] + ); + + Ok(()) +} + +#[test] +fn auth_method_ids_round_trip() { + for method in [ + CodexAuthMethod::ChatGpt, + CodexAuthMethod::CodexApiKey, + CodexAuthMethod::OpenAiApiKey, + ] { + let id = AuthMethodId::from(method); + assert_eq!(CodexAuthMethod::try_from(id).unwrap(), method); + } +} + +#[test] +fn unsupported_auth_method_is_rejected() { + assert!(CodexAuthMethod::try_from(AuthMethodId::new("unsupported")).is_err()); +} diff --git a/src/thread.rs b/src/thread.rs index fd1f1c26..5a46228b 100644 --- a/src/thread.rs +++ b/src/thread.rs @@ -4118,6 +4118,10 @@ mod tests { use super::*; + mod api_command_tests { + include!("thread/api_command_tests.rs"); + } + #[tokio::test] async fn test_prompt() -> anyhow::Result<()> { let (session_id, client, _, message_tx, _handle) = setup().await?; @@ -4682,9 +4686,12 @@ mod tests { impl ModelsManagerImpl for StubModelsManager { fn get_model( &self, - _model_id: &Option, + model_id: &Option, ) -> Pin + Send + '_>> { - Box::pin(async { all_model_presets()[0].to_owned().id }) + let model_id = model_id.clone(); + Box::pin( + async move { model_id.unwrap_or_else(|| all_model_presets()[0].to_owned().id) }, + ) } fn list_models(&self) -> Pin> + Send + '_>> { @@ -5020,6 +5027,7 @@ mod tests { | Op::ResolveElicitation { .. } | Op::RequestPermissionsResponse { .. } | Op::PatchApproval { .. } + | Op::ThreadSettings { .. } | Op::Interrupt => {} Op::Shutdown => { if let Some(active_prompt_id) = self.active_prompt_id.lock().unwrap().take() diff --git a/src/thread/api_command_tests.rs b/src/thread/api_command_tests.rs new file mode 100644 index 00000000..4d61bbd1 --- /dev/null +++ b/src/thread/api_command_tests.rs @@ -0,0 +1,428 @@ +use super::*; +use agent_client_protocol::schema::SessionNotification; +use std::{ + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; + +#[test] +fn build_prompt_items_maps_supported_prompt_content_blocks() { + let items = build_prompt_items(vec![ + ContentBlock::Text(TextContent::new("hello")), + ContentBlock::Image(ImageContent::new("Zm9v", "image/png")), + ContentBlock::ResourceLink(ResourceLink::new("file.rs", "file:///tmp/file.rs")), + ContentBlock::Resource(EmbeddedResource::new( + EmbeddedResourceResource::TextResourceContents(TextResourceContents::new( + "let x = 1;", + "file:///tmp/file.rs", + )), + )), + ]); + + assert_eq!(items.len(), 4); + assert!(matches!( + &items[0], + UserInput::Text { text, .. } if text == "hello" + )); + assert!(matches!( + &items[1], + UserInput::Image { image_url, .. } if image_url == "data:image/png;base64,Zm9v" + )); + assert!(matches!( + &items[2], + UserInput::Text { text, .. } if text == "[@file.rs](file:///tmp/file.rs)" + )); + assert!(matches!( + &items[3], + UserInput::Text { text, .. } + if text == "[@file.rs](file:///tmp/file.rs)\n\nlet x = 1;\n" + )); +} + +#[test] +fn builtin_commands_cover_advertised_slash_commands() { + let commands = ThreadActor::::builtin_commands(); + let command_names = commands + .iter() + .map(|command| command.name.as_str()) + .collect::>(); + + assert_eq!( + command_names, + vec![ + "review", + "review-branch", + "review-commit", + "init", + "compact", + "logout", + ] + ); + assert!(commands[0].input.is_some()); + assert!(commands[1].input.is_some()); + assert!(commands[2].input.is_some()); + assert!(commands[3].input.is_none()); + assert!(commands[4].input.is_none()); + assert!(commands[5].input.is_none()); +} + +#[tokio::test] +async fn load_returns_config_options_and_announces_builtin_commands() -> anyhow::Result<()> { + let (_session_id, client, _, message_tx, _handle) = setup().await?; + let (response_tx, response_rx) = tokio::sync::oneshot::channel(); + + message_tx.send(ThreadMessage::Load { response_tx })?; + + let response = response_rx.await??; + assert_config_option_ids_include( + response + .config_options + .as_deref() + .expect("load response should include config options"), + &["mode", "model"], + ); + + wait_for_notification(client.as_ref(), |notification| { + matches!( + ¬ification.update, + SessionUpdate::AvailableCommandsUpdate(update) + if update + .available_commands + .iter() + .map(|command| command.name.as_str()) + .collect::>() + == vec![ + "review", + "review-branch", + "review-commit", + "init", + "compact", + "logout", + ] + ) + }) + .await?; + + drop(message_tx); + Ok(()) +} + +#[tokio::test] +async fn get_config_options_returns_mode_and_model_options() -> anyhow::Result<()> { + let (_session_id, _client, _, message_tx, _handle) = setup().await?; + let (response_tx, response_rx) = tokio::sync::oneshot::channel(); + + message_tx.send(ThreadMessage::GetConfigOptions { response_tx })?; + + let options = response_rx.await??; + assert_config_option_ids_include(&options, &["mode", "model"]); + + drop(message_tx); + Ok(()) +} + +#[tokio::test] +async fn set_mode_submits_thread_settings_and_emits_config_options() -> anyhow::Result<()> { + let (_session_id, client, thread, message_tx, _handle) = setup().await?; + let (response_tx, response_rx) = tokio::sync::oneshot::channel(); + + message_tx.send(ThreadMessage::SetMode { + mode: SessionModeId::new("read-only"), + response_tx, + })?; + + response_rx.await??; + + let ops = thread.ops.lock().unwrap(); + assert!( + matches!(ops.last(), Some(Op::ThreadSettings { .. })), + "expected set mode to submit ThreadSettings, got {ops:?}" + ); + drop(ops); + + wait_for_notification(client.as_ref(), |notification| { + matches!( + ¬ification.update, + SessionUpdate::ConfigOptionUpdate(update) + if update + .config_options + .iter() + .any(|option| option.id.0.as_ref() == "mode") + ) + }) + .await?; + + drop(message_tx); + Ok(()) +} + +#[tokio::test] +async fn set_config_option_model_submits_thread_settings() -> anyhow::Result<()> { + let (_session_id, _client, thread, message_tx, _handle) = setup().await?; + let preset = all_model_presets() + .first() + .expect("test model presets should not be empty") + .clone(); + let (response_tx, response_rx) = tokio::sync::oneshot::channel(); + + message_tx.send(ThreadMessage::SetConfigOption { + config_id: SessionConfigId::new("model"), + value: SessionConfigOptionValue::ValueId { + value: SessionConfigValueId::new(preset.id.clone()), + }, + response_tx, + })?; + + response_rx.await??; + + let ops = thread.ops.lock().unwrap(); + assert!( + matches!(ops.last(), Some(Op::ThreadSettings { .. })), + "expected model config update to submit ThreadSettings, got {ops:?}" + ); + + drop(message_tx); + Ok(()) +} + +#[tokio::test] +async fn set_config_option_reasoning_effort_submits_thread_settings() -> anyhow::Result<()> { + let (_session_id, _client, thread, message_tx, _handle) = setup().await?; + let preset = all_model_presets() + .iter() + .find(|preset| preset.supported_reasoning_efforts.len() > 1) + .expect("at least one test model preset should expose reasoning effort") + .clone(); + let effort = preset.supported_reasoning_efforts[0].effort; + + let (model_response_tx, model_response_rx) = tokio::sync::oneshot::channel(); + message_tx.send(ThreadMessage::SetConfigOption { + config_id: SessionConfigId::new("model"), + value: SessionConfigOptionValue::ValueId { + value: SessionConfigValueId::new(preset.id.clone()), + }, + response_tx: model_response_tx, + })?; + model_response_rx.await??; + + let (effort_response_tx, effort_response_rx) = tokio::sync::oneshot::channel(); + message_tx.send(ThreadMessage::SetConfigOption { + config_id: SessionConfigId::new("reasoning_effort"), + value: SessionConfigOptionValue::ValueId { + value: SessionConfigValueId::new(effort.to_string()), + }, + response_tx: effort_response_tx, + })?; + + effort_response_rx.await??; + + let ops = thread.ops.lock().unwrap(); + assert!( + matches!(ops.last(), Some(Op::ThreadSettings { .. })), + "expected reasoning config update to submit ThreadSettings, got {ops:?}" + ); + + drop(message_tx); + Ok(()) +} + +#[tokio::test] +async fn set_config_option_rejects_unknown_option() -> anyhow::Result<()> { + let (_session_id, _client, _, message_tx, _handle) = setup().await?; + let (response_tx, response_rx) = tokio::sync::oneshot::channel(); + + message_tx.send(ThreadMessage::SetConfigOption { + config_id: SessionConfigId::new("unknown"), + value: SessionConfigOptionValue::ValueId { + value: SessionConfigValueId::new("value"), + }, + response_tx, + })?; + + assert!(response_rx.await?.is_err()); + + drop(message_tx); + Ok(()) +} + +#[tokio::test] +async fn cancel_submits_interrupt() -> anyhow::Result<()> { + let (_session_id, _client, thread, message_tx, _handle) = setup().await?; + let (response_tx, response_rx) = tokio::sync::oneshot::channel(); + + message_tx.send(ThreadMessage::Cancel { response_tx })?; + + response_rx.await??; + let ops = thread.ops.lock().unwrap(); + assert!( + matches!(ops.last(), Some(Op::Interrupt)), + "expected cancel to submit Interrupt, got {ops:?}" + ); + + drop(message_tx); + Ok(()) +} + +#[tokio::test] +async fn replay_history_sends_persisted_agent_updates() -> anyhow::Result<()> { + let (_session_id, client, _, message_tx, _handle) = setup().await?; + let thread_id = ThreadId::default(); + let history = vec![ + RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent { + message: "replayed answer".to_string(), + phase: None, + memory_citation: None, + })), + RolloutItem::EventMsg(EventMsg::AgentReasoning(AgentReasoningEvent { + text: "replayed reasoning".to_string(), + })), + RolloutItem::EventMsg(EventMsg::ThreadGoalUpdated(ThreadGoalUpdatedEvent { + thread_id, + turn_id: Some("turn-1".to_string()), + goal: ThreadGoal { + thread_id, + objective: "Replay the goal".to_string(), + status: ThreadGoalStatus::Complete, + token_budget: None, + tokens_used: 0, + time_used_seconds: 0, + created_at: 1, + updated_at: 2, + }, + })), + ]; + let (response_tx, response_rx) = tokio::sync::oneshot::channel(); + + message_tx.send(ThreadMessage::ReplayHistory { + history, + response_tx, + })?; + + response_rx.await??; + let notifications = client.notifications.lock().unwrap(); + assert!(notifications.iter().any(|notification| { + matches!( + ¬ification.update, + SessionUpdate::AgentMessageChunk(ContentChunk { + content: ContentBlock::Text(TextContent { text, .. }), + .. + }) if text == "replayed answer" + ) + })); + assert!(notifications.iter().any(|notification| { + matches!( + ¬ification.update, + SessionUpdate::AgentThoughtChunk(ContentChunk { + content: ContentBlock::Text(TextContent { text, .. }), + .. + }) if text == "replayed reasoning" + ) + })); + assert!(notifications.iter().any(|notification| { + matches!( + ¬ification.update, + SessionUpdate::AgentMessageChunk(ContentChunk { + content: ContentBlock::Text(TextContent { text, .. }), + .. + }) if text == "Goal updated (complete): Replay the goal" + ) + })); + + drop(notifications); + drop(message_tx); + Ok(()) +} + +#[tokio::test] +async fn logout_slash_command_logs_out_and_returns_auth_required() -> anyhow::Result<()> { + let session_id = SessionId::new("test"); + let client = Arc::new(StubClient::new()); + let session_client = SessionClient::with_client(session_id.clone(), client, Arc::default()); + let conversation = Arc::new(StubCodexThread::new()); + let models_manager = Arc::new(StubModelsManager); + let config = Config::load_with_cli_overrides_and_harness_overrides( + vec![], + ConfigOverrides::default(), + ) + .await?; + let logout_calls = Arc::new(AtomicUsize::new(0)); + let (message_tx, message_rx) = tokio::sync::mpsc::unbounded_channel(); + let (resolution_tx, resolution_rx) = tokio::sync::mpsc::unbounded_channel(); + let actor = ThreadActor::new( + RecordingAuth { + logout_calls: logout_calls.clone(), + }, + session_client, + conversation.clone(), + models_manager, + config, + message_rx, + resolution_tx, + resolution_rx, + ); + let _handle = tokio::spawn(actor.spawn()); + let (prompt_response_tx, prompt_response_rx) = tokio::sync::oneshot::channel(); + + message_tx.send(ThreadMessage::Prompt { + request: PromptRequest::new(session_id, vec!["/logout".into()]), + response_tx: prompt_response_tx, + })?; + + assert!(prompt_response_rx.await?.is_err()); + assert_eq!(logout_calls.load(Ordering::SeqCst), 1); + assert!( + conversation.ops.lock().unwrap().is_empty(), + "logout should not submit a Codex op" + ); + + drop(message_tx); + Ok(()) +} + +fn assert_config_option_ids_include(options: &[SessionConfigOption], expected: &[&str]) { + for expected_id in expected { + assert!( + options + .iter() + .any(|option| option.id.0.as_ref() == *expected_id), + "expected config option `{expected_id}` in {options:?}" + ); + } +} + +async fn wait_for_notification( + client: &StubClient, + mut matches_notification: impl FnMut(&SessionNotification) -> bool, +) -> anyhow::Result<()> { + tokio::time::timeout(Duration::from_millis(500), async { + loop { + if client + .notifications + .lock() + .unwrap() + .iter() + .any(&mut matches_notification) + { + return; + } + tokio::task::yield_now().await; + } + }) + .await?; + + Ok(()) +} + +struct RecordingAuth { + logout_calls: Arc, +} + +impl Auth for RecordingAuth { + async fn logout(&self) -> Result { + self.logout_calls.fetch_add(1, Ordering::SeqCst); + Ok(true) + } +}