agentic_core/executor/
request.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use crate::config::{Config, default_database_url};
5use crate::error::Error;
6use crate::executor::modes::{ConversationHandler, ResponseHandler};
7use crate::storage::backend::redact_database_urls;
8use crate::storage::{
9    ConversationStore, ConversationVersion, DatabaseBackend, ResponseStore, create_pool_with_schema_and_configs,
10};
11use crate::tool::{GatewayExecutor, GatewayExecutors};
12use crate::types::io::InputItem;
13use crate::types::messages::GatewayToolMap;
14use crate::types::request_response::{RequestPayload, ResponsePayload};
15
16/// Env var configuring client-tool → gateway-executor aliases for `/v1/messages`
17/// (e.g. `WebSearch=web_search`). Empty/unset means no aliases — client
18/// functions stay client-owned (the ownership doctrine's default).
19const GATEWAY_TOOL_ALIASES_ENV: &str = "MESSAGES_GATEWAY_TOOL_ALIASES";
20
21/// Context built by `rehydrate_conversation`, threaded through the execute pipeline.
22#[derive(Debug)]
23pub struct RequestContext {
24    /// Untouched original request from the client.
25    pub original_request: RequestPayload,
26    /// Enriched request with rehydrated conversation history injected into `.input`.
27    /// This is the request forwarded to the LLM.
28    pub enriched_request: RequestPayload,
29    /// Only the new input items submitted by the client this turn (used for persistence).
30    pub new_input_items: Vec<InputItem>,
31    /// Our generated response ID (uuid7 with "resp_" prefix).
32    pub response_id: String,
33    /// Resolved conversation ID. `None` when `store=false` or non-conversational.
34    pub conversation_id: Option<String>,
35    /// Conversation version captured with rehydrated history.
36    /// `None` for non-conversation and `previous_response_id` execution.
37    pub conversation_version: Option<ConversationVersion>,
38}
39
40impl RequestContext {
41    /// Inject our `response_id` and `conversation_id` into a `ResponsePayload`
42    /// received from the LLM (which carries the upstream's own IDs).
43    pub(crate) fn inject_ids(&self, payload: &mut ResponsePayload) {
44        payload.id.clone_from(&self.response_id);
45        payload.conversation_id.clone_from(&self.conversation_id);
46        payload
47            .previous_response_id
48            .clone_from(&self.original_request.previous_response_id);
49    }
50}
51
52/// Runtime dependencies passed into `execute()`.
53///
54/// Owns the storage handlers, HTTP client, and LLM endpoint configuration.
55/// Per-request auth is supplied via [`crate::executor::engine::ExecuteRequest::with_auth`]
56/// rather than stored here, keeping this context purely shared and immutable.
57#[derive(Clone, Debug)]
58pub struct ExecutionContext {
59    pub conv_handler: ConversationHandler,
60    pub resp_handler: ResponseHandler,
61    pub client: Arc<reqwest::Client>,
62    pub gateway_executors: GatewayExecutors,
63    /// Client-tool → gateway-executor aliases for the `/v1/messages` loop
64    /// (e.g. Claude Code's `WebSearch` → `web_search`). Empty unless configured.
65    pub messages_gateway_tools: GatewayToolMap,
66    /// Base URL for the LLM backend, e.g. `"http://localhost:8000"`.
67    pub llm_base_url: String,
68    /// Maximum wait time for the next SSE chunk.  `Duration::ZERO` disables the timeout.
69    /// Sourced from [`Config::streaming_chunk_timeout_s`](crate::config::Config::streaming_chunk_timeout_s).
70    pub streaming_timeout: Duration,
71    storage_pool: Option<Arc<crate::storage::DbPool>>,
72}
73
74impl ExecutionContext {
75    /// Returns the full URL for the `/v1/responses` endpoint.
76    #[must_use]
77    pub fn responses_url(&self) -> String {
78        format!("{}/v1/responses", self.llm_base_url)
79    }
80
81    /// Returns the full URL for the `/v1/conversations` endpoint.
82    #[must_use]
83    pub fn conversations_url(&self) -> String {
84        format!("{}/v1/conversations", self.llm_base_url)
85    }
86
87    #[must_use]
88    pub fn new(
89        conv_handler: ConversationHandler,
90        resp_handler: ResponseHandler,
91        client: Arc<reqwest::Client>,
92        llm_base_url: String,
93    ) -> Self {
94        let gateway_executors = GatewayExecutors::from_env(Arc::clone(&client));
95        Self {
96            conv_handler,
97            resp_handler,
98            client,
99            gateway_executors,
100            messages_gateway_tools: messages_gateway_tools_from_env(),
101            llm_base_url,
102            streaming_timeout: Duration::from_secs(30),
103            storage_pool: None,
104        }
105    }
106
107    #[must_use]
108    pub fn with_gateway_executor(mut self, executor: Arc<dyn GatewayExecutor>) -> Self {
109        self.gateway_executors.insert(executor);
110        self
111    }
112
113    /// Checks whether configured persistence can execute a bounded query.
114    pub async fn storage_ready(&self, timeout: Duration) -> bool {
115        let Some(pool) = &self.storage_pool else {
116            return true;
117        };
118        matches!(
119            tokio::time::timeout(timeout, crate::storage::schema::verify_persistence_ready(pool.as_ref())).await,
120            Ok(Ok(()))
121        )
122    }
123
124    /// Returns the configured persistence pool, if storage is enabled.
125    #[must_use]
126    pub fn storage_pool(&self) -> Option<&crate::storage::DbPool> {
127        self.storage_pool.as_deref()
128    }
129
130    /// Build an `ExecutionContext` directly from [`Config`](crate::config::Config).
131    ///
132    /// Creates the database pool, both storage handlers, and an HTTP client
133    /// internally so callers don't need to depend on the storage layer.
134    ///
135    /// # Errors
136    ///
137    /// Returns an error if the database pool cannot be opened or the schema
138    /// migration fails.
139    pub async fn from_config(cfg: &Config) -> Result<Self, Error> {
140        let default_db_url = cfg.db_url.is_none().then(default_database_url).transpose()?;
141        let db_url = cfg
142            .db_url
143            .as_deref()
144            .or(default_db_url.as_deref())
145            .ok_or_else(|| Error::Config("default database URL was not resolved".to_owned()))?;
146        let database_backend = DatabaseBackend::from_url(db_url)
147            .map_err(|error| Error::Config(format!("invalid DATABASE_URL: {error}")))?;
148        let pool = create_pool_with_schema_and_configs(Some(db_url), cfg.sqlite, cfg.postgres)
149            .await
150            .map_err(|error| database_open_error(database_backend, &error))?;
151        crate::storage::schema::verify_persistence_writable(pool.as_ref())
152            .await
153            .map_err(|error| database_open_error(database_backend, &error))?;
154
155        let conv_handler = ConversationHandler::new(ConversationStore::new(pool.clone()));
156        let resp_handler = ResponseHandler::new(ResponseStore::new(pool.clone()));
157        let client = Arc::new(reqwest::Client::new());
158        let gateway_executors = GatewayExecutors::from_config(Arc::clone(&client), &cfg.tools)
159            .map_err(|error| Error::Config(format!("failed to validate configured MCP server policies: {error}")))?;
160
161        Ok(Self {
162            conv_handler,
163            resp_handler,
164            client,
165            gateway_executors,
166            messages_gateway_tools: std::env::var(GATEWAY_TOOL_ALIASES_ENV)
167                .ok()
168                .as_deref()
169                .or(cfg.tools.messages_gateway_tool_aliases.as_deref())
170                .map(GatewayToolMap::from_env_str)
171                .unwrap_or_default(),
172            llm_base_url: cfg.llm_api_base.clone(),
173            streaming_timeout: Duration::from_secs(30),
174            storage_pool: Some(pool),
175        })
176    }
177}
178
179fn database_open_error(database_backend: DatabaseBackend, error: &sqlx::Error) -> Error {
180    let category = match error {
181        sqlx::Error::Configuration(_) | sqlx::Error::InvalidArgument(_) => "configuration error".to_owned(),
182        sqlx::Error::Database(database_error) => database_error
183            .code()
184            .filter(|code| code.len() <= 5 && code.bytes().all(|byte| byte.is_ascii_alphanumeric()))
185            .map_or_else(
186                || "database error".to_owned(),
187                |code| format!("database error (SQLSTATE {code})"),
188            ),
189        sqlx::Error::Io(_) => "database I/O error".to_owned(),
190        sqlx::Error::Tls(_) => "database TLS error".to_owned(),
191        sqlx::Error::Protocol(_) => "database protocol error".to_owned(),
192        sqlx::Error::PoolTimedOut => "connection pool timeout".to_owned(),
193        sqlx::Error::PoolClosed => "connection pool closed".to_owned(),
194        sqlx::Error::WorkerCrashed => "database worker crashed".to_owned(),
195        sqlx::Error::Migrate(_) => "database migration error".to_owned(),
196        _ => "database error".to_owned(),
197    };
198    let detail = redact_database_urls(&error.to_string());
199    Error::Config(format!(
200        "failed to open {} database: {category}: {detail}",
201        database_backend.display_name()
202    ))
203}
204
205/// Load the `/v1/messages` gateway-tool alias map from the environment.
206fn messages_gateway_tools_from_env() -> GatewayToolMap {
207    std::env::var(GATEWAY_TOOL_ALIASES_ENV)
208        .ok()
209        .map(|raw| GatewayToolMap::from_env_str(&raw))
210        .unwrap_or_default()
211}
212
213#[cfg(test)]
214mod tests {
215    use std::sync::Arc;
216    use std::time::Duration;
217
218    use super::{ExecutionContext, database_open_error};
219    use crate::executor::{ConversationHandler, ResponseHandler};
220    use crate::storage::{ConversationStore, DatabaseBackend, ResponseStore, create_pool_with_schema};
221
222    #[test]
223    fn database_errors_are_actionable_without_exposing_credentials() {
224        let database_url = "postgresql://agentic-api:super'secret@postgres.example.com/agentic_api";
225        let backend = DatabaseBackend::from_url(database_url).expect("valid PostgreSQL URL");
226        let tls_error = sqlx::Error::Tls(Box::new(std::io::Error::other(format!(
227            "{database_url} certificate verify failed"
228        ))));
229        let pool_error = sqlx::Error::PoolTimedOut;
230        let tls_message = database_open_error(backend, &tls_error).to_string();
231        let pool_message = database_open_error(backend, &pool_error).to_string();
232
233        assert!(tls_message.contains("database TLS error"));
234        assert!(tls_message.contains("certificate verify failed"));
235        assert!(tls_message.contains("postgresql://[redacted]"));
236        assert_eq!(
237            pool_message,
238            "failed to open PostgreSQL database: connection pool timeout: \
239             pool timed out while waiting for an open connection"
240        );
241        assert!(!tls_message.contains("super-secret"));
242        assert!(!tls_message.contains("secret"));
243        assert!(!tls_message.contains("agentic-api"));
244        assert_ne!(tls_message, pool_message);
245
246        let mysql_error = sqlx::Error::Io(std::io::Error::other(
247            "mysql://gateway:mysql-secret@mysql.example.com/agentic_api refused",
248        ));
249        let mysql_message = database_open_error(DatabaseBackend::Other, &mysql_error).to_string();
250        assert!(mysql_message.contains("mysql://[redacted]"));
251        assert!(!mysql_message.contains("mysql-secret"));
252
253        let short_postgres_url = "postgres://gateway:postgres-secret@postgres.example.com/agentic_api";
254        let short_postgres_error = sqlx::Error::Io(std::io::Error::other(format!("{short_postgres_url} refused")));
255        let short_postgres_backend = DatabaseBackend::from_url(short_postgres_url).expect("valid PostgreSQL URL");
256        let short_postgres_message = database_open_error(short_postgres_backend, &short_postgres_error).to_string();
257        assert!(short_postgres_message.contains("failed to open PostgreSQL database"));
258        assert!(short_postgres_message.contains("postgres://[redacted]"));
259        assert!(!short_postgres_message.contains("postgres-secret"));
260    }
261
262    #[test]
263    fn uppercase_database_urls_are_classified_and_redacted() {
264        let database_url = "POSTGRESQL://gateway:postgres-secret@postgres.example.com/agentic_api";
265        let error = sqlx::Error::Io(std::io::Error::other(format!("{database_url} refused")));
266        let backend = DatabaseBackend::from_url(database_url).expect("valid uppercase PostgreSQL URL");
267        let message = database_open_error(backend, &error).to_string();
268
269        assert!(message.contains("failed to open PostgreSQL database"));
270        assert!(message.contains("postgresql://[redacted]"));
271        assert!(!message.contains("postgres-secret"));
272    }
273
274    #[tokio::test]
275    async fn storage_readiness_is_bounded_by_the_probe_timeout() {
276        let pool = create_pool_with_schema(Some("sqlite://?mode=memory"))
277            .await
278            .expect("create single-connection pool");
279        let held_connection = pool.acquire().await.expect("hold the only connection");
280        let mut context = ExecutionContext::new(
281            ConversationHandler::new(ConversationStore::disabled()),
282            ResponseHandler::new(ResponseStore::disabled()),
283            Arc::new(reqwest::Client::new()),
284            "http://localhost:8000".to_owned(),
285        );
286        context.storage_pool = Some(pool.clone());
287
288        assert!(!context.storage_ready(Duration::from_millis(10)).await);
289        drop(held_connection);
290        assert!(context.storage_ready(Duration::from_secs(1)).await);
291    }
292
293    #[tokio::test]
294    async fn storage_readiness_rejects_missing_persistence_tables() {
295        let pool = create_pool_with_schema(Some("sqlite://?mode=memory"))
296            .await
297            .expect("create persistence pool");
298        let mut context = ExecutionContext::new(
299            ConversationHandler::new(ConversationStore::disabled()),
300            ResponseHandler::new(ResponseStore::disabled()),
301            Arc::new(reqwest::Client::new()),
302            "http://localhost:8000".to_owned(),
303        );
304        context.storage_pool = Some(pool.clone());
305        assert!(context.storage_ready(Duration::from_secs(1)).await);
306
307        sqlx::query("DROP TABLE responses")
308            .execute(pool.as_ref())
309            .await
310            .expect("drop persistence table");
311
312        assert!(!context.storage_ready(Duration::from_secs(1)).await);
313    }
314
315    #[tokio::test]
316    async fn storage_readiness_rejects_read_only_persistence() {
317        let pool = create_pool_with_schema(Some("sqlite://?mode=memory"))
318            .await
319            .expect("create persistence pool");
320        let mut context = ExecutionContext::new(
321            ConversationHandler::new(ConversationStore::disabled()),
322            ResponseHandler::new(ResponseStore::disabled()),
323            Arc::new(reqwest::Client::new()),
324            "http://localhost:8000".to_owned(),
325        );
326        context.storage_pool = Some(pool.clone());
327        assert!(context.storage_ready(Duration::from_secs(1)).await);
328
329        sqlx::query("PRAGMA query_only = ON")
330            .execute(pool.as_ref())
331            .await
332            .expect("make persistence read-only");
333
334        assert!(!context.storage_ready(Duration::from_secs(1)).await);
335    }
336
337    #[tokio::test]
338    async fn storage_readiness_rolls_back_probe_rows() {
339        let pool = create_pool_with_schema(Some("sqlite://?mode=memory"))
340            .await
341            .expect("create persistence pool");
342
343        crate::storage::schema::verify_persistence_writable(pool.as_ref())
344            .await
345            .expect("run functional persistence probe");
346        for table in ["conversations", "items", "responses"] {
347            let row_count: i64 = sqlx::query_scalar(&format!("SELECT COUNT(*) FROM {table}"))
348                .fetch_one(pool.as_ref())
349                .await
350                .expect("count persistence rows");
351            assert_eq!(row_count, 0, "readiness probe leaked a row into {table}");
352        }
353    }
354}
agentic_core/executor/request.rs

agentic_core/executor/
request.rs
