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::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 db_url = cfg.db_url.as_deref().unwrap_or("sqlite://./agentic_api.db");
141        let database_backend = DatabaseBackend::from_url(db_url)
142            .map_err(|error| Error::Config(format!("invalid DATABASE_URL: {error}")))?;
143        let pool = create_pool_with_schema_and_configs(Some(db_url), cfg.sqlite, cfg.postgres)
144            .await
145            .map_err(|error| database_open_error(database_backend, &error))?;
146        crate::storage::schema::verify_persistence_writable(pool.as_ref())
147            .await
148            .map_err(|error| database_open_error(database_backend, &error))?;
149
150        let conv_handler = ConversationHandler::new(ConversationStore::new(pool.clone()));
151        let resp_handler = ResponseHandler::new(ResponseStore::new(pool.clone()));
152        let client = Arc::new(reqwest::Client::new());
153        let gateway_executors = GatewayExecutors::from_env(Arc::clone(&client));
154
155        Ok(Self {
156            conv_handler,
157            resp_handler,
158            client,
159            gateway_executors,
160            messages_gateway_tools: messages_gateway_tools_from_env(),
161            llm_base_url: cfg.llm_api_base.clone(),
162            streaming_timeout: Duration::from_secs(30),
163            storage_pool: Some(pool),
164        })
165    }
166}
167
168fn database_open_error(database_backend: DatabaseBackend, error: &sqlx::Error) -> Error {
169    let category = match error {
170        sqlx::Error::Configuration(_) | sqlx::Error::InvalidArgument(_) => "configuration error".to_owned(),
171        sqlx::Error::Database(database_error) => database_error
172            .code()
173            .filter(|code| code.len() <= 5 && code.bytes().all(|byte| byte.is_ascii_alphanumeric()))
174            .map_or_else(
175                || "database error".to_owned(),
176                |code| format!("database error (SQLSTATE {code})"),
177            ),
178        sqlx::Error::Io(_) => "database I/O error".to_owned(),
179        sqlx::Error::Tls(_) => "database TLS error".to_owned(),
180        sqlx::Error::Protocol(_) => "database protocol error".to_owned(),
181        sqlx::Error::PoolTimedOut => "connection pool timeout".to_owned(),
182        sqlx::Error::PoolClosed => "connection pool closed".to_owned(),
183        sqlx::Error::WorkerCrashed => "database worker crashed".to_owned(),
184        sqlx::Error::Migrate(_) => "database migration error".to_owned(),
185        _ => "database error".to_owned(),
186    };
187    let detail = redact_database_urls(&error.to_string());
188    Error::Config(format!(
189        "failed to open {} database: {category}: {detail}",
190        database_backend.display_name()
191    ))
192}
193
194/// Load the `/v1/messages` gateway-tool alias map from the environment.
195fn messages_gateway_tools_from_env() -> GatewayToolMap {
196    std::env::var(GATEWAY_TOOL_ALIASES_ENV)
197        .ok()
198        .map(|raw| GatewayToolMap::from_env_str(&raw))
199        .unwrap_or_default()
200}
201
202#[cfg(test)]
203mod tests {
204    use std::sync::Arc;
205    use std::time::Duration;
206
207    use super::{ExecutionContext, database_open_error};
208    use crate::executor::{ConversationHandler, ResponseHandler};
209    use crate::storage::{ConversationStore, DatabaseBackend, ResponseStore, create_pool_with_schema};
210
211    #[test]
212    fn database_errors_are_actionable_without_exposing_credentials() {
213        let database_url = "postgresql://agentic-api:super'secret@postgres.example.com/agentic_api";
214        let backend = DatabaseBackend::from_url(database_url).expect("valid PostgreSQL URL");
215        let tls_error = sqlx::Error::Tls(Box::new(std::io::Error::other(format!(
216            "{database_url} certificate verify failed"
217        ))));
218        let pool_error = sqlx::Error::PoolTimedOut;
219        let tls_message = database_open_error(backend, &tls_error).to_string();
220        let pool_message = database_open_error(backend, &pool_error).to_string();
221
222        assert!(tls_message.contains("database TLS error"));
223        assert!(tls_message.contains("certificate verify failed"));
224        assert!(tls_message.contains("postgresql://[redacted]"));
225        assert_eq!(
226            pool_message,
227            "failed to open PostgreSQL database: connection pool timeout: \
228             pool timed out while waiting for an open connection"
229        );
230        assert!(!tls_message.contains("super-secret"));
231        assert!(!tls_message.contains("secret"));
232        assert!(!tls_message.contains("agentic-api"));
233        assert_ne!(tls_message, pool_message);
234
235        let mysql_error = sqlx::Error::Io(std::io::Error::other(
236            "mysql://gateway:mysql-secret@mysql.example.com/agentic_api refused",
237        ));
238        let mysql_message = database_open_error(DatabaseBackend::Other, &mysql_error).to_string();
239        assert!(mysql_message.contains("mysql://[redacted]"));
240        assert!(!mysql_message.contains("mysql-secret"));
241
242        let short_postgres_url = "postgres://gateway:postgres-secret@postgres.example.com/agentic_api";
243        let short_postgres_error = sqlx::Error::Io(std::io::Error::other(format!("{short_postgres_url} refused")));
244        let short_postgres_backend = DatabaseBackend::from_url(short_postgres_url).expect("valid PostgreSQL URL");
245        let short_postgres_message = database_open_error(short_postgres_backend, &short_postgres_error).to_string();
246        assert!(short_postgres_message.contains("failed to open PostgreSQL database"));
247        assert!(short_postgres_message.contains("postgres://[redacted]"));
248        assert!(!short_postgres_message.contains("postgres-secret"));
249    }
250
251    #[test]
252    fn uppercase_database_urls_are_classified_and_redacted() {
253        let database_url = "POSTGRESQL://gateway:postgres-secret@postgres.example.com/agentic_api";
254        let error = sqlx::Error::Io(std::io::Error::other(format!("{database_url} refused")));
255        let backend = DatabaseBackend::from_url(database_url).expect("valid uppercase PostgreSQL URL");
256        let message = database_open_error(backend, &error).to_string();
257
258        assert!(message.contains("failed to open PostgreSQL database"));
259        assert!(message.contains("postgresql://[redacted]"));
260        assert!(!message.contains("postgres-secret"));
261    }
262
263    #[tokio::test]
264    async fn storage_readiness_is_bounded_by_the_probe_timeout() {
265        let pool = create_pool_with_schema(Some("sqlite://?mode=memory"))
266            .await
267            .expect("create single-connection pool");
268        let held_connection = pool.acquire().await.expect("hold the only connection");
269        let mut context = ExecutionContext::new(
270            ConversationHandler::new(ConversationStore::disabled()),
271            ResponseHandler::new(ResponseStore::disabled()),
272            Arc::new(reqwest::Client::new()),
273            "http://localhost:8000".to_owned(),
274        );
275        context.storage_pool = Some(pool.clone());
276
277        assert!(!context.storage_ready(Duration::from_millis(10)).await);
278        drop(held_connection);
279        assert!(context.storage_ready(Duration::from_secs(1)).await);
280    }
281
282    #[tokio::test]
283    async fn storage_readiness_rejects_missing_persistence_tables() {
284        let pool = create_pool_with_schema(Some("sqlite://?mode=memory"))
285            .await
286            .expect("create persistence pool");
287        let mut context = ExecutionContext::new(
288            ConversationHandler::new(ConversationStore::disabled()),
289            ResponseHandler::new(ResponseStore::disabled()),
290            Arc::new(reqwest::Client::new()),
291            "http://localhost:8000".to_owned(),
292        );
293        context.storage_pool = Some(pool.clone());
294        assert!(context.storage_ready(Duration::from_secs(1)).await);
295
296        sqlx::query("DROP TABLE responses")
297            .execute(pool.as_ref())
298            .await
299            .expect("drop persistence table");
300
301        assert!(!context.storage_ready(Duration::from_secs(1)).await);
302    }
303
304    #[tokio::test]
305    async fn storage_readiness_rejects_read_only_persistence() {
306        let pool = create_pool_with_schema(Some("sqlite://?mode=memory"))
307            .await
308            .expect("create persistence pool");
309        let mut context = ExecutionContext::new(
310            ConversationHandler::new(ConversationStore::disabled()),
311            ResponseHandler::new(ResponseStore::disabled()),
312            Arc::new(reqwest::Client::new()),
313            "http://localhost:8000".to_owned(),
314        );
315        context.storage_pool = Some(pool.clone());
316        assert!(context.storage_ready(Duration::from_secs(1)).await);
317
318        sqlx::query("PRAGMA query_only = ON")
319            .execute(pool.as_ref())
320            .await
321            .expect("make persistence read-only");
322
323        assert!(!context.storage_ready(Duration::from_secs(1)).await);
324    }
325
326    #[tokio::test]
327    async fn storage_readiness_rolls_back_probe_rows() {
328        let pool = create_pool_with_schema(Some("sqlite://?mode=memory"))
329            .await
330            .expect("create persistence pool");
331
332        crate::storage::schema::verify_persistence_writable(pool.as_ref())
333            .await
334            .expect("run functional persistence probe");
335        for table in ["conversations", "items", "responses"] {
336            let row_count: i64 = sqlx::query_scalar(&format!("SELECT COUNT(*) FROM {table}"))
337                .fetch_one(pool.as_ref())
338                .await
339                .expect("count persistence rows");
340            assert_eq!(row_count, 0, "readiness probe leaked a row into {table}");
341        }
342    }
343}
agentic_core/executor/request.rs

agentic_core/executor/
request.rs
