winx_code_agent/state/execution_history/
observation.rs

1//! Bounded, independently resumable observations from one catalog transaction.
2
3use schemars::JsonSchema;
4use serde::Deserialize;
5
6use super::{
7    paging, storage, validate_id, ExecutionCatalog, ExecutionRecord, ExecutionScope, HistoryError,
8    HistoryResult, OutputPage, OutputSource, MAX_PAGE_BYTES, MAX_WATCHED_EXECUTIONS,
9};
10
11pub const DEFAULT_OBSERVATION_PAGE_BYTES: usize = 2 * 1024;
12pub const MAX_OBSERVATION_ITEMS: usize = MAX_WATCHED_EXECUTIONS;
13pub const MAX_OBSERVATION_OUTPUT_BYTES: usize = MAX_PAGE_BYTES;
14
15#[derive(Clone, Debug, Deserialize, JsonSchema)]
16pub struct ObservationRequest {
17    /// Execution retained for the initialized principal, thread and workspace.
18    #[schemars(length(min = 1, max = 512))]
19    pub execution_id: String,
20    #[serde(default)]
21    pub source: OutputSource,
22    /// An opaque Read/Observe cursor for this same execution and source.
23    #[serde(default)]
24    pub cursor: Option<String>,
25    /// Maximum UTF-8 text bytes for this page; all items together allow 64 KiB.
26    #[serde(default = "default_page_bytes")]
27    #[schemars(range(min = 4, max = 65536))]
28    pub page_bytes: usize,
29}
30
31const fn default_page_bytes() -> usize {
32    DEFAULT_OBSERVATION_PAGE_BYTES
33}
34
35/// Metadata remains available when its output page cannot be read.
36#[derive(Debug)]
37pub struct ObservationItem {
38    pub execution_id: String,
39    pub source: OutputSource,
40    pub record: Option<ExecutionRecord>,
41    pub page: HistoryResult<OutputPage>,
42}
43
44/// The cursor is a catalog change baseline, independent of every page cursor.
45#[derive(Debug)]
46pub struct CatalogObservation {
47    pub sequence: u64,
48    pub cursor: String,
49    pub items: Vec<ObservationItem>,
50}
51
52/// Validate resource admission before session, catalog or runtime I/O.
53///
54/// A well-formed missing execution or invalid page cursor is an item error;
55/// malformed identities, duplicate streams and invalid budgets reject the call.
56pub fn validate_observation_requests(items: &[ObservationRequest]) -> HistoryResult<()> {
57    if items.is_empty() || items.len() > MAX_OBSERVATION_ITEMS {
58        return Err(HistoryError::InvalidInput(format!(
59            "observe requires between 1 and {MAX_OBSERVATION_ITEMS} items"
60        )));
61    }
62    let mut total_page_bytes = 0;
63    for (index, item) in items.iter().enumerate() {
64        validate_id(&item.execution_id)?;
65        if !(4..=MAX_PAGE_BYTES).contains(&item.page_bytes) {
66            return Err(HistoryError::InvalidInput(format!(
67                "observe page_bytes must be between 4 and {MAX_PAGE_BYTES}"
68            )));
69        }
70        // Both the item count and each summand are bounded before addition.
71        total_page_bytes += item.page_bytes;
72        if total_page_bytes > MAX_OBSERVATION_OUTPUT_BYTES {
73            return Err(HistoryError::InvalidInput(format!(
74                "observe total page_bytes must not exceed {MAX_OBSERVATION_OUTPUT_BYTES}"
75            )));
76        }
77        if items[..index].iter().any(|previous| {
78            previous.execution_id == item.execution_id && previous.source == item.source
79        }) {
80            return Err(HistoryError::InvalidInput(
81                "observe requires unique execution_id and source pairs".into(),
82            ));
83        }
84    }
85    Ok(())
86}
87
88impl ExecutionCatalog {
89    /// Read retained metadata and bounded pages under one scoped catalog lock.
90    ///
91    /// This performs no runtime refresh and consumes no delivery cursor. Item
92    /// failures preserve their metadata and do not discard successful siblings.
93    /// The outer error reports a shared catalog or request-admission failure.
94    pub fn observe(
95        &self,
96        scope: &ExecutionScope,
97        items: &[ObservationRequest],
98        now: u64,
99    ) -> HistoryResult<CatalogObservation> {
100        validate_observation_requests(items)?;
101        storage::transaction(self, scope, now, |transaction| {
102            let items = items
103                .iter()
104                .map(|item| ObservationItem {
105                    execution_id: item.execution_id.clone(),
106                    source: item.source,
107                    record: transaction.find(&item.execution_id).cloned(),
108                    page: paging::read(
109                        transaction,
110                        &item.execution_id,
111                        item.source,
112                        item.cursor.as_deref(),
113                        item.page_bytes,
114                    ),
115                })
116                .collect();
117            Ok(CatalogObservation {
118                sequence: transaction.index.sequence,
119                cursor: paging::event_cursor(transaction)?,
120                items,
121            })
122        })
123    }
124}
winx_code_agent/state/execution_history/observation.rs

winx_code_agent/state/execution_history/
observation.rs
