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_with_conversation_id(
86                &ctx.response_id,
87                ctx.conversation_id.as_deref(),
88                metadata.previous_response_id.as_deref(),
89                new_items,
90                &metadata,
91            )
92            .await
93            .map_err(ExecutorError::Storage)
94    }
95}
96
97#[cfg(test)]
98mod tests {
99    use super::*;
100    use crate::types::io::ResponsesInput;
101    use crate::types::request_response::RequestPayload;
102
103    fn disabled_handler() -> ResponseHandler {
104        ResponseHandler::new(ResponseStore::disabled())
105    }
106
107    fn make_ctx(previous_response_id: Option<&str>) -> RequestContext {
108        let req = RequestPayload {
109            model: "test".into(),
110            input: ResponsesInput::Text("hi".into()),
111            instructions: None,
112            previous_response_id: previous_response_id.map(str::to_string),
113            conversation_id: None,
114            tools: None,
115            tool_choice: None,
116            stream: false,
117            store: true,
118            include: None,
119            temperature: None,
120            top_p: None,
121            max_output_tokens: None,
122            truncation: None,
123            metadata: None,
124            parallel_tool_calls: None,
125            cache_salt: None,
126            context_management: None,
127        };
128        RequestContext {
129            enriched_request: req.clone(),
130            original_request: req,
131            new_input_items: vec![],
132            response_id: "resp_test".into(),
133            conversation_id: None,
134            conversation_version: None,
135        }
136    }
137
138    #[tokio::test]
139    async fn test_get_missing_prev_id_returns_error() {
140        let result = disabled_handler().get(&make_ctx(None)).await;
141        assert!(result.is_err());
142    }
143
144    #[tokio::test]
145    async fn test_validate_exists_missing_prev_id_returns_error() {
146        let result = disabled_handler().validate_exists(&make_ctx(None)).await;
147        assert!(result.is_err());
148    }
149
150    #[tokio::test]
151    async fn test_rehydrate_no_prev_id_returns_empty() {
152        let result = disabled_handler().rehydrate(&make_ctx(None)).await;
153        assert!(result.is_ok());
154        assert!(result.unwrap().is_empty());
155    }
156
157    #[tokio::test]
158    async fn test_rehydrate_disabled_store_returns_error() {
159        let result = disabled_handler().rehydrate(&make_ctx(Some("resp_prev"))).await;
160        assert!(result.is_err());
161    }
162
163    #[tokio::test]
164    async fn test_execute_turn_disabled_store_returns_error() {
165        let result = disabled_handler().execute_turn(make_ctx(None), vec![]).await;
166        assert!(result.is_err());
167    }
168}
agentic_core/executor/modes/response.rs

agentic_core/executor/modes/
response.rs
