agentic_core/executor/modes/
response.rs

1//! Response storage handler — owns all response store operations.
2
3use crate::storage::{InOutItem, ResponseData, ResponseMetadata, ResponseStore};
4use crate::types::io::OutputItem;
5
6use crate::executor::error::{ExecutorError, ExecutorResult};
7use crate::executor::request::RequestContext;
8
9/// Handles all response store operations: lookup, rehydration, and persistence.
10#[derive(Clone, Debug)]
11pub struct ResponseHandler {
12    store: ResponseStore,
13}
14
15impl ResponseHandler {
16    #[must_use]
17    pub fn new(store: ResponseStore) -> Self {
18        Self { store }
19    }
20
21    /// Retrieves the stored response for `previous_response_id`.
22    ///
23    /// Reads `previous_response_id` from `ctx.original_request`.
24    ///
25    /// # Errors
26    /// Returns `ExecutorError` if `previous_response_id` is absent, the response
27    /// is not found, the store is disabled, or the database query fails.
28    pub async fn get(&self, ctx: &RequestContext) -> ExecutorResult<ResponseData> {
29        let prev_id = ctx
30            .original_request
31            .previous_response_id
32            .as_deref()
33            .ok_or_else(|| ExecutorError::InvalidRequest("previous_response_id is required for get".into()))?;
34        self.store.get(prev_id).await.map_err(ExecutorError::Storage)
35    }
36
37    /// Validates that the response for `previous_response_id` exists.
38    ///
39    /// Used in the `store=false` path where we only need to confirm the ID is
40    /// valid without loading any history.
41    ///
42    /// # Errors
43    /// Returns `ExecutorError` if `previous_response_id` is absent, the response
44    /// is not found, or the store is disabled.
45    pub async fn validate_exists(&self, ctx: &RequestContext) -> ExecutorResult<()> {
46        self.get(ctx).await.map(|_| ())
47    }
48
49    /// Loads all history items referenced by the previous response.
50    ///
51    /// Reads `previous_response_id` from `ctx.original_request`. Returns an empty
52    /// vec if there is no previous response.
53    ///
54    /// # Errors
55    /// Returns `ExecutorError` if the store is disabled or the database query fails.
56    pub async fn rehydrate(&self, ctx: &RequestContext) -> ExecutorResult<Vec<InOutItem>> {
57        let Some(prev_id) = ctx.original_request.previous_response_id.as_deref() else {
58            return Ok(vec![]);
59        };
60        self.store.rehydrate(prev_id).await.map_err(ExecutorError::Storage)
61    }
62
63    /// Persists a response record — only the new items from this turn.
64    ///
65    /// Takes `ctx` and `output_items` by value so fields can be moved directly
66    /// into [`ResponseMetadata`] without cloning. Prior history must not be
67    /// re-inserted; the response store records item IDs for this response only.
68    ///
69    /// # Errors
70    /// Returns `ExecutorError` if the store is disabled or the database operation fails.
71    pub async fn execute_turn(&self, ctx: RequestContext, output_items: Vec<OutputItem>) -> ExecutorResult<()> {
72        let metadata = ResponseMetadata {
73            model: ctx.enriched_request.model,
74            previous_response_id: ctx.original_request.previous_response_id,
75            effective_tools: ctx.enriched_request.tools,
76            effective_tool_choice: ctx.enriched_request.tool_choice.unwrap_or_default(),
77            effective_instructions: ctx.enriched_request.instructions,
78        };
79
80        let mut new_items = Vec::with_capacity(ctx.new_input_items.len() + output_items.len());
81        new_items.extend(ctx.new_input_items.into_iter().map(InOutItem::Input));
82        new_items.extend(output_items.into_iter().map(InOutItem::Output));
83
84        self.store
85            .persist(
86                &ctx.response_id,
87                metadata.previous_response_id.as_deref(),
88                new_items,
89                &metadata,
90            )
91            .await
92            .map_err(ExecutorError::Storage)
93    }
94}
95
96#[cfg(test)]
97mod tests {
98    use super::*;
99    use crate::types::io::ResponsesInput;
100    use crate::types::request_response::RequestPayload;
101
102    fn disabled_handler() -> ResponseHandler {
103        ResponseHandler::new(ResponseStore::disabled())
104    }
105
106    fn make_ctx(previous_response_id: Option<&str>) -> RequestContext {
107        let req = RequestPayload {
108            model: "test".into(),
109            input: ResponsesInput::Text("hi".into()),
110            instructions: None,
111            previous_response_id: previous_response_id.map(str::to_string),
112            conversation_id: None,
113            tools: None,
114            tool_choice: None,
115            stream: false,
116            store: true,
117            include: None,
118            temperature: None,
119            top_p: None,
120            max_output_tokens: None,
121            truncation: None,
122            metadata: None,
123        };
124        RequestContext {
125            enriched_request: req.clone(),
126            original_request: req,
127            new_input_items: vec![],
128            response_id: "resp_test".into(),
129            conversation_id: None,
130        }
131    }
132
133    #[tokio::test]
134    async fn test_get_missing_prev_id_returns_error() {
135        let result = disabled_handler().get(&make_ctx(None)).await;
136        assert!(result.is_err());
137    }
138
139    #[tokio::test]
140    async fn test_validate_exists_missing_prev_id_returns_error() {
141        let result = disabled_handler().validate_exists(&make_ctx(None)).await;
142        assert!(result.is_err());
143    }
144
145    #[tokio::test]
146    async fn test_rehydrate_no_prev_id_returns_empty() {
147        let result = disabled_handler().rehydrate(&make_ctx(None)).await;
148        assert!(result.is_ok());
149        assert!(result.unwrap().is_empty());
150    }
151
152    #[tokio::test]
153    async fn test_rehydrate_disabled_store_returns_error() {
154        let result = disabled_handler().rehydrate(&make_ctx(Some("resp_prev"))).await;
155        assert!(result.is_err());
156    }
157
158    #[tokio::test]
159    async fn test_execute_turn_disabled_store_returns_error() {
160        let result = disabled_handler().execute_turn(make_ctx(None), vec![]).await;
161        assert!(result.is_err());
162    }
163}
agentic_core/executor/modes/response.rs

agentic_core/executor/modes/
response.rs
