runifold_ops/
lib.rs

1//! Stable, read-only operational views over canonical Runifold artifacts.
2
3use std::collections::BTreeSet;
4
5use runifold_core::{
6    Budget, BudgetEvent, LifecycleEvent, RunError, RunEvent, RunEventKind, RunId, Usage,
7};
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10use thiserror::Error;
11
12/// Largest event page accepted by the stable operational query API.
13pub const MAX_EVENT_PAGE_SIZE: usize = 1_000;
14
15/// Exclusive position in one run's canonical event sequence.
16#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
17#[serde(transparent)]
18pub struct RunEventCursor(u64);
19
20impl RunEventCursor {
21    /// Creates a cursor positioned after `sequence`.
22    #[must_use]
23    pub const fn after(sequence: u64) -> Self {
24        Self(sequence)
25    }
26
27    /// Returns the last sequence already observed by the caller.
28    #[must_use]
29    pub const fn sequence(self) -> u64 {
30        self.0
31    }
32}
33
34/// Validated number of events requested from an operational source.
35#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
36#[serde(transparent)]
37pub struct RunEventPageSize(usize);
38
39impl RunEventPageSize {
40    /// Creates a bounded page size.
41    ///
42    /// # Errors
43    ///
44    /// Rejects zero and values above [`MAX_EVENT_PAGE_SIZE`].
45    pub const fn new(value: usize) -> Result<Self, RunEventQueryError> {
46        if value == 0 || value > MAX_EVENT_PAGE_SIZE {
47            Err(RunEventQueryError::InvalidPageSize { value })
48        } else {
49            Ok(Self(value))
50        }
51    }
52
53    /// Returns the validated page size.
54    #[must_use]
55    pub const fn get(self) -> usize {
56        self.0
57    }
58}
59
60/// One ordered page read from a durable canonical journal.
61#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
62pub struct RunEventPage {
63    /// Canonical events ordered by ascending sequence.
64    pub events: Vec<RunEvent>,
65    /// Cursor for the next page, absent when the source is exhausted.
66    pub next: Option<RunEventCursor>,
67}
68
69/// Stable read-only boundary implemented by durable journal adapters.
70pub trait RunEventSource: Send + Sync {
71    /// Reads canonical events after an exclusive sequence cursor.
72    ///
73    /// # Errors
74    ///
75    /// Returns a typed source error when storage access or decoding fails.
76    fn event_page(
77        &self,
78        run_id: RunId,
79        after: Option<RunEventCursor>,
80        limit: RunEventPageSize,
81    ) -> Result<RunEventPage, RunEventSourceError>;
82}
83
84/// Stable operational source failure category.
85#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
86#[serde(rename_all = "snake_case")]
87#[non_exhaustive]
88pub enum RunEventSourceErrorKind {
89    /// Durable storage could not complete the query.
90    Storage,
91    /// Persisted data was not a canonical `RunEvent`.
92    CorruptData,
93}
94
95/// Failure while reading canonical events from a durable source.
96#[derive(Clone, Debug, Error, Deserialize, Eq, PartialEq, Serialize)]
97#[error("run event source {kind:?}: {message}")]
98pub struct RunEventSourceError {
99    /// Stable error category suitable for application mapping.
100    pub kind: RunEventSourceErrorKind,
101    /// Redacted diagnostic message.
102    pub message: String,
103}
104
105impl RunEventSourceError {
106    /// Creates a storage failure.
107    #[must_use]
108    pub fn storage(message: impl Into<String>) -> Self {
109        Self {
110            kind: RunEventSourceErrorKind::Storage,
111            message: message.into(),
112        }
113    }
114
115    /// Creates a persisted-data decoding failure.
116    #[must_use]
117    pub fn corrupt_data(message: impl Into<String>) -> Self {
118        Self {
119            kind: RunEventSourceErrorKind::CorruptData,
120            message: message.into(),
121        }
122    }
123}
124
125/// Invalid operational event query.
126#[derive(Clone, Copy, Debug, Error, Eq, PartialEq)]
127#[non_exhaustive]
128pub enum RunEventQueryError {
129    /// The requested page size is outside the supported bounds.
130    #[error("event page size {value} must be between 1 and {MAX_EVENT_PAGE_SIZE}")]
131    InvalidPageSize {
132        /// Rejected page size.
133        value: usize,
134    },
135}
136
137/// Current terminal state inferred from canonical lifecycle events.
138#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
139#[serde(rename_all = "snake_case")]
140#[non_exhaustive]
141pub enum RunStatus {
142    /// A start exists without a terminal lifecycle event.
143    Running,
144    /// The run completed successfully.
145    Completed,
146    /// The run failed with a typed error.
147    Failed,
148    /// The run was cancelled.
149    Cancelled,
150}
151
152/// Read-only operational summary of one run event stream.
153#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
154pub struct RunInspection {
155    /// Inspection schema version.
156    pub schema_version: u32,
157    /// Inspected run identity.
158    pub run_id: RunId,
159    /// Inferred lifecycle state.
160    pub status: RunStatus,
161    /// Number of canonical events.
162    pub event_count: usize,
163    /// Last observed monotonic sequence.
164    pub last_sequence: u64,
165    /// Latest cumulative usage snapshot.
166    pub usage: Usage,
167    /// Terminal typed error, when the run failed.
168    pub error: Option<RunError>,
169    /// Domain event names in first-observed order.
170    pub domain_events: Vec<String>,
171}
172
173impl RunInspection {
174    /// Current inspection contract version.
175    pub const SCHEMA_VERSION: u32 = 1;
176
177    /// Validates and summarizes one canonical run history.
178    ///
179    /// # Errors
180    ///
181    /// Rejects empty, mixed-run, non-monotonic, causally invalid, or
182    /// multiply-terminal histories.
183    pub fn inspect(events: &[RunEvent]) -> Result<Self, InspectionError> {
184        let first = events.first().ok_or(InspectionError::Empty)?;
185        let run_id = first.meta.run_id;
186        let mut seen = BTreeSet::new();
187        let mut usage = Usage::default();
188        let mut terminal = None;
189        let mut error = None;
190        let mut domain_events = Vec::new();
191        for (index, event) in events.iter().enumerate() {
192            if event.meta.run_id != run_id {
193                return Err(InspectionError::MixedRun { index });
194            }
195            let expected = u64::try_from(index).unwrap_or(u64::MAX);
196            if event.meta.sequence != expected {
197                return Err(InspectionError::Sequence {
198                    index,
199                    expected,
200                    actual: event.meta.sequence,
201                });
202            }
203            if event
204                .meta
205                .caused_by
206                .is_some_and(|cause| !seen.contains(&cause))
207            {
208                return Err(InspectionError::UnknownCause { index });
209            }
210            seen.insert(event.meta.event_id);
211            match &event.kind {
212                RunEventKind::Budget(BudgetEvent::Updated { usage: current }) => usage = *current,
213                RunEventKind::Domain(domain) => {
214                    domain_events.push(format!("{}.{}", domain.namespace, domain.name));
215                }
216                RunEventKind::Lifecycle(LifecycleEvent::Completed { .. }) => {
217                    set_terminal(&mut terminal, RunStatus::Completed, index)?;
218                }
219                RunEventKind::Lifecycle(LifecycleEvent::Failed { error: failure }) => {
220                    set_terminal(&mut terminal, RunStatus::Failed, index)?;
221                    error = Some(failure.clone());
222                }
223                RunEventKind::Lifecycle(LifecycleEvent::Cancelled) => {
224                    set_terminal(&mut terminal, RunStatus::Cancelled, index)?;
225                }
226                _ => {}
227            }
228        }
229        Ok(Self {
230            schema_version: Self::SCHEMA_VERSION,
231            run_id,
232            status: terminal.unwrap_or(RunStatus::Running),
233            event_count: events.len(),
234            last_sequence: events.last().map_or(0, |event| event.meta.sequence),
235            usage,
236            error,
237            domain_events,
238        })
239    }
240}
241
242fn set_terminal(
243    terminal: &mut Option<RunStatus>,
244    status: RunStatus,
245    index: usize,
246) -> Result<(), InspectionError> {
247    if terminal.replace(status).is_some() {
248        Err(InspectionError::MultipleTerminalEvents { index })
249    } else {
250        Ok(())
251    }
252}
253
254/// One budget dimension and its remaining headroom.
255#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
256pub struct BudgetDimension {
257    /// Stable dimension name.
258    pub name: String,
259    /// Configured limit, or `None` when unbounded.
260    pub limit: Option<u64>,
261    /// Observed usage in the same unit.
262    pub used: u64,
263    /// Remaining capacity, or `None` when unbounded.
264    pub remaining: Option<u64>,
265    /// Whether observed usage exceeds the configured limit.
266    pub exceeded: bool,
267}
268
269/// Machine-readable explanation of budget consumption.
270#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
271pub struct BudgetExplanation {
272    /// One entry per canonical budget dimension.
273    pub dimensions: Vec<BudgetDimension>,
274}
275
276impl BudgetExplanation {
277    /// Computes remaining headroom without mutating runtime accounting.
278    #[must_use]
279    pub fn new(budget: Budget, usage: Usage) -> Self {
280        let duration_limit = budget
281            .duration
282            .map(|value| u64::try_from(value.as_micros()).unwrap_or(u64::MAX));
283        Self {
284            dimensions: vec![
285                dimension("tokens", budget.tokens, usage.tokens),
286                dimension("cost_microusd", budget.cost_microusd, usage.cost_microusd),
287                dimension("duration_micros", duration_limit, usage.duration_micros),
288                dimension("turns", budget.turns, usage.turns),
289                dimension("tool_calls", budget.tool_calls, usage.tool_calls),
290                dimension("delegations", budget.delegations, usage.delegations),
291            ],
292        }
293    }
294}
295
296fn dimension(name: &str, limit: Option<u64>, used: u64) -> BudgetDimension {
297    BudgetDimension {
298        name: name.into(),
299        limit,
300        used,
301        remaining: limit.map(|limit| limit.saturating_sub(used)),
302        exceeded: limit.is_some_and(|limit| used > limit),
303    }
304}
305
306/// Kind of structural JSON checkpoint change.
307#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
308#[serde(rename_all = "snake_case")]
309pub enum CheckpointChangeKind {
310    /// A path exists only in the newer checkpoint.
311    Added,
312    /// A path exists only in the older checkpoint.
313    Removed,
314    /// Both checkpoints contain a different scalar or container kind.
315    Changed,
316}
317
318/// One value-free checkpoint change safe for operator output.
319#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
320pub struct CheckpointChange {
321    /// JSON Pointer identifying the changed value.
322    pub path: String,
323    /// Structural change kind.
324    pub kind: CheckpointChangeKind,
325}
326
327/// Computes a bounded, value-free structural checkpoint diff.
328#[must_use]
329pub fn diff_checkpoints(before: &Value, after: &Value) -> Vec<CheckpointChange> {
330    let mut changes = Vec::new();
331    diff_at("", before, after, &mut changes);
332    changes.truncate(1_024);
333    changes
334}
335
336fn diff_at(path: &str, before: &Value, after: &Value, changes: &mut Vec<CheckpointChange>) {
337    if changes.len() >= 1_024 || before == after {
338        return;
339    }
340    match (before, after) {
341        (Value::Object(before), Value::Object(after)) => {
342            let keys = before.keys().chain(after.keys()).collect::<BTreeSet<_>>();
343            for key in keys {
344                let child = format!("{path}/{}", key.replace('~', "~0").replace('/', "~1"));
345                match (before.get(key), after.get(key)) {
346                    (Some(left), Some(right)) => diff_at(&child, left, right, changes),
347                    (None, Some(_)) => changes.push(CheckpointChange {
348                        path: child,
349                        kind: CheckpointChangeKind::Added,
350                    }),
351                    (Some(_), None) => changes.push(CheckpointChange {
352                        path: child,
353                        kind: CheckpointChangeKind::Removed,
354                    }),
355                    (None, None) => {}
356                }
357            }
358        }
359        _ => changes.push(CheckpointChange {
360            path: if path.is_empty() {
361                "/".into()
362            } else {
363                path.into()
364            },
365            kind: CheckpointChangeKind::Changed,
366        }),
367    }
368}
369
370/// Typed run-inspection failure.
371#[derive(Clone, Debug, Error, Eq, PartialEq)]
372#[non_exhaustive]
373pub enum InspectionError {
374    /// No events were supplied.
375    #[error("run history is empty")]
376    Empty,
377    /// An event belongs to another run.
378    #[error("event {index} belongs to a different run")]
379    MixedRun {
380        /// Zero-based event index.
381        index: usize,
382    },
383    /// Sequence numbers are not contiguous.
384    #[error("event {index} has sequence {actual}; expected {expected}")]
385    Sequence {
386        /// Zero-based event index.
387        index: usize,
388        /// Required canonical sequence.
389        expected: u64,
390        /// Observed sequence.
391        actual: u64,
392    },
393    /// A causal parent did not precede its event.
394    #[error("event {index} references an unknown or future cause")]
395    UnknownCause {
396        /// Zero-based event index.
397        index: usize,
398    },
399    /// More than one terminal lifecycle event was present.
400    #[error("event {index} adds a second terminal lifecycle state")]
401    MultipleTerminalEvents {
402        /// Zero-based event index.
403        index: usize,
404    },
405}
406
407#[cfg(test)]
408mod tests {
409    use runifold_core::{Budget, RunEvent, Usage};
410
411    use super::{
412        BudgetExplanation, CheckpointChangeKind, RunEventPageSize, RunEventQueryError,
413        RunInspection, RunStatus, diff_checkpoints,
414    };
415
416    #[test]
417    fn inspection_validates_and_summarizes_canonical_history() {
418        let scenario = runifold_test_fixture();
419        let inspection = RunInspection::inspect(&scenario).unwrap();
420
421        assert_eq!(inspection.status, RunStatus::Completed);
422        assert_eq!(inspection.event_count, 2);
423    }
424
425    #[test]
426    fn budget_explanation_is_saturating_and_marks_excess() {
427        let explanation = BudgetExplanation::new(
428            Budget {
429                tokens: Some(10),
430                ..Budget::default()
431            },
432            Usage {
433                tokens: 12,
434                ..Usage::default()
435            },
436        );
437        let tokens = &explanation.dimensions[0];
438        assert_eq!(tokens.remaining, Some(0));
439        assert!(tokens.exceeded);
440    }
441
442    #[test]
443    fn checkpoint_diff_reports_paths_without_values() {
444        let changes = diff_checkpoints(
445            &serde_json::json!({"secret": "old", "keep": 1}),
446            &serde_json::json!({"secret": "new", "add": true}),
447        );
448        assert!(changes.iter().any(|change| {
449            change.path == "/secret" && change.kind == CheckpointChangeKind::Changed
450        }));
451        assert!(!serde_json::to_string(&changes).unwrap().contains("old"));
452    }
453
454    #[test]
455    fn event_page_size_enforces_public_query_bounds() {
456        assert_eq!(RunEventPageSize::new(1).unwrap().get(), 1);
457        assert!(matches!(
458            RunEventPageSize::new(0),
459            Err(RunEventQueryError::InvalidPageSize { value: 0 })
460        ));
461        assert!(RunEventPageSize::new(1_001).is_err());
462    }
463
464    fn runifold_test_fixture() -> Vec<RunEvent> {
465        use runifold_core::{
466            BudgetTracker, CapabilitySet, EventFactory, LifecycleEvent, RunContext, RunEventKind,
467        };
468        let run = RunContext::root(BudgetTracker::new(Budget::default()), CapabilitySet::new());
469        let factory = EventFactory::new(run.run_id(), None);
470        let started = factory.emit(RunEventKind::Lifecycle(LifecycleEvent::Started), None);
471        let completed = factory.emit(
472            RunEventKind::Lifecycle(LifecycleEvent::Completed {
473                output: serde_json::json!({"ok": true}),
474            }),
475            Some(started.meta.event_id),
476        );
477        vec![started, completed]
478    }
479}
runifold_ops/lib.rs

runifold_ops/
lib.rs
