vv_agent/
checkpoint.rs

1//! Public building blocks for the current durable checkpoint protocol.
2//!
3//! This module owns the language-neutral values and canonical JSON helpers used
4//! by definitions, journals, and event identities. The wire discriminator is
5//! validated strictly; no historical checkpoint decoder is retained.
6
7use std::cmp::Ordering;
8use std::collections::{BTreeMap, BTreeSet};
9use std::fmt;
10use std::sync::{Arc, Mutex};
11
12use serde::{Deserialize, Serialize};
13use serde_json::{Map, Number, Value};
14use sha2::{Digest, Sha256};
15
16use crate::runtime::backends::CapabilityRef;
17use crate::runtime::state::CheckpointStore;
18
19mod canonical;
20
21pub use canonical::*;
22use canonical::{
23    require_non_empty, require_positive, utf16_cmp, validate_capability_ref,
24    validate_capability_slot, validate_i_json, validate_pointer,
25};
26
27pub const MAX_WIRE_INTEGER: u64 = (1_u64 << 53) - 1;
28pub const MAX_CHECKPOINT_KEY_BYTES: usize = 512;
29pub const MAX_EXTENSION_NAMESPACE_BYTES: usize = 128;
30pub const MAX_EXTENSION_ENTRY_BYTES: usize = 65_536;
31pub const DEFAULT_MAX_EXTENSION_STATE_BYTES: u64 = 262_144;
32pub const CHECKPOINT_SCHEMA: &str = "vv-agent.checkpoint.v3";
33pub const RUN_DEFINITION_SCHEMA: &str = "vv-agent.run-definition.v2";
34pub const OPERATION_REQUEST_SCHEMA: &str = "vv-agent.operation-request.v1";
35pub const EVENT_CURSOR_SCHEMA: &str = "vv-agent.event-cursor.v1";
36pub const CREDENTIAL_REDACTED: &str = "<credential-redacted>";
37
38/// A stable, observable checkpoint error.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub struct CheckpointError {
41    code: String,
42    message: String,
43}
44
45impl CheckpointError {
46    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
47        Self {
48            code: code.into(),
49            message: message.into(),
50        }
51    }
52
53    pub fn code(&self) -> &str {
54        &self.code
55    }
56
57    pub fn error_code(&self) -> &str {
58        self.code()
59    }
60
61    pub fn message(&self) -> &str {
62        &self.message
63    }
64}
65
66impl fmt::Display for CheckpointError {
67    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
68        write!(formatter, "{}: {}", self.code, self.message)
69    }
70}
71
72impl std::error::Error for CheckpointError {}
73
74impl From<serde_json::Error> for CheckpointError {
75    fn from(error: serde_json::Error) -> Self {
76        Self::new("checkpoint_json_invalid", error.to_string())
77    }
78}
79
80impl From<std::io::Error> for CheckpointError {
81    fn from(error: std::io::Error) -> Self {
82        Self::new("checkpoint_store_io", error.to_string())
83    }
84}
85
86pub type CheckpointResult<T> = Result<T, CheckpointError>;
87
88#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
89#[serde(rename_all = "snake_case")]
90pub enum ResumePolicy {
91    #[default]
92    New,
93    ResumeIfPresent,
94    RequireExisting,
95}
96
97#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
98#[serde(rename_all = "snake_case")]
99pub enum AmbiguousModelPolicy {
100    #[default]
101    RequireReconciliation,
102    RetryWithDuplicateRisk,
103}
104
105#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
106#[serde(rename_all = "snake_case")]
107pub enum AmbiguousToolPolicy {
108    #[default]
109    RequireReconciliation,
110    RetryIdempotentOnly,
111}
112
113#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
114#[serde(rename_all = "snake_case")]
115pub enum ToolIdempotency {
116    Supported,
117    Unsupported,
118    #[default]
119    Unknown,
120}
121
122#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
123#[serde(rename_all = "snake_case")]
124pub enum OperationKind {
125    Model,
126    Tool,
127}
128
129#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
130#[serde(rename_all = "snake_case")]
131pub enum OperationState {
132    Planned,
133    Started,
134    Succeeded,
135    Failed,
136    Ambiguous,
137}
138
139#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
140#[serde(rename_all = "snake_case")]
141pub enum ClaimMode {
142    Continue,
143    Recovery,
144}
145
146#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
147#[serde(rename_all = "snake_case")]
148pub enum CheckpointStatus {
149    Pending,
150    Running,
151    WaitUser,
152    Completed,
153    Failed,
154    MaxCycles,
155    ReconciliationRequired,
156}
157
158impl CheckpointStatus {
159    pub fn as_str(self) -> &'static str {
160        match self {
161            Self::Pending => "pending",
162            Self::Running => "running",
163            Self::WaitUser => "wait_user",
164            Self::Completed => "completed",
165            Self::Failed => "failed",
166            Self::MaxCycles => "max_cycles",
167            Self::ReconciliationRequired => "reconciliation_required",
168        }
169    }
170
171    pub fn is_terminal(self) -> bool {
172        matches!(
173            self,
174            Self::WaitUser | Self::Completed | Self::Failed | Self::MaxCycles
175        )
176    }
177}
178
179#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
180#[serde(rename_all = "snake_case")]
181pub enum ReconciliationDecisionKind {
182    Defer,
183    Retry,
184    ReplaySuccess,
185    RecordFailure,
186    Abort,
187}
188
189/// A typed observation retained when a started external operation has no
190/// durable receipt.
191#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
192pub struct ResumeObservation {
193    pub operation_id: String,
194    pub operation_kind: OperationKind,
195    pub cycle_index: u64,
196    #[serde(default = "ambiguous_state")]
197    pub state: OperationState,
198    pub risk: String,
199    pub idempotency_support: Option<ToolIdempotency>,
200}
201
202fn ambiguous_state() -> OperationState {
203    OperationState::Ambiguous
204}
205
206impl ResumeObservation {
207    pub fn validate(&self) -> CheckpointResult<()> {
208        require_non_empty(&self.operation_id, "resume observation operation_id")?;
209        require_positive(self.cycle_index, "resume observation cycle_index")?;
210        if self.state != OperationState::Ambiguous {
211            return Err(CheckpointError::new(
212                "checkpoint_resume_observation_invalid",
213                "resume observation state must be ambiguous",
214            ));
215        }
216        require_non_empty(&self.risk, "resume observation risk")
217    }
218}
219
220#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
221pub struct ReconciliationError {
222    pub code: String,
223    pub message: String,
224    #[serde(default)]
225    pub retryable: bool,
226}
227
228impl ReconciliationError {
229    pub fn new(code: impl Into<String>, message: impl Into<String>, retryable: bool) -> Self {
230        Self {
231            code: code.into(),
232            message: message.into(),
233            retryable,
234        }
235    }
236
237    pub fn validate(&self) -> CheckpointResult<()> {
238        require_non_empty(&self.code, "reconciliation error code")?;
239        require_non_empty(&self.message, "reconciliation error message")
240    }
241}
242
243#[derive(Debug, Clone, PartialEq)]
244pub struct ReconciliationDecision {
245    pub kind: ReconciliationDecisionKind,
246    pub response: Option<Value>,
247    pub result: Option<Value>,
248    pub error: Option<ReconciliationError>,
249}
250
251impl ReconciliationDecision {
252    pub fn defer() -> Self {
253        Self {
254            kind: ReconciliationDecisionKind::Defer,
255            response: None,
256            result: None,
257            error: None,
258        }
259    }
260
261    pub fn retry() -> Self {
262        Self {
263            kind: ReconciliationDecisionKind::Retry,
264            response: None,
265            result: None,
266            error: None,
267        }
268    }
269
270    pub fn replay_response(response: Value) -> Self {
271        Self {
272            kind: ReconciliationDecisionKind::ReplaySuccess,
273            response: Some(response),
274            result: None,
275            error: None,
276        }
277    }
278
279    pub fn replay_result(result: Value) -> Self {
280        Self {
281            kind: ReconciliationDecisionKind::ReplaySuccess,
282            response: None,
283            result: Some(result),
284            error: None,
285        }
286    }
287
288    pub fn record_failure(error: ReconciliationError) -> Self {
289        Self {
290            kind: ReconciliationDecisionKind::RecordFailure,
291            response: None,
292            result: None,
293            error: Some(error),
294        }
295    }
296
297    pub fn abort(error: ReconciliationError) -> Self {
298        Self {
299            kind: ReconciliationDecisionKind::Abort,
300            response: None,
301            result: None,
302            error: Some(error),
303        }
304    }
305
306    pub fn validate(&self) -> CheckpointResult<()> {
307        match self.kind {
308            ReconciliationDecisionKind::ReplaySuccess
309                if self.response.is_some() ^ self.result.is_some() => {}
310            ReconciliationDecisionKind::ReplaySuccess => {
311                return Err(CheckpointError::new(
312                    "checkpoint_reconciliation_decision_invalid",
313                    "replay_success requires exactly one response or result",
314                ));
315            }
316            ReconciliationDecisionKind::RecordFailure | ReconciliationDecisionKind::Abort => {
317                let Some(error) = &self.error else {
318                    return Err(CheckpointError::new(
319                        "checkpoint_reconciliation_decision_invalid",
320                        "failure and abort decisions require a typed error",
321                    ));
322                };
323                error.validate()?;
324            }
325            _ if self.response.is_none() && self.result.is_none() && self.error.is_none() => {}
326            _ => {
327                return Err(CheckpointError::new(
328                    "checkpoint_reconciliation_decision_invalid",
329                    "this reconciliation decision carries an unexpected payload",
330                ));
331            }
332        }
333        if let Some(response) = &self.response {
334            validate_i_json(response, "reconciliation response")?;
335        }
336        if let Some(result) = &self.result {
337            validate_i_json(result, "reconciliation result")?;
338        }
339        Ok(())
340    }
341}
342
343pub trait ReconciliationProvider: Send + Sync {
344    fn reconcile(
345        &self,
346        observation: &ResumeObservation,
347    ) -> CheckpointResult<ReconciliationDecision>;
348}
349
350pub trait CheckpointExtension: Send + Sync {
351    fn namespace(&self) -> &str;
352    fn version(&self) -> &str;
353    fn required(&self) -> bool;
354    fn snapshot(&self) -> CheckpointResult<Value>;
355    fn restore(&self, state: &Value) -> CheckpointResult<()>;
356}
357
358/// Runtime configuration for enabling checkpoint v2.
359#[derive(Clone)]
360pub struct CheckpointConfig {
361    pub store: Option<Arc<dyn CheckpointStore>>,
362    pub store_ref: Option<CapabilityRef>,
363    pub key: Option<String>,
364    pub resume_policy: ResumePolicy,
365    pub ambiguous_model_policy: AmbiguousModelPolicy,
366    pub ambiguous_tool_policy: AmbiguousToolPolicy,
367    pub required_extension_namespaces: Vec<String>,
368    pub max_extension_state_bytes: u64,
369    pub credential_slots: Vec<String>,
370    pub capability_refs: BTreeMap<String, CapabilityRef>,
371}
372
373impl Default for CheckpointConfig {
374    fn default() -> Self {
375        Self {
376            store: None,
377            store_ref: None,
378            key: None,
379            resume_policy: ResumePolicy::New,
380            ambiguous_model_policy: AmbiguousModelPolicy::RequireReconciliation,
381            ambiguous_tool_policy: AmbiguousToolPolicy::RequireReconciliation,
382            required_extension_namespaces: Vec::new(),
383            max_extension_state_bytes: DEFAULT_MAX_EXTENSION_STATE_BYTES,
384            credential_slots: Vec::new(),
385            capability_refs: BTreeMap::new(),
386        }
387    }
388}
389
390impl fmt::Debug for CheckpointConfig {
391    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
392        formatter
393            .debug_struct("CheckpointConfig")
394            .field("store", &self.store.as_ref().map(|_| "configured"))
395            .field("store_ref", &self.store_ref)
396            .field("key", &self.key)
397            .field("resume_policy", &self.resume_policy)
398            .field("ambiguous_model_policy", &self.ambiguous_model_policy)
399            .field("ambiguous_tool_policy", &self.ambiguous_tool_policy)
400            .field(
401                "required_extension_namespaces",
402                &self.required_extension_namespaces,
403            )
404            .field("max_extension_state_bytes", &self.max_extension_state_bytes)
405            .field("credential_slots", &self.credential_slots)
406            .field("capability_refs", &self.capability_refs)
407            .finish()
408    }
409}
410
411impl CheckpointConfig {
412    pub fn new(store: Arc<dyn CheckpointStore>) -> Self {
413        Self {
414            store: Some(store),
415            ..Self::default()
416        }
417    }
418
419    pub fn with_store<S>(store: S) -> Self
420    where
421        S: CheckpointStore + 'static,
422    {
423        Self::new(Arc::new(store))
424    }
425
426    pub fn validate(&self) -> CheckpointResult<()> {
427        if self.store.is_some() == self.store_ref.is_some() {
428            return Err(CheckpointError::new(
429                "checkpoint_store_selection_invalid",
430                "exactly one of store or store_ref is required",
431            ));
432        }
433        if let Some(reference) = &self.store_ref {
434            validate_capability_ref(reference, "CheckpointConfig.store_ref")?;
435        }
436        match (&self.key, self.resume_policy) {
437            (None, ResumePolicy::New) => {}
438            (None, _) => {
439                return Err(CheckpointError::new(
440                    "checkpoint_key_required",
441                    "resume_if_present and require_existing need an explicit key",
442                ));
443            }
444            (Some(key), _) => validate_checkpoint_key(key)?,
445        }
446        if self.max_extension_state_bytes > MAX_WIRE_INTEGER {
447            return Err(CheckpointError::new(
448                "checkpoint_extension_limit_invalid",
449                "max_extension_state_bytes exceeds the JSON-safe integer range",
450            ));
451        }
452        let mut previous_slot: Option<&str> = None;
453        for slot in &self.credential_slots {
454            validate_pointer(slot).map_err(|error| {
455                CheckpointError::new("checkpoint_credential_slots_invalid", error.message)
456            })?;
457            if previous_slot.is_some_and(|previous| utf16_cmp(previous, slot) != Ordering::Less) {
458                return Err(CheckpointError::new(
459                    "checkpoint_credential_slots_invalid",
460                    "credential slots must be sorted and unique",
461                ));
462            }
463            previous_slot = Some(slot);
464        }
465        let mut seen = BTreeSet::new();
466        for namespace in &self.required_extension_namespaces {
467            validate_extension_namespace(namespace)?;
468            if !seen.insert(namespace) {
469                return Err(CheckpointError::new(
470                    "checkpoint_extension_namespace_duplicate",
471                    format!("duplicate extension namespace {namespace}"),
472                ));
473            }
474        }
475        if self
476            .required_extension_namespaces
477            .windows(2)
478            .any(|window| window[0] > window[1])
479        {
480            return Err(CheckpointError::new(
481                "checkpoint_extension_namespace_invalid",
482                "required extension namespaces must be sorted",
483            ));
484        }
485        for (slot, reference) in &self.capability_refs {
486            validate_capability_slot(slot)?;
487            validate_capability_ref(reference, &format!("capability_refs.{slot}"))?;
488        }
489        Ok(())
490    }
491
492    pub fn capability_ref(&self, slot: &str) -> Option<&CapabilityRef> {
493        self.capability_refs.get(slot)
494    }
495}
496
497#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
498pub struct EventCursor {
499    pub schema_version: String,
500    pub store_ref: CapabilityRef,
501    pub value: Value,
502    pub last_event_id: Option<String>,
503}
504
505impl EventCursor {
506    pub fn new(store_ref: CapabilityRef, value: Value, last_event_id: Option<String>) -> Self {
507        Self {
508            schema_version: EVENT_CURSOR_SCHEMA.to_string(),
509            store_ref,
510            value,
511            last_event_id,
512        }
513    }
514
515    pub fn validate(&self) -> CheckpointResult<()> {
516        if self.schema_version != EVENT_CURSOR_SCHEMA {
517            return Err(CheckpointError::new(
518                "checkpoint_event_cursor_schema_unsupported",
519                "unsupported event cursor schema_version",
520            ));
521        }
522        validate_capability_ref(&self.store_ref, "event cursor store_ref")?;
523        validate_i_json(&self.value, "event cursor value")?;
524        if self
525            .last_event_id
526            .as_ref()
527            .is_some_and(|event_id| event_id.trim().is_empty())
528        {
529            return Err(CheckpointError::new(
530                "checkpoint_event_cursor_invalid",
531                "last_event_id must be non-empty when present",
532            ));
533        }
534        Ok(())
535    }
536}
537
538#[derive(Debug, Clone, PartialEq)]
539pub struct AppendOnceResult {
540    pub inserted: bool,
541    pub cursor: Value,
542}
543
544pub trait IdempotentRunEventStore: Send + Sync {
545    fn append_once(
546        &self,
547        event_id: &str,
548        payload_digest: &str,
549        event: &Value,
550    ) -> CheckpointResult<AppendOnceResult>;
551}
552
553/// A process-local accepted-once event store useful for deterministic tests
554/// and hosts that already own their process lifetime.
555#[derive(Debug, Default, Clone)]
556pub struct InMemoryRunEventStore {
557    entries: Arc<Mutex<RunEventEntries>>,
558    next_sequence: Arc<Mutex<u64>>,
559}
560
561type RunEventEntries = BTreeMap<String, (String, Value, Value)>;
562
563impl IdempotentRunEventStore for InMemoryRunEventStore {
564    fn append_once(
565        &self,
566        event_id: &str,
567        payload_digest: &str,
568        event: &Value,
569    ) -> CheckpointResult<AppendOnceResult> {
570        require_non_empty(event_id, "event_id")?;
571        validate_sha256(payload_digest, "payload_digest")?;
572        if !event.is_object() {
573            return Err(CheckpointError::new(
574                "event_payload_invalid",
575                "event payload must be an object",
576            ));
577        }
578        let calculated = event_payload_digest(event)?;
579        if calculated != payload_digest {
580            return Err(CheckpointError::new(
581                "event_payload_digest_mismatch",
582                "payload_digest does not match the canonical event",
583            ));
584        }
585        let mut entries = self.entries.lock().map_err(|_| {
586            CheckpointError::new(
587                "checkpoint_store_lock_poisoned",
588                "event store lock poisoned",
589            )
590        })?;
591        if let Some((existing_digest, _existing_event, cursor)) = entries.get(event_id) {
592            if existing_digest != payload_digest {
593                return Err(CheckpointError::new(
594                    "event_identity_conflict",
595                    format!("event id {event_id} was recorded with a different digest"),
596                ));
597            }
598            return Ok(AppendOnceResult {
599                inserted: false,
600                cursor: cursor.clone(),
601            });
602        }
603        let mut next = self.next_sequence.lock().map_err(|_| {
604            CheckpointError::new(
605                "checkpoint_store_lock_poisoned",
606                "event sequence lock poisoned",
607            )
608        })?;
609        *next = next.checked_add(1).ok_or_else(|| {
610            CheckpointError::new("event_cursor_overflow", "event sequence overflow")
611        })?;
612        let cursor = serde_json::json!({"sequence": *next});
613        entries.insert(
614            event_id.to_string(),
615            (payload_digest.to_string(), event.clone(), cursor.clone()),
616        );
617        Ok(AppendOnceResult {
618            inserted: true,
619            cursor,
620        })
621    }
622}
623
624impl crate::event_store::RunEventStore for InMemoryRunEventStore {
625    fn append(
626        &self,
627        event: &crate::events::RunEvent,
628    ) -> Result<(), crate::event_store::EventStoreError> {
629        let value = serde_json::to_value(event).map_err(|error| {
630            crate::event_store::EventStoreError::new(
631                "event_store_serialization_error",
632                error.to_string(),
633            )
634        })?;
635        let digest = event_payload_digest(&value).map_err(|error| {
636            crate::event_store::EventStoreError::new(
637                "event_store_checkpoint_error",
638                error.to_string(),
639            )
640        })?;
641        IdempotentRunEventStore::append_once(self, event.event_id().as_str(), &digest, &value)
642            .map_err(|error| {
643                crate::event_store::EventStoreError::new(
644                    "event_store_checkpoint_error",
645                    error.to_string(),
646                )
647            })?;
648        Ok(())
649    }
650
651    fn replay(
652        &self,
653        query: crate::event_store::RunEventReplayQuery,
654    ) -> Result<crate::event_store::RunEventIter, crate::event_store::EventStoreError> {
655        let entries = self.entries.lock().map_err(|_| {
656            crate::event_store::EventStoreError::new(
657                "event_store_lock_poisoned",
658                "event store lock poisoned",
659            )
660        })?;
661        let mut events = entries
662            .values()
663            .filter_map(|(_, event, cursor)| {
664                let sequence = cursor.get("sequence").and_then(Value::as_u64)?;
665                let event =
666                    serde_json::from_value::<crate::events::RunEvent>(event.clone()).ok()?;
667                let include = event.run_id() == query.run_id()
668                    || (query.should_include_children()
669                        && event.parent_run_id() == Some(query.run_id()));
670                include.then_some((sequence, event))
671            })
672            .collect::<Vec<_>>();
673        events.sort_by_key(|(sequence, _)| *sequence);
674        Ok(Box::new(events.into_iter().map(|(_, event)| Ok(event))))
675    }
676
677    fn append_once(
678        &self,
679        event_id: &str,
680        payload_digest: &str,
681        event: &crate::events::RunEvent,
682    ) -> Result<Option<EventCursor>, crate::event_store::EventStoreError> {
683        let value = serde_json::to_value(event).map_err(|error| {
684            crate::event_store::EventStoreError::new(
685                "event_store_serialization_error",
686                error.to_string(),
687            )
688        })?;
689        let result = IdempotentRunEventStore::append_once(self, event_id, payload_digest, &value)
690            .map_err(|error| {
691            crate::event_store::EventStoreError::new(
692                "event_store_checkpoint_error",
693                error.to_string(),
694            )
695        })?;
696        Ok(Some(EventCursor::new(
697            crate::runtime::backends::CapabilityRef {
698                id: "events.in-memory".to_string(),
699                version: "1".to_string(),
700            },
701            result.cursor,
702            Some(event_id.to_string()),
703        )))
704    }
705}
vv_agent/checkpoint.rs

vv_agent/
checkpoint.rs
