starweaver_runtime/
executor.rs

1//! Durable execution checkpoints for agent runs.
2
3use std::sync::Arc;
4
5use async_trait::async_trait;
6use serde::{Deserialize, Serialize};
7use starweaver_core::{CheckpointId, ConversationId, Metadata, RunId, TraceContext};
8use starweaver_usage::Usage;
9use thiserror::Error;
10
11use crate::run::{AgentRunState, RunStatus};
12
13/// Named execution boundary in the agent loop.
14#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
15#[serde(rename_all = "snake_case")]
16pub enum AgentExecutionNode {
17    /// Run initialization completed.
18    RunStart,
19    /// The next model request is about to be prepared.
20    PrepareModelRequest,
21    /// The model request is about to be sent or skipped by policy.
22    BeforeModelRequest,
23    /// A model response has been applied to state.
24    ModelResponse,
25    /// A function tool call is about to execute.
26    ToolCall,
27    /// A function tool result has been applied to state.
28    ToolReturn,
29    /// Final output validation is about to run.
30    ValidateOutput,
31    /// The run completed.
32    RunComplete,
33    /// The run failed.
34    RunFailed,
35}
36
37/// Compact cursor values durable stores use to resume and audit a run.
38#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
39pub struct AgentResumeCursor {
40    /// Index of the next model request attempt.
41    pub model_request_attempt: usize,
42    /// Current tool call batch identifier when the run is inside a tool boundary.
43    #[serde(default, skip_serializing_if = "Option::is_none")]
44    pub tool_call_batch_id: Option<String>,
45    /// Index of the next output validation attempt.
46    pub output_validation_attempt: usize,
47    /// Cursor of the last persisted stream event.
48    #[serde(default, skip_serializing_if = "Option::is_none")]
49    pub stream_cursor: Option<usize>,
50    /// Last canonical message index persisted by the runtime.
51    pub message_cursor: usize,
52}
53
54/// Stable resume evidence for durable service runtimes and external `SessionStore` implementations.
55#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
56pub struct AgentResumeEvidence {
57    /// Execution boundary captured by this checkpoint.
58    pub node: AgentExecutionNode,
59    /// Current run status.
60    pub status: RunStatus,
61    /// Completed run step at this boundary.
62    pub run_step: usize,
63    /// Resume cursors for replay and continuation.
64    pub cursor: AgentResumeCursor,
65    /// Accumulated usage snapshot.
66    pub usage: Usage,
67    /// Context state revision or hash provided by a service layer.
68    #[serde(default, skip_serializing_if = "Option::is_none")]
69    pub context_revision: Option<String>,
70    /// Environment provider state reference provided by a service layer.
71    #[serde(default, skip_serializing_if = "Option::is_none")]
72    pub environment_ref: Option<String>,
73    /// Pending approval count.
74    pub pending_approval_count: usize,
75    /// Deferred tool return count.
76    pub deferred_tool_count: usize,
77    /// Trace correlation snapshot.
78    #[serde(default, skip_serializing_if = "TraceContext::is_empty")]
79    pub trace_context: TraceContext,
80    /// Resume metadata for `SessionStore` implementations.
81    #[serde(default, skip_serializing_if = "Metadata::is_empty")]
82    pub metadata: Metadata,
83}
84
85impl AgentResumeEvidence {
86    /// Build resume evidence from run state and boundary metadata.
87    #[must_use]
88    pub fn new(node: AgentExecutionNode, state: &AgentRunState) -> Self {
89        Self {
90            node,
91            status: state.status,
92            run_step: state.run_step,
93            cursor: AgentResumeCursor {
94                model_request_attempt: state.run_step,
95                tool_call_batch_id: (!state.pending_tool_calls.is_empty())
96                    .then(|| format!("tool_batch_{}", state.run_step)),
97                output_validation_attempt: 0,
98                stream_cursor: None,
99                message_cursor: state.message_history.len(),
100            },
101            usage: state.usage.clone(),
102            context_revision: state
103                .metadata
104                .get("context_revision")
105                .and_then(serde_json::Value::as_str)
106                .map(str::to_string),
107            environment_ref: state
108                .metadata
109                .get("environment_ref")
110                .and_then(serde_json::Value::as_str)
111                .map(str::to_string),
112            pending_approval_count: state.pending_approval_tool_returns.len(),
113            deferred_tool_count: state.deferred_tool_returns.len(),
114            trace_context: TraceContext::default(),
115            metadata: Metadata::default(),
116        }
117    }
118
119    /// Attach stream cursor.
120    #[must_use]
121    pub const fn with_stream_cursor(mut self, stream_cursor: usize) -> Self {
122        self.cursor.stream_cursor = Some(stream_cursor);
123        self
124    }
125
126    /// Attach trace context.
127    #[must_use]
128    pub fn with_trace_context(mut self, trace_context: TraceContext) -> Self {
129        self.trace_context = trace_context;
130        self
131    }
132
133    /// Attach evidence metadata.
134    #[must_use]
135    pub fn with_metadata(mut self, metadata: Metadata) -> Self {
136        self.metadata = metadata;
137        self
138    }
139}
140
141/// Serializable checkpoint emitted at a durable execution boundary.
142#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
143pub struct AgentCheckpoint {
144    /// Checkpoint identifier.
145    pub checkpoint_id: CheckpointId,
146    /// Run identifier.
147    pub run_id: RunId,
148    /// Conversation identifier.
149    pub conversation_id: ConversationId,
150    /// Execution boundary.
151    pub node: AgentExecutionNode,
152    /// Completed run step at this boundary.
153    pub run_step: usize,
154    /// Stable resume evidence for durable services.
155    pub resume: AgentResumeEvidence,
156    /// Full checkpointable run state.
157    pub state: AgentRunState,
158    /// Boundary metadata for node-specific details.
159    #[serde(default, skip_serializing_if = "Metadata::is_empty")]
160    pub metadata: Metadata,
161}
162
163impl AgentCheckpoint {
164    /// Build a checkpoint from run state.
165    #[must_use]
166    pub fn new(node: AgentExecutionNode, state: &AgentRunState) -> Self {
167        Self {
168            checkpoint_id: CheckpointId::new(),
169            run_id: state.run_id.clone(),
170            conversation_id: state.conversation_id.clone(),
171            node,
172            run_step: state.run_step,
173            resume: AgentResumeEvidence::new(node, state),
174            state: state.clone(),
175            metadata: Metadata::default(),
176        }
177    }
178
179    /// Attach checkpoint metadata.
180    #[must_use]
181    pub fn with_metadata(mut self, metadata: Metadata) -> Self {
182        self.metadata = metadata;
183        self
184    }
185
186    /// Attach the last persisted stream cursor.
187    #[must_use]
188    pub fn with_stream_cursor(mut self, stream_cursor: usize) -> Self {
189        self.resume = self.resume.with_stream_cursor(stream_cursor);
190        self
191    }
192}
193
194/// Decision returned by an execution checkpoint handler.
195#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
196#[serde(tag = "kind", rename_all = "snake_case")]
197pub enum AgentExecutionDecision {
198    /// Continue executing the run.
199    Continue,
200    /// Suspend execution after persisting the checkpoint.
201    Suspend {
202        /// Human-readable suspend reason.
203        reason: String,
204    },
205}
206
207/// Executor failure.
208#[derive(Debug, Error)]
209pub enum AgentExecutorError {
210    /// Executor storage or policy failed.
211    #[error("executor failed: {0}")]
212    Failed(String),
213}
214
215/// Fine-grained executor hook for persistence, interruption, and durable scheduling.
216#[async_trait]
217pub trait AgentExecutor: Send + Sync {
218    /// Persist or inspect a checkpoint and decide whether execution should continue.
219    async fn checkpoint(
220        &self,
221        checkpoint: AgentCheckpoint,
222    ) -> Result<AgentExecutionDecision, AgentExecutorError>;
223}
224
225/// Shared executor reference.
226pub type DynAgentExecutor = Arc<dyn AgentExecutor>;
227
228/// Direct in-process executor that always continues.
229#[derive(Clone, Debug, Default)]
230pub struct DirectAgentExecutor;
231
232#[async_trait]
233impl AgentExecutor for DirectAgentExecutor {
234    async fn checkpoint(
235        &self,
236        _checkpoint: AgentCheckpoint,
237    ) -> Result<AgentExecutionDecision, AgentExecutorError> {
238        Ok(AgentExecutionDecision::Continue)
239    }
240}
starweaver_runtime/executor.rs

starweaver_runtime/
executor.rs
