agentic_core/executor/
persist.rs

1//! Step 3 of the conversation pipeline — response persistence.
2//!
3//! Writes the completed response and output items to storage, routing to the
4//! appropriate handler based on whether the turn belongs to a conversation.
5
6use crate::executor::error::{ExecutorError, ExecutorResult};
7use crate::executor::modes::{ConversationHandler, ResponseHandler};
8use crate::executor::request::RequestContext;
9use crate::types::event::ResponseStatus;
10use crate::types::io::OutputItem;
11use crate::types::request_response::ResponsePayload;
12use tracing::error;
13
14#[must_use]
15pub(crate) fn should_persist(ctx: &RequestContext) -> bool {
16    ctx.original_request.store
17        || ctx.original_request.previous_response_id.is_some()
18        || ctx.original_request.conversation_id.is_some()
19}
20
21pub(crate) async fn persist_if_needed(
22    payload: ResponsePayload,
23    ctx: RequestContext,
24    conv_handler: ConversationHandler,
25    resp_handler: ResponseHandler,
26) -> ExecutorResult<()> {
27    if should_persist(&ctx) {
28        persist_response(payload, ctx, conv_handler, resp_handler)
29            .await
30            .map_err(|source| {
31                error!(error = ?source, "failed to persist response");
32                ExecutorError::Persistence(Box::new(source))
33            })
34    } else {
35        Ok(())
36    }
37}
38
39/// Step 3 — Persist the completed response to storage.
40///
41/// Skipped if [`ResponseStatus`] is not `Completed`/`Incomplete` or `payload.id` is empty.
42/// Routes explicit `conversation_id` requests to [`ConversationHandler`] and
43/// all other requests, including `previous_response_id` continuations, to [`ResponseHandler`].
44///
45/// # Errors
46/// Returns [`ExecutorError`] if the storage operation fails.
47pub async fn persist_response(
48    payload: ResponsePayload,
49    ctx: RequestContext,
50    conv_handler: ConversationHandler,
51    resp_handler: ResponseHandler,
52) -> ExecutorResult<()> {
53    // Use typed enum — no hardcoded status strings.
54    if !matches!(
55        payload.status.parse::<ResponseStatus>().unwrap_or_default(),
56        ResponseStatus::Completed | ResponseStatus::Incomplete
57    ) || payload.id.is_empty()
58    {
59        return Ok(());
60    }
61
62    persist_turn(ctx, payload.output, &conv_handler, &resp_handler).await
63}
64
65/// Persists one completed turn with the handler selected by its explicit conversation discriminator.
66///
67/// # Errors
68/// Returns [`ExecutorError`] if the selected storage operation fails.
69pub async fn persist_turn(
70    ctx: RequestContext,
71    output_items: Vec<OutputItem>,
72    conv_handler: &ConversationHandler,
73    resp_handler: &ResponseHandler,
74) -> ExecutorResult<()> {
75    if ctx.original_request.conversation_id.is_some() {
76        conv_handler.execute_turn(ctx, output_items).await
77    } else {
78        resp_handler.execute_turn(ctx, output_items).await
79    }
80}
agentic_core/executor/persist.rs

agentic_core/executor/
persist.rs
