feat: codex integration

crates/agentic-core/benches/executor_throughput.rs

@@ -136,7 +136,7 @@ fn make_request(input: &str, stream: bool, prev_id: Option<String>) -> RequestPa
         previous_response_id: prev_id,
         conversation_id: None,
         tools: None,
-        tool_choice: ToolChoice::Auto,
+        tool_choice: Some(ToolChoice::Auto),
         stream,
         store: true,
         include: None,

crates/agentic-core/src/config.rs

@@ -4,6 +4,7 @@ pub struct Config {
     pub openai_api_key: Option<String>,
     pub llm_ready_timeout_s: f64,
     pub llm_ready_interval_s: f64,
+    pub skip_llm_ready_check: bool,
     /// Database URL for conversation and response storage.
     /// `None` means stateful features are disabled; all requests are proxied.
     pub db_url: Option<String>,

crates/agentic-core/src/events/normalize.rs

@@ -129,6 +129,7 @@ fn extract_output_item_added(json: &Value) -> EventPayload {
         item_type: SSEItemType::from(json_str(item, "type")),
         output_index: json_u32(json, "output_index"),
         name: json_str_opt(item, "name"),
+        namespace: json_str_opt(item, "namespace"),
         call_id: json_str_opt(item, "call_id"),
     }
 }

crates/agentic-core/src/events/types.rs

@@ -114,6 +114,7 @@ pub enum EventPayload {
         item_type: SSEItemType,
         output_index: u32,
         name: Option<String>,
+        namespace: Option<String>,
         call_id: Option<String>,
     },
 

crates/agentic-core/src/executor/accumulator.rs

@@ -182,10 +182,7 @@ impl ResponseAccumulator {
         for line in rx {
             acc.process_sse_line(&line);
         }
-        acc.finalize_all();
-        if acc.status == ResponseStatus::InProgress {
-            acc.status = ResponseStatus::Completed;
-        }
+        acc.finish_stream();
         acc
     }
 
@@ -211,12 +208,19 @@ impl ResponseAccumulator {
         }
     }
 
-    fn process_sse_line(&mut self, line: &str) {
+    pub(crate) fn process_sse_line(&mut self, line: &str) {
         if let Some(frame) = normalize_sse_line(line) {
             self.process_event(&frame);
         }
     }
 
+    pub(crate) fn finish_stream(&mut self) {
+        self.finalize_all();
+        if self.status == ResponseStatus::InProgress {
+            self.status = ResponseStatus::Completed;
+        }
+    }
+
     /// Processes a typed [`EventFrame`], updating accumulator state.
     ///
     /// This is the core state machine — callers that already have a normalized
@@ -280,6 +284,16 @@ impl ResponseAccumulator {
                 self.status = ResponseStatus::Completed;
                 self.usage = *usage;
             }
+            (SSEEventType::ResponseFailed, EventPayload::Response { usage, .. }) => {
+                self.finalize_all();
+                self.status = ResponseStatus::Error;
+                self.usage = *usage;
+            }
+            (SSEEventType::ResponseIncomplete, EventPayload::Response { usage, .. }) => {
+                self.finalize_all();
+                self.status = ResponseStatus::Incomplete;
+                self.usage = *usage;
+            }
             _ => {}
         }
     }
@@ -434,6 +448,7 @@ mod tests {
                 item_type: "message".into(),
                 output_index: 0,
                 name: None,
+                namespace: None,
                 call_id: None,
             },
             sequence_number: Some(1),
@@ -502,6 +517,36 @@ mod tests {
         assert_eq!(acc.usage.unwrap().total_tokens, 15);
     }
 
