agentic_core/executor/
rehydrate.rs

1//! Step 1 of the conversation pipeline — history rehydration.
2//!
3//! Builds a [`RequestContext`] by loading prior turns from storage and
4//! injecting them into the enriched request before it is forwarded to the LLM.
5
6use crate::executor::error::{ExecutorError, ExecutorResult};
7use crate::executor::request::{ExecutionContext, RequestContext};
8use crate::storage::InOutItem;
9use crate::types::io::{InputItem, ResponsesInput, resolve_tool_choice, resolve_tools};
10use crate::types::request_response::RequestPayload;
11use crate::utils::uuid7_str;
12
13/// Step 1 — Build [`RequestContext`] by rehydrating conversation history.
14///
15/// `request` is moved into the context as `enriched_request`; one clone is taken
16/// for `original_request` so the engine retains an unmodified copy for persistence
17/// and ID resolution.
18///
19/// Dispatches based on `store` flag and which ID is present:
20/// - `previous_response_id`: rehydrate from the prior response checkpoint
21/// - `conversation_id`:      rehydrate from the conversation
22/// - no ids:                 forward only the new input
23///
24/// # Errors
25/// Returns [`ExecutorError`] if storage is unavailable or a referenced ID does not exist.
26pub async fn rehydrate_conversation(
27    request: RequestPayload,
28    exec_ctx: &ExecutionContext,
29) -> ExecutorResult<RequestContext> {
30    let response_id = uuid7_str("resp_");
31    let new_input_items: Vec<InputItem> = Vec::from(&request.input);
32
33    // One clone for the unmodified original; `request` is moved as enriched_request.
34    let original_request = request.clone();
35    let mut ctx = RequestContext {
36        enriched_request: request,
37        original_request,
38        new_input_items,
39        response_id,
40        conversation_id: None,
41        conversation_version: None,
42    };
43
44    if ctx.original_request.conversation_id.is_some() && ctx.original_request.previous_response_id.is_some() {
45        return Err(ExecutorError::InvalidRequest(
46            "provide only one of conversation_id or previous_response_id".into(),
47        ));
48    }
49
50    if ctx.original_request.conversation_id.is_some() {
51        from_conversation(&mut ctx, exec_ctx).await?;
52        return Ok(ctx);
53    }
54
55    if ctx.original_request.previous_response_id.is_some() {
56        from_response(&mut ctx, exec_ctx).await?;
57        return Ok(ctx);
58    }
59
60    ctx.enriched_request.input = ResponsesInput::Items(ctx.new_input_items.clone());
61    Ok(ctx)
62}
63
64/// Hydrates `ctx` from the previous response chain.
65///
66/// Loads the stored response, rehydrates its history items, resolves effective
67/// tools and tool choice from the stored metadata, and prepends the history to
68/// the enriched request input.
69async fn from_response(ctx: &mut RequestContext, exec_ctx: &ExecutionContext) -> ExecutorResult<()> {
70    let stored = exec_ctx.resp_handler.get(ctx).await?;
71    let history = exec_ctx.resp_handler.rehydrate(ctx).await?;
72
73    let mut items = InOutItem::into_input_items(history);
74    items.reserve(ctx.new_input_items.len());
75    items.extend(ctx.new_input_items.iter().cloned());
76
77    ctx.enriched_request.previous_response_id = None;
78    ctx.enriched_request.input = ResponsesInput::Items(items);
79    ctx.enriched_request.tools = resolve_tools(
80        ctx.original_request.tools.as_deref(),
81        stored.metadata.effective_tools.as_deref(),
82        ctx.original_request.tools.is_some(),
83    );
84    ctx.enriched_request.tool_choice = Some(resolve_tool_choice(
85        ctx.original_request.tool_choice.as_ref(),
86        &stored.metadata.effective_tool_choice,
87        ctx.original_request.tool_choice.is_some(),
88    ));
89    ctx.conversation_id = stored.conversation_id;
90    Ok(())
91}
92
93/// Hydrates `ctx` from the conversation store.
94///
95/// Gets or creates the conversation (depending on `store`) and rehydrates its
96/// history in parallel, then prepends the history items to the enriched request input.
97async fn from_conversation(ctx: &mut RequestContext, exec_ctx: &ExecutionContext) -> ExecutorResult<()> {
98    let (conv_data, snapshot) = tokio::try_join!(
99        async {
100            if ctx.original_request.store {
101                exec_ctx.conv_handler.get_or_create(ctx).await
102            } else {
103                exec_ctx.conv_handler.get(ctx).await
104            }
105        },
106        exec_ctx.conv_handler.rehydrate_snapshot(ctx),
107    )?;
108
109    let mut items = InOutItem::into_input_items(snapshot.items);
110    items.reserve(ctx.new_input_items.len());
111    items.extend(ctx.new_input_items.iter().cloned());
112
113    ctx.enriched_request.input = ResponsesInput::Items(items);
114    ctx.conversation_id = Some(conv_data.conversation_id);
115    ctx.conversation_version = Some(snapshot.version);
116    Ok(())
117}
118
119#[cfg(test)]
120mod tests {
121    use std::sync::Arc;
122
123    use super::*;
124    use crate::executor::modes::{ConversationHandler, ResponseHandler};
125    use crate::storage::{
126        ConversationStore, ConversationVersion, InOutItem, ResponseMetadata, ResponseStore, create_pool_with_schema,
127    };
128    use crate::types::request_response::RequestPayload;
129
130    fn request(conversation_id: Option<&str>, previous_response_id: Option<&str>) -> RequestPayload {
131        RequestPayload {
132            model: "test".into(),
133            input: ResponsesInput::Text("new input".into()),
134            instructions: None,
135            previous_response_id: previous_response_id.map(str::to_owned),
136            conversation_id: conversation_id.map(str::to_owned),
137            tools: None,
138            tool_choice: None,
139            stream: false,
140            store: true,
141            include: None,
142            temperature: None,
143            top_p: None,
144            max_output_tokens: None,
145            truncation: None,
146            metadata: None,
147            parallel_tool_calls: None,
148            cache_salt: None,
149            context_management: None,
150        }
151    }
152
153    fn execution_context(conversation_store: ConversationStore, response_store: ResponseStore) -> ExecutionContext {
154        ExecutionContext::new(
155            ConversationHandler::new(conversation_store),
156            ResponseHandler::new(response_store),
157            Arc::new(reqwest::Client::new()),
158            "http://localhost:8000".to_owned(),
159        )
160    }
161
162    #[tokio::test]
163    async fn new_conversation_rehydration_captures_empty_version() -> Result<(), Box<dyn std::error::Error>> {
164        let pool = create_pool_with_schema(Some("sqlite://?mode=memory")).await?;
165        let conversation_store = ConversationStore::new(pool);
166        let conversation = conversation_store.create().await?;
167        let exec_ctx = execution_context(conversation_store, ResponseStore::disabled());
168
169        let ctx = rehydrate_conversation(request(Some(&conversation.conversation_id), None), &exec_ctx).await?;
170
171        assert_eq!(ctx.conversation_version, Some(ConversationVersion::Empty));
172        Ok(())
173    }
174
175    #[tokio::test]
176    async fn existing_conversation_rehydration_captures_last_sequence() -> Result<(), Box<dyn std::error::Error>> {
177        let pool = create_pool_with_schema(Some("sqlite://?mode=memory")).await?;
178        let conversation_store = ConversationStore::new(pool);
179        let conversation = conversation_store.create().await?;
180        let prior_items = Vec::<InputItem>::from(&ResponsesInput::Text("prior input".into()))
181            .into_iter()
182            .map(InOutItem::Input)
183            .collect();
184        conversation_store
185            .persist(
186                &conversation.conversation_id,
187                "resp_prior",
188                None,
189                prior_items,
190                &ResponseMetadata::default(),
191            )
192            .await?;
193        let exec_ctx = execution_context(conversation_store, ResponseStore::disabled());
194
195        let ctx = rehydrate_conversation(request(Some(&conversation.conversation_id), None), &exec_ctx).await?;
196
197        assert_eq!(ctx.conversation_version, Some(ConversationVersion::LastSequence(0)));
198        Ok(())
199    }
200
201    #[tokio::test]
202    async fn request_without_continuation_has_no_conversation_version() -> Result<(), Box<dyn std::error::Error>> {
203        let exec_ctx = execution_context(ConversationStore::disabled(), ResponseStore::disabled());
204
205        let ctx = rehydrate_conversation(request(None, None), &exec_ctx).await?;
206
207        assert_eq!(ctx.conversation_version, None);
208        Ok(())
209    }
210
211    #[tokio::test]
212    async fn previous_response_rehydration_has_no_conversation_version() -> Result<(), Box<dyn std::error::Error>> {
213        let pool = create_pool_with_schema(Some("sqlite://?mode=memory")).await?;
214        let response_store = ResponseStore::new(pool);
215        response_store
216            .persist("resp_prior", None, Vec::new(), &ResponseMetadata::default())
217            .await?;
218        let exec_ctx = execution_context(ConversationStore::disabled(), response_store);
219
220        let ctx = rehydrate_conversation(request(None, Some("resp_prior")), &exec_ctx).await?;
221
222        assert_eq!(ctx.conversation_version, None);
223        Ok(())
224    }
225}
agentic_core/executor/rehydrate.rs

agentic_core/executor/
rehydrate.rs
