agent_engine/runtime/openai/
mod.rs

1//! OpenAI-compatible provider engine.
2//!
3//! Ported from the `openai-runtime` prototype. Translates between Anthropic-shaped
4//! messages/tools/content-blocks (the internal synaps representation) and
5//! OpenAI `chat/completions` SSE wire.
6
7pub mod catalog;
8pub(crate) mod extension_route;
9pub mod net;
10pub mod ping;
11pub mod reasoning;
12pub mod registry;
13pub mod stream;
14pub mod translate;
15pub mod types;
16pub mod wire;
17
18use std::sync::Arc;
19
20use crate::extensions::manager::ExtensionManager;
21use crate::extensions::providers::ProviderRegistry;
22
23pub use types::{
24    ChatMessage, ChatOptions, ChatRequest, FunctionCall, FunctionDefinition, OaiEvent,
25    ProviderConfig, StreamOptions, ToolCall, ToolChoice, ToolDefinition,
26};
27pub use wire::StreamDecoder;
28
29static EXTENSION_MANAGER: std::sync::RwLock<Option<Arc<tokio::sync::RwLock<ExtensionManager>>>> =
30    std::sync::RwLock::new(None);
31
32pub fn set_extension_manager_for_routing(manager: Arc<tokio::sync::RwLock<ExtensionManager>>) {
33    *EXTENSION_MANAGER
34        .write()
35        .expect("extension manager routing lock poisoned") = Some(manager);
36}
37
38pub fn clear_extension_manager_for_routing() {
39    *EXTENSION_MANAGER
40        .write()
41        .expect("extension manager routing lock poisoned") = None;
42}
43
44pub fn extension_manager_for_routing() -> Option<Arc<tokio::sync::RwLock<ExtensionManager>>> {
45    EXTENSION_MANAGER
46        .read()
47        .expect("extension manager routing lock poisoned")
48        .clone()
49}
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq)]
52pub enum AuthPolicy {
53    OAuthAccessToken(crate::auth::OAuthProviderId),
54    BrokerProxy,
55}
56#[derive(Debug, Clone, Copy, PartialEq, Eq)]
57pub enum WireProtocol {
58    AnthropicMessages,
59    OpenAiChatCompletions,
60    OpenAiResponses,
61    CodexResponses,
62    /// Google Gemini Code Assist v1internal:streamGenerateContent envelope.
63    /// Broker-proxied only.
64    GoogleGeminiCodeAssist,
65}
66#[derive(Debug, Clone, PartialEq, Eq)]
67pub struct ResolvedRoute {
68    pub endpoint: String,
69    pub model: String,
70    pub provider: String,
71    pub auth: AuthPolicy,
72    pub wire: WireProtocol,
73}
74
75/// Resolve typed credential and wire dimensions. Unknown explicit prefixes fail closed.
76pub fn resolve_route(model: &str) -> Option<ResolvedRoute> {
77    if let Some((prefix, rest)) = model.split_once('/') {
78        if prefix == "anthropic" && rest.starts_with("claude-") {
79            return Some(ResolvedRoute {
80                endpoint: "https://api.anthropic.com".into(),
81                model: rest.into(),
82                provider: prefix.into(),
83                auth: AuthPolicy::OAuthAccessToken(crate::auth::OAuthProviderId::Anthropic),
84                wire: WireProtocol::AnthropicMessages,
85            });
86        }
87        if prefix == "xai-auth" && catalog::xai_model(rest).is_some() {
88            return Some(ResolvedRoute {
89                endpoint: "https://api.x.ai/v1".into(),
90                model: rest.into(),
91                provider: prefix.into(),
92                auth: AuthPolicy::OAuthAccessToken(crate::auth::OAuthProviderId::Xai),
93                wire: WireProtocol::OpenAiResponses,
94            });
95        }
96        if prefix == "github-copilot" {
97            let descriptor = catalog::github_copilot_runtime_model(rest)?;
98            return Some(ResolvedRoute {
99                endpoint: catalog::MODELS_BASE_URL.into(),
100                model: rest.into(),
101                provider: prefix.into(),
102                auth: AuthPolicy::BrokerProxy,
103                wire: descriptor,
104            });
105        }
106        if prefix == "google-gemini" {
107            // Broker-proxied only: broker owns the Google OAuth access token
108            // and pins cloudcode-pa.googleapis.com. Wire IDs are limited to
109            // the conservative catalog.
110            catalog::google_gemini_model(rest)?;
111            return Some(ResolvedRoute {
112                endpoint: crate::auth::broker::GOOGLE_GEMINI_CODE_ASSIST_BASE_URL.into(),
113                model: rest.into(),
114                provider: prefix.into(),
115                auth: AuthPolicy::BrokerProxy,
116                wire: WireProtocol::GoogleGeminiCodeAssist,
117            });
118        }
119        if prefix == "openai-codex" {
120            let c = registry::resolve_codex_shorthand(model)?;
121            return Some(ResolvedRoute {
122                endpoint: c.base_url,
123                model: c.model,
124                provider: c.provider,
125                auth: AuthPolicy::OAuthAccessToken(crate::auth::OAuthProviderId::OpenAiCodex),
126                wire: WireProtocol::CodexResponses,
127            });
128        }
129        if prefix == "kimi-code" {
130            // Broker-proxied managed Kimi Code endpoint: the broker owns the
131            // short-lived OAuth access token (and its rotating refresh token)
132            // and pins api.kimi.com/coding/v1. Wire ids are limited to the
133            // conservative managed catalog — unknown ids fail closed.
134            catalog::kimi_code_model(rest)?;
135            return Some(ResolvedRoute {
136                endpoint: crate::auth::kimi_code::API_BASE_URL.into(),
137                model: rest.into(),
138                provider: prefix.into(),
139                auth: AuthPolicy::BrokerProxy,
140                wire: WireProtocol::OpenAiChatCompletions,
141            });
142        }
143        if prefix == "local" {
144            let c = registry::resolve_provider_model(prefix, rest)?;
145            return Some(ResolvedRoute {
146                endpoint: c.base_url,
147                model: c.model,
148                provider: c.provider,
149                auth: AuthPolicy::BrokerProxy,
150                wire: WireProtocol::OpenAiChatCompletions,
151            });
152        }
153        if let Some(spec) = registry::providers().iter().find(|s| s.key == prefix) {
154            return Some(ResolvedRoute {
155                endpoint: spec.base_url.into(),
156                model: rest.into(),
157                provider: prefix.into(),
158                auth: AuthPolicy::BrokerProxy,
159                wire: WireProtocol::OpenAiChatCompletions,
160            });
161        }
162        return None;
163    }
164    Some(ResolvedRoute {
165        endpoint: "https://api.anthropic.com".into(),
166        model: model.into(),
167        provider: "anthropic".into(),
168        auth: AuthPolicy::OAuthAccessToken(crate::auth::OAuthProviderId::Anthropic),
169        wire: WireProtocol::AnthropicMessages,
170    })
171}
172fn provider_config(r: &ResolvedRoute) -> ProviderConfig {
173    ProviderConfig {
174        base_url: r.endpoint.clone(),
175        model: r.model.clone(),
176        provider: r.provider.clone(),
177    }
178}
179
180/// Try to route a request through an OpenAI-compatible provider.
181///
182/// Returns `Some(Ok(value))` if the model resolved to an OpenAI provider and the
183/// request completed (successfully or with error). Returns `None` if the model
184/// should be handled by the Anthropic path.
185///
186/// This is the single routing entry point — both streaming and non-streaming
187/// callers in `api.rs` use this instead of duplicating the routing logic.
188/// `tool_session_id` scopes the execution gate for extension-provider
189/// interior tool loops (Task 16); `None` fails closed inside the route.
190#[allow(clippy::too_many_arguments)]
191pub async fn try_route(
192    model: &str,
193    client: &reqwest::Client,
194    tools_schema: &std::sync::Arc<Vec<serde_json::Value>>,
195    system_prompt: &Option<String>,
196    messages: &[crate::SharedMessage],
197    tx: &tokio::sync::mpsc::UnboundedSender<crate::runtime::types::StreamEvent>,
198    temperature: Option<f32>,
199    max_tokens: Option<u32>,
200    thinking_budget: u32,
201    reasoning_level: agent_core::reasoning::ReasoningLevel,
202    cancel: &tokio_util::sync::CancellationToken,
203    source: &crate::auth::CredentialSource,
204    cache: &crate::auth::TokenCache,
205    max_retries: u32,
206    codex_request_role: catalog::CodexRequestRole,
207    tool_session_id: Option<&crate::tools::activation::SessionId>,
208    session_tool_set: Option<&crate::tools::activation::SharedSessionToolSet>,
209    trace: &crate::runtime::trace::TraceContext,
210    suppress_stream_deltas: bool,
211) -> Option<Result<serde_json::Value, Box<dyn std::error::Error + Send + Sync>>> {
212    try_route_with_memory_backend(
213        model,
214        client,
215        tools_schema,
216        system_prompt,
217        messages,
218        tx,
219        temperature,
220        max_tokens,
221        thinking_budget,
222        reasoning_level,
223        cancel,
224        source,
225        cache,
226        max_retries,
227        codex_request_role,
228        tool_session_id,
229        session_tool_set,
230        None,
231        trace,
232        suppress_stream_deltas,
233    )
234    .await
235}
236
237/// Runtime route retaining the host's immutable memory binding. The public
238/// compatibility facade above has no runtime capability and cannot mint one.
239#[allow(clippy::too_many_arguments)]
240pub(crate) async fn try_route_with_memory_backend(
241    model: &str,
242    client: &reqwest::Client,
243    tools_schema: &std::sync::Arc<Vec<serde_json::Value>>,
244    system_prompt: &Option<String>,
245    messages: &[crate::SharedMessage],
246    tx: &tokio::sync::mpsc::UnboundedSender<crate::runtime::types::StreamEvent>,
247    temperature: Option<f32>,
248    max_tokens: Option<u32>,
249    thinking_budget: u32,
250    reasoning_level: agent_core::reasoning::ReasoningLevel,
251    cancel: &tokio_util::sync::CancellationToken,
252    source: &crate::auth::CredentialSource,
253    cache: &crate::auth::TokenCache,
254    max_retries: u32,
255    codex_request_role: catalog::CodexRequestRole,
256    tool_session_id: Option<&crate::tools::activation::SessionId>,
257    session_tool_set: Option<&crate::tools::activation::SharedSessionToolSet>,
258    memory_backend: Option<&crate::memory_backend::MemoryBinding>,
259    trace: &crate::runtime::trace::TraceContext,
260    suppress_stream_deltas: bool,
261) -> Option<Result<serde_json::Value, Box<dyn std::error::Error + Send + Sync>>> {
262    if let Err(error) = super::attachments::validate_messages(model, messages) {
263        return Some(Err(error.into()));
264    }
265    if let Some((plugin_id, provider_id, model_id)) = ProviderRegistry::parse_model_id(model) {
266        if let Some(manager) = extension_manager_for_routing() {
267            // Extension-hosted provider: routing gates, audits, and the
268            // Task 10C transport trace live in `extension_route`.
269            return Some(
270                extension_route::route_extension_provider(
271                    manager,
272                    plugin_id,
273                    provider_id,
274                    model_id,
275                    model,
276                    tools_schema,
277                    system_prompt,
278                    messages,
279                    tx,
280                    temperature,
281                    max_tokens,
282                    thinking_budget,
283                    cancel,
284                    tool_session_id,
285                    session_tool_set,
286                    memory_backend,
287                    trace,
288                )
289                .await,
290            );
291        }
292        return Some(Err(format!(
293            "Extension provider model '{}' is not available",
294            model
295        )
296        .into()));
297    }
298
299    let Some(route) = resolve_route(model) else {
300        return Some(Err(
301            format!("Unknown provider route for model '{model}'").into()
302        ));
303    };
304    let cfg = provider_config(&route);
305    let broker = crate::auth::broker_from_source(source, cache, client.clone());
306    // Trace honesty (Task 10A): only a local (in-process) broker sends the
307    // exact bytes this process serialized (`ProxyRequest::body_bytes`); a
308    // remote broker daemon re-serializes upstream bodies out of process.
309    let exact_wire_bytes = !source.is_remote();
310    match route.wire {
311        WireProtocol::OpenAiChatCompletions => Some(
312            stream::call_oai_stream_inner(
313                &cfg,
314                &broker,
315                tools_schema,
316                system_prompt,
317                messages,
318                tx,
319                temperature,
320                max_tokens,
321                thinking_budget,
322                reasoning_level,
323                cancel,
324                trace,
325                exact_wire_bytes,
326                suppress_stream_deltas,
327                max_retries,
328            )
329            .await,
330        ),
331        WireProtocol::OpenAiResponses => {
332            if suppress_stream_deltas {
333                return Some(Err(
334                    "internal sync routing is unavailable for this streaming provider".into(),
335                ));
336            }
337            Some(
338                stream::call_xai_responses_stream_inner(
339                    &cfg,
340                    &broker,
341                    tools_schema,
342                    system_prompt,
343                    messages,
344                    tx,
345                    max_tokens,
346                    reasoning_level,
347                    cancel,
348                    // In-stream capacity errors (HTTP 200 + lone `error`
349                    // event) are transient; re-send identical bytes with
350                    // the configured retry budget.
351                    max_retries,
352                    trace,
353                    exact_wire_bytes,
354                )
355                .await,
356            )
357        }
358        WireProtocol::CodexResponses => {
359            if suppress_stream_deltas {
360                return Some(Err(
361                    "internal sync routing is unavailable for this streaming provider".into(),
362                ));
363            }
364            Some(
365                stream::call_codex_stream_inner(
366                    &cfg,
367                    client,
368                    &broker,
369                    tools_schema,
370                    system_prompt,
371                    messages,
372                    tx,
373                    temperature,
374                    max_tokens,
375                    reasoning_level,
376                    codex_request_role,
377                    cancel,
378                    // Codex transport rides out chatgpt.com edge bursts with the
379                    // same persistent posture as Anthropic OAuth overloads (10
380                    // retries), not the generic three-attempt budget.
381                    stream::codex_retry_budget(max_retries),
382                    trace,
383                )
384                .await,
385            )
386        }
387        WireProtocol::AnthropicMessages => None,
388        WireProtocol::GoogleGeminiCodeAssist => {
389            if suppress_stream_deltas {
390                return Some(Err(
391                    "internal sync routing is unavailable for this streaming provider".into(),
392                ));
393            }
394            Some(
395                crate::runtime::google_gemini::runtime::call_google_gemini_stream_inner(
396                    &cfg,
397                    &broker,
398                    tools_schema,
399                    system_prompt,
400                    messages,
401                    tx,
402                    cancel,
403                    trace,
404                    exact_wire_bytes,
405                )
406                .await,
407            )
408        }
409    }
410}
411
412#[cfg(test)]
413mod tests {
414    use super::*;
415
416    #[test]
417    fn anthropic_identity_is_explicit_while_legacy_bare_ids_still_route() {
418        for id in ["claude-sonnet-4-6", "claude-opus-4-7"] {
419            let qualified = resolve_route(&format!("anthropic/{id}"))
420                .expect("qualified Anthropic model must route");
421            assert_eq!(qualified.provider, "anthropic");
422            assert_eq!(qualified.model, id);
423            assert_eq!(qualified.wire, WireProtocol::AnthropicMessages);
424
425            let legacy = resolve_route(id).expect("legacy bare Claude model must route");
426            assert_eq!(legacy.provider, "anthropic");
427            assert_eq!(legacy.model, id);
428        }
429        assert_eq!(
430            resolve_route("github-copilot/claude-sonnet-4.6")
431                .unwrap()
432                .provider,
433            "github-copilot"
434        );
435    }
436
437    #[test]
438    fn brokered_provider_routing_is_typed_and_fail_closed() {
439        for descriptor in catalog::XAI_TEXT_MODELS {
440            let runtime_id = format!("xai-auth/{}", descriptor.id);
441            let xai = resolve_route(&runtime_id).unwrap();
442            assert_eq!(xai.model, descriptor.id);
443            assert_eq!(xai.provider, "xai-auth");
444            assert_eq!(xai.endpoint, "https://api.x.ai/v1");
445            assert_eq!(
446                xai.auth,
447                AuthPolicy::OAuthAccessToken(crate::auth::OAuthProviderId::Xai)
448            );
449            assert_eq!(xai.wire, WireProtocol::OpenAiResponses);
450        }
451        let static_route = resolve_route("groq/llama-3.3-70b-versatile").unwrap();
452        assert_eq!(static_route.auth, AuthPolicy::BrokerProxy);
453        assert_eq!(static_route.wire, WireProtocol::OpenAiChatCompletions);
454
455        let copilot_chat = resolve_route("github-copilot/claude-sonnet-4.6").unwrap();
456        assert_eq!(copilot_chat.auth, AuthPolicy::BrokerProxy);
457        assert_eq!(copilot_chat.wire, WireProtocol::OpenAiChatCompletions);
458        let copilot_responses = resolve_route("github-copilot/gpt-5.3-codex").unwrap();
459        assert_eq!(copilot_responses.auth, AuthPolicy::BrokerProxy);
460        assert_eq!(copilot_responses.wire, WireProtocol::OpenAiResponses);
461        assert!(resolve_route("github-copilot/unverified-model").is_none());
462        assert!(resolve_route("unknown-provider/model").is_none());
463        assert!(resolve_route("xai-auth/not-a-real-model").is_none());
464        assert!(resolve_route("xai-auth/grok-build-0.1").is_none());
465        assert!(resolve_route("xai-auth/grok-build-latest").is_none());
466        assert!(resolve_route("xai-auth/grok-imagine-image").is_none());
467
468        // Every verified concrete text/tool wire ID in the catalog must resolve.
469        for descriptor in catalog::GOOGLE_GEMINI_TEXT_MODELS {
470            let route = resolve_route(&format!("google-gemini/{}", descriptor.id)).unwrap();
471            assert_eq!(route.provider, "google-gemini");
472            assert_eq!(route.model, descriptor.id);
473            assert_eq!(route.endpoint, "https://cloudcode-pa.googleapis.com");
474            assert_eq!(route.auth, AuthPolicy::BrokerProxy);
475            assert_eq!(route.wire, WireProtocol::GoogleGeminiCodeAssist);
476        }
477        // Aliases, unsupported media/embedding models, and family prefixes fail closed.
478        for banned in [
479            "auto-gemini-2.5",
480            "auto-gemini-3",
481            "text-embedding-004",
482            "gemini-2.5-image",
483            "gemini-2.5", // family prefix is not a wire id
484        ] {
485            assert!(
486                resolve_route(&format!("google-gemini/{banned}")).is_none(),
487                "{banned} must not resolve"
488            );
489        }
490    }
491
492    /// Regression: `gemini-pro-latest` must NOT resolve — it is a public
493    /// Gemini API alias that Code Assist rejects with 404 NOT_FOUND. Routing
494    /// fails closed instead of emitting a doomed upstream request.
495    #[test]
496    fn does_not_resolve_public_api_alias_gemini_pro_latest() {
497        assert!(
498            resolve_route("google-gemini/gemini-pro-latest").is_none(),
499            "gemini-pro-latest is not a Code Assist wire ID (404 upstream)"
500        );
501    }
502
503    #[test]
504    fn kimi_code_routing_is_broker_proxied_and_fail_closed() {
505        for descriptor in catalog::KIMI_CODE_TEXT_MODELS {
506            let route = resolve_route(&format!("kimi-code/{}", descriptor.id)).unwrap();
507            assert_eq!(route.provider, "kimi-code");
508            assert_eq!(route.model, descriptor.id);
509            assert_eq!(route.endpoint, "https://api.kimi.com/coding/v1");
510            // Broker-proxied: the short-lived OAuth bearer never reaches the
511            // runtime; the broker resolves + attaches it per request.
512            assert_eq!(route.auth, AuthPolicy::BrokerProxy);
513            assert_eq!(route.wire, WireProtocol::OpenAiChatCompletions);
514        }
515        // Unknown ids and the static Moonshot-platform ids fail closed on
516        // this route (they belong to the API-key `kimi` provider).
517        assert!(resolve_route("kimi-code/unknown-model").is_none());
518        assert!(resolve_route("kimi-code/kimi-k3").is_none());
519        assert!(resolve_route("kimi-code/kimi-k2.7-code").is_none());
520    }
521
522    #[test]
523    fn resolves_openai_codex_without_requiring_eager_credentials() {
524        std::env::remove_var("OPENAI_CODEX_ACCESS_TOKEN");
525        let route = resolve_route("openai-codex/gpt-5.1-codex-mini").unwrap();
526        assert_eq!(route.provider, "openai-codex");
527        assert_eq!(route.wire, WireProtocol::CodexResponses);
528    }
529
530    /// Regression: models under the `google-gemini/*` prefix must route through
531    /// `try_route` (i.e. the OpenAI-runtime dispatcher must return `Some(_)` so
532    /// the caller invokes the Gemini stream rather than falling through to the
533    /// Anthropic path and reporting a misleading 401. Before this fix,
534    /// `WireProtocol::GoogleGeminiCodeAssist => None` made `try_route` return
535    /// `None`.
536    ///
537    /// We drive `try_route` through the `Remote` credential source pointed at a
538    /// closed loopback port so the broker request errors out fast without
539    /// touching the developer's real auth.json or the network. The outcome we
540    /// assert is only that the routing dispatcher returned `Some(_)` — the
541    /// specific broker error is irrelevant.
542    #[tokio::test]
543    async fn try_route_google_gemini_is_not_none_and_does_not_fall_through_to_anthropic() {
544        let client = reqwest::Client::builder()
545            .connect_timeout(std::time::Duration::from_secs(2))
546            .build()
547            .expect("client");
548        // 127.0.0.1:1 is a reserved port; on Linux connect(2) returns
549        // ECONNREFUSED immediately so the broker call fails fast.
550        let source = crate::auth::CredentialSource::Remote {
551            endpoint: "http://127.0.0.1:1".into(),
552            machine_token: "not-a-real-token".into(),
553        };
554        let cache = crate::auth::TokenCache::new();
555        let tools_schema: std::sync::Arc<Vec<serde_json::Value>> = std::sync::Arc::new(Vec::new());
556        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
557        let cancel = tokio_util::sync::CancellationToken::new();
558
559        let messages: Vec<crate::SharedMessage> = vec![std::sync::Arc::new(serde_json::json!({
560            "role": "user",
561            "content": "hi"
562        }))];
563
564        let trace = crate::runtime::trace::TraceContext::disabled();
565        let fut = try_route(
566            "google-gemini/gemini-2.5-pro",
567            &client,
568            &tools_schema,
569            &None,
570            &messages,
571            &tx,
572            None,
573            None,
574            0,
575            agent_core::reasoning::ReasoningLevel::Adaptive,
576            &cancel,
577            &source,
578            &cache,
579            0,
580            catalog::CodexRequestRole::Foreground,
581            None,
582            None,
583            &trace,
584            false,
585        );
586        let result = tokio::time::timeout(std::time::Duration::from_secs(10), fut)
587            .await
588            .expect("try_route completed within budget");
589
590        assert!(
591            result.is_some(),
592            "google-gemini must route through try_route (was None → fell through to Anthropic)"
593        );
594    }
595
596    /// Headless harness (no live credentials): rejected xAI reasoning
597    /// combinations must fail BEFORE any credential or network access on the
598    /// Responses path. The Remote credential source points at a closed
599    /// loopback port — if validation ran after broker access this would
600    /// surface a connection error instead of the capability message.
601    #[tokio::test]
602    async fn xai_off_is_rejected_pre_network_through_try_route() {
603        let client = reqwest::Client::builder()
604            .connect_timeout(std::time::Duration::from_secs(2))
605            .build()
606            .expect("client");
607        let source = crate::auth::CredentialSource::Remote {
608            endpoint: "http://127.0.0.1:1".into(),
609            machine_token: "not-a-real-token".into(),
610        };
611        let cache = crate::auth::TokenCache::new();
612        let tools_schema: std::sync::Arc<Vec<serde_json::Value>> = std::sync::Arc::new(Vec::new());
613        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
614        let cancel = tokio_util::sync::CancellationToken::new();
615        let messages: Vec<crate::SharedMessage> = vec![std::sync::Arc::new(serde_json::json!({
616            "role": "user",
617            "content": "hi"
618        }))];
619
620        for (level, needle) in [
621            (
622                agent_core::reasoning::ReasoningLevel::Off,
623                "cannot be disabled",
624            ),
625            (
626                agent_core::reasoning::ReasoningLevel::XHigh,
627                "not supported",
628            ),
629            (
630                agent_core::reasoning::ReasoningLevel::Ultra,
631                "not supported",
632            ),
633        ] {
634            let result = try_route(
635                "xai-auth/grok-4.5",
636                &client,
637                &tools_schema,
638                &None,
639                &messages,
640                &tx,
641                None,
642                None,
643                0,
644                level,
645                &cancel,
646                &source,
647                &cache,
648                0,
649                catalog::CodexRequestRole::Foreground,
650                None,
651                None,
652                &crate::runtime::trace::TraceContext::disabled(),
653                false,
654            )
655            .await
656            .expect("xai-auth must route through try_route");
657            let err = result.expect_err("rejected level must error");
658            assert!(
659                err.to_string().contains(needle),
660                "{level}: pre-network capability rejection expected, got: {err}"
661            );
662        }
663    }
664
665    #[test]
666    fn set_extension_manager_for_routing_overwrites_previous_manager() {
667        clear_extension_manager_for_routing();
668        let first = Arc::new(tokio::sync::RwLock::new(ExtensionManager::new(Arc::new(
669            crate::extensions::hooks::HookBus::new(),
670        ))));
671        let second = Arc::new(tokio::sync::RwLock::new(ExtensionManager::new(Arc::new(
672            crate::extensions::hooks::HookBus::new(),
673        ))));
674
675        set_extension_manager_for_routing(first.clone());
676        assert!(Arc::ptr_eq(
677            &extension_manager_for_routing().unwrap(),
678            &first
679        ));
680
681        set_extension_manager_for_routing(second.clone());
682        assert!(Arc::ptr_eq(
683            &extension_manager_for_routing().unwrap(),
684            &second
685        ));
686
687        clear_extension_manager_for_routing();
688        assert!(extension_manager_for_routing().is_none());
689    }
690}
agent_engine/runtime/openai/mod.rs

agent_engine/runtime/openai/
mod.rs