+    #[test]
+    fn test_process_event_failed_sets_error_status() {
+        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
+        acc.process_event(&EventFrame {
+            event_type: SSEEventType::ResponseFailed,
+            payload: EventPayload::Response {
+                id: "resp_1".into(),
+                status: "failed".into(),
+                usage: None,
+            },
+            sequence_number: Some(4),
+        });
+        assert_eq!(acc.status, ResponseStatus::Error);
+    }
+
+    #[test]
+    fn test_process_event_incomplete_sets_incomplete_status() {
+        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
+        acc.process_event(&EventFrame {
+            event_type: SSEEventType::ResponseIncomplete,
+            payload: EventPayload::Response {
+                id: "resp_1".into(),
+                status: "incomplete".into(),
+                usage: None,
+            },
+            sequence_number: Some(4),
+        });
+        assert_eq!(acc.status, ResponseStatus::Incomplete);
+    }
+
     #[test]
     fn test_process_event_unknown_payload_ignored() {
         let mut acc = ResponseAccumulator::new("resp_1".into(), None);
@@ -625,6 +670,7 @@ mod tests {
                 item_type: "function_call".into(),
                 output_index: 0,
                 name: Some("get_weather".into()),
+                namespace: Some("mcp__weather".into()),
                 call_id: Some("call_abc".into()),
             },
             sequence_number: Some(1),
@@ -680,6 +726,7 @@ mod tests {
             assert_eq!(fc.id, "fc_1");
             assert_eq!(fc.call_id, "call_abc");
             assert_eq!(fc.name, "get_weather");
+            assert_eq!(fc.namespace.as_deref(), Some("mcp__weather"));
             assert_eq!(fc.arguments, r#"{"location":"Paris"}"#);
             assert_eq!(fc.status, MessageStatus::Completed);
         } else {
@@ -698,6 +745,7 @@ mod tests {
                 item_type: "function_call".into(),
                 output_index: 0,
                 name: Some("search".into()),
+                namespace: None,
                 call_id: Some("call_1".into()),
             },
             sequence_number: Some(1),
@@ -746,6 +794,7 @@ mod tests {
                 item_type: "function_call".into(),
                 output_index: 0,
                 name: Some("get_weather".into()),
+                namespace: None,
                 call_id: Some("call_1".into()),
             },
             sequence_number: Some(1),
@@ -769,6 +818,7 @@ mod tests {
                 item_type: "function_call".into(),
                 output_index: 1,
                 name: Some("get_time".into()),
+                namespace: None,
                 call_id: Some("call_2".into()),
             },
             sequence_number: Some(3),
@@ -811,6 +861,7 @@ mod tests {
                 item_type: "message".into(),
                 output_index: 0,
                 name: None,
+                namespace: None,
                 call_id: None,
             },
             sequence_number: Some(1),
@@ -833,6 +884,7 @@ mod tests {
                 item_type: "function_call".into(),
                 output_index: 1,
                 name: Some("lookup".into()),
+                namespace: None,
                 call_id: Some("call_x".into()),
             },
             sequence_number: Some(3),
@@ -875,6 +927,7 @@ mod tests {
                 item_type: "function_call".into(),
                 output_index: 0,
                 name: Some("old_name".into()),
+                namespace: None,
                 call_id: Some("old_call".into()),
             },
             sequence_number: Some(1),
@@ -912,6 +965,7 @@ mod tests {
                 item_type: "function_call".into(),
                 output_index: 0,
                 name: Some("tool".into()),
+                namespace: None,
                 call_id: Some("c1".into()),
             },
             sequence_number: Some(1),
@@ -968,6 +1022,7 @@ mod tests {
                 item_type: "function_call".into(),
                 output_index: 0,
                 name: Some("partial".into()),
+                namespace: None,
                 call_id: Some("c1".into()),
             },
             sequence_number: Some(1),

crates/agentic-core/src/executor/engine.rs

@@ -10,7 +10,7 @@ use std::sync::Arc;
 use async_stream::stream;
 use either::Either;
 use tokio::sync::mpsc;
-use tracing::warn;
+use tracing::{debug, warn};
 
 use super::gateway::{
     append_gateway_calls_to_new_input, append_output_items_to_input, append_tool_outputs,
@@ -58,7 +58,12 @@ fn accumulate_usage(total: &mut Option<ResponseUsage>, usage: Option<ResponseUsa
 }
 
 fn error_sse_chunk(message: &str) -> String {
-    let event = serde_json::json!({ "error": message });
+    let event = serde_json::json!({
+        "type": "error",
+        "error": {
+            "message": message,
+        },
+    });
     let event_json = serialize_to_string(&event).unwrap_or_else(|_| "{\"error\":\"stream error\"}".to_owned());
     format!("data: {event_json}\n\n")
 }
@@ -102,22 +107,23 @@ async fn run_until_gateway_tools_complete(
     stream_upstream: bool,
     stream_events: Option<&mpsc::UnboundedSender<String>>,
 ) -> ExecutorResult<(ResponsePayload, RequestContext)> {
-    let registry = ctx
+    let registry: ToolRegistry = ctx
         .enriched_request
         .tools
         .as_ref()
         .map_or_else(ToolRegistry::default, |tools| {
             ToolRegistry::build_with_handlers(tools, |tool_type| exec_ctx.gateway_executors.get(tool_type))
         });
-    let mut combined_output = Vec::new();
-    let mut combined_usage = None;
+    let mut combined_output: Vec<crate::OutputItem> = Vec::new();
+    let mut combined_usage: Option<ResponseUsage> = None;
 
     for _ in 0..MAX_GATEWAY_TOOL_ROUNDS {
-        let mut payload = if stream_upstream {
-            fetch_stream_payload(&ctx, exec_ctx, auth).await?
+        let mut payload: ResponsePayload = if stream_upstream {
+            fetch_stream_payload(&ctx, exec_ctx, auth, &registry, stream_events).await?
         } else {
             fetch_blocking_payload(&ctx, exec_ctx, auth).await?
         };
+        registry.restore_final_payload_output(&mut payload.output);
         accumulate_usage(&mut combined_usage, payload.usage);
         let current_output = std::mem::take(&mut payload.output);
         let has_client_owned_calls = has_client_owned_calls(&current_output, &registry);
@@ -147,7 +153,7 @@ async fn run_until_gateway_tools_complete(
         }
 
         combined_output.extend(public_output);
-        ctx.enriched_request.tool_choice = ToolChoice::Auto;
+        ctx.enriched_request.tool_choice = Some(ToolChoice::Auto);
         append_output_items_to_input(&mut ctx.enriched_request.input, &current_output);
         append_gateway_calls_to_new_input(&mut ctx, &current_output, &registry);
         append_tool_outputs(
@@ -211,7 +217,7 @@ fn run_stream(ctx: RequestContext, exec_ctx: Arc<ExecutionContext>, auth: Option
                             yield DONE_MARKER.to_string();
                         }
                         Ok(Ok((payload, ctx))) => {
-                            yield payload.as_responses_chunk();
+                            yield payload.as_terminal_response_chunk();
                             yield DONE_MARKER.to_string();
 
                             let ch = exec_ctx.conv_handler.clone();
@@ -277,6 +283,15 @@ impl ExecuteRequest {
     /// # Errors
     /// Returns [`ExecutorError`] if rehydration or (non-streaming) LLM inference fails.
     pub async fn run(self) -> ExecutorResult<Either<ResponsePayload, BoxStream>> {
+        debug!(
+            model = %self.payload.model,
+            store = self.payload.store,
+            stream = self.payload.stream,
+            has_previous_response_id = self.payload.previous_response_id.is_some(),
+            has_conversation_id = self.payload.conversation_id.is_some(),
+            tools = self.payload.tools.as_ref().map_or(0, Vec::len),
+            "executor received responses request"
+        );
         let ctx = rehydrate_conversation(self.payload, &self.exec_ctx).await?;
         if ctx.original_request.stream {
             Ok(Either::Right(run_stream(ctx, self.exec_ctx, self.client_auth)))

crates/agentic-core/src/executor/gateway.rs

@@ -38,7 +38,7 @@ fn function_calls(output_items: &[OutputItem]) -> Vec<FunctionToolCall> {
 fn is_gateway_owned_call(call: &FunctionToolCall, registry: &ToolRegistry) -> bool {
     registry
         .lookup(&call.name)
-        .is_some_and(|entry| entry.tool_type != ToolType::Function)
+        .is_some_and(|entry| entry.tool_type.is_gateway_owned())
 }
 
 pub(super) fn has_client_owned_calls(output_items: &[OutputItem], registry: &ToolRegistry) -> bool {
@@ -83,7 +83,11 @@ fn gateway_public_output(
 ) -> Option<OutputItem> {
     match tool_type {
         ToolType::WebSearch => Some(crate::tool::web_search::output_item(call, output, status)),
-        ToolType::Function | ToolType::Mcp | ToolType::FileSearch | ToolType::CodeInterpreter => None,
+        ToolType::Function
+        | ToolType::CodexNamespace
+        | ToolType::Mcp
+        | ToolType::FileSearch
+        | ToolType::CodeInterpreter => None,
     }
 }
 
@@ -141,14 +145,18 @@ fn gateway_event_plans(
     for item in output_items {
         if let OutputItem::FunctionCall(call) = item
             && let Some(entry) = registry.lookup(&call.name)
-            && entry.tool_type != ToolType::Function
+            && entry.tool_type.is_gateway_owned()
         {
             plans.push(GatewayCallEventPlan {
                 call_id: call.call_id.clone(),
                 output_index: u32::try_from(output_index).unwrap_or(u32::MAX),
                 started_output: match entry.tool_type {
                     ToolType::WebSearch => Some(crate::tool::web_search::started_output_item(call)),
-                    ToolType::Function | ToolType::Mcp | ToolType::FileSearch | ToolType::CodeInterpreter => None,
+                    ToolType::Function
+                    | ToolType::CodexNamespace
+                    | ToolType::Mcp
+                    | ToolType::FileSearch
+                    | ToolType::CodeInterpreter => None,
                 },
             });
         }