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    };
42
43    if ctx.original_request.conversation_id.is_some() && ctx.original_request.previous_response_id.is_some() {
44        return Err(ExecutorError::InvalidRequest(
45            "provide only one of conversation_id or previous_response_id".into(),
46        ));
47    }
48
49    if ctx.original_request.conversation_id.is_some() {
50        from_conversation(&mut ctx, exec_ctx).await?;
51        return Ok(ctx);
52    }
53
54    if ctx.original_request.previous_response_id.is_some() {
55        from_response(&mut ctx, exec_ctx).await?;
56        return Ok(ctx);
57    }
58
59    ctx.enriched_request.input = ResponsesInput::Items(ctx.new_input_items.clone());
60    Ok(ctx)
61}
62
63/// Hydrates `ctx` from the previous response chain.
64///
65/// Loads the stored response, rehydrates its history items, resolves effective
66/// tools and tool choice from the stored metadata, and prepends the history to
67/// the enriched request input.
68async fn from_response(ctx: &mut RequestContext, exec_ctx: &ExecutionContext) -> ExecutorResult<()> {
69    let stored = exec_ctx.resp_handler.get(ctx).await?;
70    let history = exec_ctx.resp_handler.rehydrate(ctx).await?;
71
72    let mut items = InOutItem::into_input_items(history);
73    items.reserve(ctx.new_input_items.len());
74    items.extend(ctx.new_input_items.iter().cloned());
75
76    ctx.enriched_request.previous_response_id = None;
77    ctx.enriched_request.input = ResponsesInput::Items(items);
78    ctx.enriched_request.tools = resolve_tools(
79        ctx.original_request.tools.as_deref(),
80        stored.metadata.effective_tools.as_deref(),
81        ctx.original_request.tools.is_some(),
82    );
83    ctx.enriched_request.tool_choice = Some(resolve_tool_choice(
84        ctx.original_request.tool_choice.as_ref(),
85        &stored.metadata.effective_tool_choice,
86        ctx.original_request.tool_choice.is_some(),
87    ));
88    ctx.conversation_id = stored.conversation_id;
89    Ok(())
90}
91
92/// Hydrates `ctx` from the conversation store.
93///
94/// Gets or creates the conversation (depending on `store`) and rehydrates its
95/// history in parallel, then prepends the history items to the enriched request input.
96async fn from_conversation(ctx: &mut RequestContext, exec_ctx: &ExecutionContext) -> ExecutorResult<()> {
97    let (conv_data, history) = tokio::try_join!(
98        async {
99            if ctx.original_request.store {
100                exec_ctx.conv_handler.get_or_create(ctx).await
101            } else {
102                exec_ctx.conv_handler.get(ctx).await
103            }
104        },
105        exec_ctx.conv_handler.rehydrate(ctx),
106    )?;
107
108    let mut items = InOutItem::into_input_items(history);
109    items.reserve(ctx.new_input_items.len());
110    items.extend(ctx.new_input_items.iter().cloned());
111
112    ctx.enriched_request.input = ResponsesInput::Items(items);
113    ctx.conversation_id = Some(conv_data.conversation_id);
114    Ok(())
115}
agentic_core/executor/rehydrate.rs

agentic_core/executor/
rehydrate.rs
