fix(acp): Fixup WireApi::ACP basic loop

CSRessel · CSRessel · commit aa47c803e8fd · 2025-11-23T17:24:16.000-05:00
diff --git a/codex-rs/acp/src/agent.rs b/codex-rs/acp/src/agent.rs
@@ -138,7 +138,7 @@ impl AgentProcess {
         debug!("Initializing ACP agent");
 
         let request = InitializeRequest {
-            protocol_version: agent_client_protocol::V0,  // Gemini uses protocol version 0
+            protocol_version: agent_client_protocol::V0, // Gemini uses protocol version 0
             client_capabilities: serde_json::from_value(client_capabilities.clone())
                 .context("Invalid client capabilities")?,
             client_info: None,
@@ -147,7 +147,10 @@ impl AgentProcess {
 
         // Log the initialization request
         match serde_json::to_string_pretty(&request) {
-            Ok(json) => debug!("=== INITIALIZE REQUEST JSON ===\n{}\n=== END REQUEST ===", json),
+            Ok(json) => debug!(
+                "=== INITIALIZE REQUEST JSON ===\n{}\n=== END REQUEST ===",
+                json
+            ),
             Err(e) => debug!("Failed to serialize init request to JSON: {}", e),
         }
 
@@ -159,7 +162,10 @@ impl AgentProcess {
 
         // Log the full initialization response
         match serde_json::to_string_pretty(&response) {
-            Ok(json) => debug!("=== INITIALIZE RESPONSE JSON ===\n{}\n=== END RESPONSE ===", json),
+            Ok(json) => debug!(
+                "=== INITIALIZE RESPONSE JSON ===\n{}\n=== END RESPONSE ===",
+                json
+            ),
             Err(e) => debug!("Failed to serialize init response to JSON: {}", e),
         }
 
@@ -187,7 +193,10 @@ impl AgentProcess {
 
         // Serialize request to JSON for debugging
         match serde_json::to_string_pretty(&request) {
-            Ok(json) => debug!("=== NEW_SESSION REQUEST JSON ===\n{}\n=== END REQUEST ===", json),
+            Ok(json) => debug!(
+                "=== NEW_SESSION REQUEST JSON ===\n{}\n=== END REQUEST ===",
+                json
+            ),
             Err(e) => debug!("Failed to serialize request to JSON: {}", e),
         }
 
@@ -197,16 +206,22 @@ impl AgentProcess {
             error!("Protocol error creating session: {:?}", e);
 
             // Try to extract and log error details as JSON if available
-            if let Some(err_str) = format!("{:?}", e).split("data:").nth(1) {
-                error!("=== ERROR RESPONSE DETAILS ===\n{}\n=== END ERROR ===", err_str);
+            if let Some(err_str) = format!("{e:?}").split("data:").nth(1) {
+                error!(
+                    "=== ERROR RESPONSE DETAILS ===\n{}\n=== END ERROR ===",
+                    err_str
+                );
             }
 
             anyhow::anyhow!("Failed to create session: {e}")
         })?;
 
         // Log successful response as JSON
         match serde_json::to_string_pretty(&response) {
-            Ok(json) => debug!("=== NEW_SESSION RESPONSE JSON ===\n{}\n=== END RESPONSE ===", json),
+            Ok(json) => debug!(
+                "=== NEW_SESSION RESPONSE JSON ===\n{}\n=== END RESPONSE ===",
+                json
+            ),
             Err(e) => debug!("Failed to serialize response to JSON: {}", e),
         }
 
diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs
@@ -224,9 +224,10 @@ impl ModelClient {
                 debug!("Looking up ACP agent for model: {}", &self.config.model);
                 let agent_config = codex_acp::get_agent_config(&self.config.model)
                     .map_err(|e| CodexErr::Fatal(format!("ACP agent config error: {e}")))?;
-                debug!("Resolved ACP provider: {}, command: {}",
-                    agent_config.provider,
-                    agent_config.command);
+                debug!(
+                    "Resolved ACP provider: {}, command: {}",
+                    agent_config.provider, agent_config.command
+                );
 
                 // Create ACP model client
                 let acp_client = codex_acp::AcpModelClient::new(
@@ -251,12 +252,57 @@ impl ModelClient {
 
                 tokio::spawn(async move {
                     use futures::StreamExt;
+                    let mut created_sent = false;
+                    let mut assistant_item_sent = false;
+                    let mut reasoning_item_sent = false;
+
                     while let Some(acp_event_result) = acp_stream.next().await {
+                        // Send Created event at stream start
+                        if !created_sent {
+                            if tx.send(Ok(ResponseEvent::Created)).await.is_err() {
+                                break;
+                            }
+                            created_sent = true;
+                        }
+
                         let response_event = match acp_event_result {
                             Ok(codex_acp::AcpEvent::TextDelta(text)) => {
+                                // Send OutputItemAdded before first TextDelta
+                                if !assistant_item_sent {
+                                    let item = ResponseItem::Message {
+                                        id: None,
+                                        role: "assistant".to_string(),
+                                        content: vec![],
+                                    };
+                                    if tx
+                                        .send(Ok(ResponseEvent::OutputItemAdded(item)))
+                                        .await
+                                        .is_err()
+                                    {
+                                        break;
+                                    }
+                                    assistant_item_sent = true;
+                                }
                                 Ok(ResponseEvent::OutputTextDelta(text))
                             }
                             Ok(codex_acp::AcpEvent::ReasoningDelta(text)) => {
+                                // Send OutputItemAdded before first ReasoningDelta
+                                if !reasoning_item_sent {
+                                    let item = ResponseItem::Reasoning {
+                                        id: String::new(),
+                                        summary: Vec::new(),
+                                        content: Some(vec![]),
+                                        encrypted_content: None,
+                                    };
+                                    if tx
+                                        .send(Ok(ResponseEvent::OutputItemAdded(item)))
+                                        .await
+                                        .is_err()
+                                    {
+                                        break;
+                                    }
+                                    reasoning_item_sent = true;
+                                }
                                 Ok(ResponseEvent::ReasoningContentDelta {
                                     delta: text,
                                     content_index: 0,
diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs
@@ -2235,7 +2235,7 @@ async fn try_run_turn(
                     sess.send_event(&turn_context, EventMsg::AgentMessageContentDelta(event))
                         .await;
                 } else {
-                    error_or_panic("ReasoningSummaryDelta without active item".to_string());
+                    error_or_panic("OutputTextDelta without active item".to_string());
                 }
             }
             ResponseEvent::ReasoningSummaryDelta {
diff --git a/codex-rs/core/tests/acp_integration.rs b/codex-rs/core/tests/acp_integration.rs
@@ -121,3 +121,107 @@ async fn test_acp_stream_with_mock_agent() {
         .any(|e| matches!(e, ResponseEvent::Completed { .. }));
     assert!(completed, "Should receive Completed event");
 }
+
+#[tokio::test]
+async fn test_acp_event_ordering() {
+    // Create ACP provider for mock-acp-agent
+    let provider = ModelProviderInfo {
+        name: "mock-acp".into(),
+        base_url: None,
+        env_key: None,
+        env_key_instructions: None,
+        experimental_bearer_token: None,
+        wire_api: WireApi::Acp,
+        query_params: None,
+        http_headers: None,
+        env_http_headers: None,
+        request_max_retries: Some(0),
+        stream_max_retries: Some(0),
+        stream_idle_timeout_ms: Some(5_000),
+        requires_openai_auth: false,
+    };
+
+    // Load default config
+    let codex_home = TempDir::new().expect("Failed to create temp dir");
+    let mut config = load_default_config_for_test(&codex_home);
+    config.model = "mock-model".to_string();
+    config.model_provider_id = provider.name.clone();
+    config.model_provider = provider.clone();
+    let effort = config.model_reasoning_effort;
+    let summary = config.model_reasoning_summary;
+    let config = Arc::new(config);
+
+    let conversation_id = ConversationId::new();
+
+    let otel_event_manager = OtelEventManager::new(
+        conversation_id,
+        config.model.as_str(),
+        config.model_family.slug.as_str(),
+        None,
+        Some("test@test.com".to_string()),
+        Some(AuthMode::ChatGPT),
+        false,
+        "test".to_string(),
+    );
+
+    let client = ModelClient::new(
+        Arc::clone(&config),
+        None,
+        otel_event_manager,
+        provider,
+        effort,
+        summary,
+        conversation_id,
+        SessionSource::Exec,
+    );
+
+    let mut prompt = Prompt::default();
+    prompt.input = vec![ResponseItem::Message {
+        id: None,
+        role: "user".to_string(),
+        content: vec![ContentItem::InputText {
+            text: "Hello".to_string(),
+        }],
+    }];
+
+    // Stream response
+    let mut stream = client.stream(&prompt).await.expect("Stream should start");
+
+    // Collect events
+    let mut events = Vec::new();
+    while let Some(event_result) = stream.next().await {
+        let event = event_result.expect("Event should not be error");
+        events.push(event);
+    }
+
+    // Verify event ordering follows Created -> OutputItemAdded -> Deltas pattern
+    assert!(!events.is_empty(), "Should receive events from mock agent");
+
+    // First event should be Created
+    assert!(
+        matches!(events[0], ResponseEvent::Created),
+        "First event should be Created, got: {:?}",
+        events[0]
+    );
+
+    // Find first OutputItemAdded event
+    let output_item_added_index = events
+        .iter()
+        .position(|e| matches!(e, ResponseEvent::OutputItemAdded(_)))
+        .expect("Should have OutputItemAdded event");
+
+    // OutputItemAdded should come before any deltas
+    for (i, event) in events.iter().enumerate() {
+        match event {
+            ResponseEvent::OutputTextDelta(_) | ResponseEvent::ReasoningContentDelta { .. } => {
+                assert!(
+                    i > output_item_added_index,
+                    "Delta event at index {} should come after OutputItemAdded at index {}",
+                    i,
+                    output_item_added_index
+                );
+            }
+            _ => {}
+        }
+    }
+}
diff --git a/codex-rs/tui-integration-tests/src/lib.rs b/codex-rs/tui-integration-tests/src/lib.rs
@@ -56,7 +56,7 @@ impl Drop for TuiSession {
 }
 
 impl TuiSession {
-    /// Spawn codex with mock-acp-agent in a temporary directory
+    /// Spawn codex using mock-acp-agent binary in a temporary directory
     pub fn spawn(rows: u16, cols: u16) -> Result<Self> {
         let temp_dir = tempfile::tempdir()?;
         let hello_py = temp_dir.path().join("hello.py");
@@ -409,7 +409,7 @@ impl Default for SessionConfig {
 impl SessionConfig {
     pub fn new() -> Self {
         Self {
-            model: "mock-acp".to_string(),
+            model: "mock-model".to_string(),
             mock_agent_env: HashMap::new(),
             no_color: true,
             approval_policy: Some(ApprovalPolicy::OnFailure),
@@ -419,6 +419,11 @@ impl SessionConfig {
         }
     }
 
+    pub fn with_model(mut self, model: String) -> Self {
+        self.model = model;
+        self
+    }
+
     pub fn with_mock_response(mut self, response: impl Into<String>) -> Self {
         self.mock_agent_env
             .insert("MOCK_AGENT_RESPONSE".to_string(), response.into());
diff --git a/codex-rs/tui-integration-tests/tests/prompt_flow.rs b/codex-rs/tui-integration-tests/tests/prompt_flow.rs
@@ -43,6 +43,36 @@ fn test_submit_prompt_default_response() {
     assert_snapshot!("prompt_submitted", session.screen_contents());
 }
 
+#[test]
+fn test_submit_prompt_missing_model() {
+    let mut session = TuiSession::spawn_with_config(
+        24,
+        80,
+        SessionConfig::new().with_model("nonexistent".to_owned()),
+    )
+    .expect("Failed to spawn codex");
+
+    session.wait_for_text("? for shortcuts", TIMEOUT).unwrap();
+
+    // Type prompt
+    session.send_str("Hello").unwrap();
+    std::thread::sleep(Duration::from_millis(100));
+    session.wait_for_text("Hello", TIMEOUT).unwrap();
+
+    // Submit
+    session.send_key(Key::Enter).unwrap();
+    std::thread::sleep(Duration::from_millis(100));
+
+    session
+        .wait_for_text(
+            "ACP agent config error: Unknown ACP model: nonexistent-acp",
+            Duration::from_secs(10),
+        )
+        .unwrap();
+
+    assert_snapshot!("missing_model", session.screen_contents());
+}
+
 // #[test]
 // fn test_submit_prompt_custom_response() {
 //     let config = SessionConfig::new()

Original file line number	Diff line number	Diff line change
`@@ -2235,7 +2235,7 @@ async fn try_run_turn(`
`2235`	`2235`	`sess.send_event(&turn_context, EventMsg::AgentMessageContentDelta(event))`
`2236`	`2236`	`.await;`
`2237`	`2237`	`} else {`
`2238`		`- error_or_panic("ReasoningSummaryDelta without active item".to_string());`
	`2238`	`+ error_or_panic("OutputTextDelta without active item".to_string());`
`2239`	`2239`	`}`
`2240`	`2240`	`}`
`2241`	`2241`	`ResponseEvent::ReasoningSummaryDelta {`