starweaver_context/
checkpoint.rs

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

starweaver_context/
checkpoint.rs
