agentic_core/executor/
request.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use crate::config::Config;
5use crate::error::Error;
6use crate::executor::modes::{ConversationHandler, ResponseHandler};
7use crate::storage::{ConversationStore, ResponseStore, create_pool_with_schema};
8use crate::tool::{GatewayExecutor, ToolType, WebSearchHandler};
9use crate::types::io::InputItem;
10use crate::types::request_response::{RequestPayload, ResponsePayload};
11
12#[derive(Clone, Default)]
13pub struct GatewayExecutors {
14    web_search: Option<Arc<dyn GatewayExecutor>>,
15}
16
17impl GatewayExecutors {
18    #[must_use]
19    pub fn from_env(client: Arc<reqwest::Client>) -> Self {
20        Self {
21            web_search: Some(Arc::new(WebSearchHandler::from_env(client))),
22        }
23    }
24
25    pub fn insert(&mut self, executor: Arc<dyn GatewayExecutor>) {
26        match executor.tool_type() {
27            ToolType::WebSearch => self.web_search = Some(executor),
28            other => tracing::debug!(tool_type = ?other, "gateway executor type not wired yet"),
29        }
30    }
31
32    #[must_use]
33    pub fn get(&self, tool_type: ToolType) -> Option<Arc<dyn GatewayExecutor>> {
34        match tool_type {
35            ToolType::WebSearch => self.web_search.clone(),
36            ToolType::Function
37            | ToolType::CodexNamespace
38            | ToolType::Mcp
39            | ToolType::FileSearch
40            | ToolType::CodeInterpreter => None,
41        }
42    }
43}
44
45impl std::fmt::Debug for GatewayExecutors {
46    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47        f.debug_struct("GatewayExecutors")
48            .field("web_search", &self.web_search.is_some())
49            .finish()
50    }
51}
52
53/// Context built by `rehydrate_conversation`, threaded through the execute pipeline.
54#[derive(Debug)]
55pub struct RequestContext {
56    /// Untouched original request from the client.
57    pub original_request: RequestPayload,
58    /// Enriched request with rehydrated conversation history injected into `.input`.
59    /// This is the request forwarded to the LLM.
60    pub enriched_request: RequestPayload,
61    /// Only the new input items submitted by the client this turn (used for persistence).
62    pub new_input_items: Vec<InputItem>,
63    /// Our generated response ID (uuid7 with "resp_" prefix).
64    pub response_id: String,
65    /// Resolved conversation ID. `None` when `store=false` or non-conversational.
66    pub conversation_id: Option<String>,
67}
68
69impl RequestContext {
70    /// Inject our `response_id` and `conversation_id` into a `ResponsePayload`
71    /// received from the LLM (which carries the upstream's own IDs).
72    pub(crate) fn inject_ids(&self, payload: &mut ResponsePayload) {
73        payload.id.clone_from(&self.response_id);
74        payload.conversation_id.clone_from(&self.conversation_id);
75        payload
76            .previous_response_id
77            .clone_from(&self.original_request.previous_response_id);
78    }
79}
80
81/// Runtime dependencies passed into `execute()`.
82///
83/// Owns the storage handlers, HTTP client, and LLM endpoint configuration.
84/// Per-request auth is supplied via [`crate::executor::engine::ExecuteRequest::with_auth`]
85/// rather than stored here, keeping this context purely shared and immutable.
86#[derive(Clone, Debug)]
87pub struct ExecutionContext {
88    pub conv_handler: ConversationHandler,
89    pub resp_handler: ResponseHandler,
90    pub client: Arc<reqwest::Client>,
91    pub gateway_executors: GatewayExecutors,
92    /// Base URL for the LLM backend, e.g. `"http://localhost:8000"`.
93    pub llm_base_url: String,
94    /// Maximum wait time for the next SSE chunk.  `Duration::ZERO` disables the timeout.
95    /// Sourced from [`Config::streaming_chunk_timeout_s`](crate::config::Config::streaming_chunk_timeout_s).
96    pub streaming_timeout: Duration,
97}
98
99impl ExecutionContext {
100    /// Returns the full URL for the `/v1/responses` endpoint.
101    #[must_use]
102    pub fn responses_url(&self) -> String {
103        format!("{}/v1/responses", self.llm_base_url)
104    }
105
106    /// Returns the full URL for the `/v1/conversations` endpoint.
107    #[must_use]
108    pub fn conversations_url(&self) -> String {
109        format!("{}/v1/conversations", self.llm_base_url)
110    }
111
112    #[must_use]
113    pub fn new(
114        conv_handler: ConversationHandler,
115        resp_handler: ResponseHandler,
116        client: Arc<reqwest::Client>,
117        llm_base_url: String,
118    ) -> Self {
119        let gateway_executors = GatewayExecutors::from_env(Arc::clone(&client));
120        Self {
121            conv_handler,
122            resp_handler,
123            client,
124            gateway_executors,
125            llm_base_url,
126            streaming_timeout: Duration::from_secs(30),
127        }
128    }
129
130    #[must_use]
131    pub fn with_gateway_executor(mut self, executor: Arc<dyn GatewayExecutor>) -> Self {
132        self.gateway_executors.insert(executor);
133        self
134    }
135
136    /// Build an `ExecutionContext` directly from [`Config`](crate::config::Config).
137    ///
138    /// Creates the database pool, both storage handlers, and an HTTP client
139    /// internally so callers don't need to depend on the storage layer.
140    ///
141    /// # Errors
142    ///
143    /// Returns an error if the database pool cannot be opened or the schema
144    /// migration fails.
145    pub async fn from_config(cfg: &Config) -> Result<Self, Error> {
146        let db_url = cfg.db_url.as_deref().unwrap_or("sqlite://./agentic_api.db");
147        let pool = create_pool_with_schema(Some(db_url))
148            .await
149            .map_err(|e| Error::Config(format!("failed to open database '{db_url}': {e}")))?;
150
151        let conv_handler = ConversationHandler::new(ConversationStore::new(pool.clone()));
152        let resp_handler = ResponseHandler::new(ResponseStore::new(pool));
153        let client = Arc::new(reqwest::Client::new());
154        let gateway_executors = GatewayExecutors::from_env(Arc::clone(&client));
155
156        Ok(Self {
157            conv_handler,
158            resp_handler,
159            client,
160            gateway_executors,
161            llm_base_url: cfg.llm_api_base.clone(),
162            streaming_timeout: Duration::from_secs(30),
163        })
164    }
165}
agentic_core/executor/request.rs

agentic_core/executor/
request.rs
