winx_code_agent/state/
execution_history.rs

1//! Bounded, session-scoped execution history independent of MCP Task retention.
2//!
3//! Runtime identities and process states are authoritative. Stored output is a
4//! projection for reading; it never determines completion, ownership or routing.
5
6mod observation;
7mod paging;
8mod storage;
9mod update;
10
11pub use observation::{
12    validate_observation_requests, CatalogObservation, ObservationItem, ObservationRequest,
13    DEFAULT_OBSERVATION_PAGE_BYTES, MAX_OBSERVATION_ITEMS, MAX_OBSERVATION_OUTPUT_BYTES,
14};
15
16use std::path::PathBuf;
17
18use schemars::JsonSchema;
19use serde::{Deserialize, Serialize};
20
21use crate::runtime::workflow::PromptReceipt;
22use crate::runtime::ShellExecutionToken;
23use crate::state::turn::{TurnExplanation, TurnState};
24use crate::tools::bash_command::BashCommandState;
25
26pub const DEFAULT_PAGE_BYTES: usize = 16 * 1024;
27pub const MAX_PAGE_BYTES: usize = 64 * 1024;
28pub const MAX_WATCHED_EXECUTIONS: usize = 32;
29
30pub type HistoryResult<T> = Result<T, HistoryError>;
31
32#[derive(Debug, thiserror::Error)]
33pub enum HistoryError {
34    #[error("execution history I/O failed: {0}")]
35    Io(#[from] std::io::Error),
36    #[error("execution history data is invalid: {0}")]
37    Json(#[from] serde_json::Error),
38    #[error("invalid execution history request: {0}")]
39    InvalidInput(String),
40    #[error("invalid or foreign execution cursor")]
41    InvalidCursor,
42    #[error("execution was not found in this session or its retention expired")]
43    NotFound,
44    #[error("execution history capacity is occupied by active executions")]
45    Capacity,
46    #[error("execution identity conflicts with its retained record")]
47    IdentityConflict,
48}
49
50#[derive(Clone, Debug)]
51pub struct HistoryLimits {
52    pub max_records: usize,
53    pub retention_ms: u64,
54    pub max_output_bytes: usize,
55    pub max_session_output_bytes: usize,
56}
57
58impl Default for HistoryLimits {
59    fn default() -> Self {
60        Self {
61            max_records: 128,
62            retention_ms: 24 * 60 * 60 * 1000,
63            max_output_bytes: 1024 * 1024,
64            max_session_output_bytes: 16 * 1024 * 1024,
65        }
66    }
67}
68
69/// The thread ID is already authenticated and internally scoped by the server.
70#[derive(Clone, Debug)]
71pub struct ExecutionScope {
72    pub thread_id: String,
73    pub workspace_root: PathBuf,
74}
75
76#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
77#[serde(rename_all = "snake_case")]
78pub enum OutputSource {
79    #[default]
80    Output,
81    Screen,
82    Detection,
83}
84
85#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
86#[serde(rename_all = "snake_case")]
87pub enum ExecutionProcessStatus {
88    Running,
89    Exited,
90    Unknown,
91}
92
93impl ExecutionProcessStatus {
94    pub const fn as_str(self) -> &'static str {
95        match self {
96            Self::Running => "running",
97            Self::Exited => "exited",
98            Self::Unknown => "unknown",
99        }
100    }
101}
102
103/// Output observed without changing the legacy delivery cursor.
104#[derive(Clone, Debug)]
105pub enum CapturedOutput {
106    /// A retained runtime snapshot; truncation is supplied separately.
107    Snapshot(String),
108    /// A delivered fragment from a runtime without workflow observation.
109    Fragment(String),
110    /// Metadata-only inspection must never replace stored output with empty text.
111    None,
112}
113
114#[derive(Clone, Debug)]
115pub struct ExecutionCapture {
116    pub execution_id: String,
117    pub execution_token: Option<ShellExecutionToken>,
118    pub command_generation: Option<u64>,
119    pub state: BashCommandState,
120    pub output: CapturedOutput,
121    pub screen: Option<String>,
122    pub explanation: Option<TurnExplanation>,
123    pub prompt_receipt: Option<PromptReceipt>,
124    pub activity_seq: Option<u64>,
125    pub observed_at_unix_ms: u64,
126    pub label: Option<String>,
127    pub output_truncated: bool,
128    pub dropped_output_file: Option<PathBuf>,
129}
130
131// These independent flags are part of the persisted and public metadata contract.
132#[expect(clippy::struct_excessive_bools, reason = "independent wire metadata flags")]
133#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
134pub struct ExecutionRecord {
135    pub execution_id: String,
136    pub execution_token: Option<ShellExecutionToken>,
137    pub command_generation: Option<u64>,
138    pub background_id: Option<String>,
139    pub cwd: PathBuf,
140    pub label: Option<String>,
141    pub process_status: ExecutionProcessStatus,
142    pub exit_code: Option<i32>,
143    pub turn_state: Option<TurnState>,
144    pub explanation: Option<TurnExplanation>,
145    pub prompt_receipt: Option<PromptReceipt>,
146    pub first_observed_at_unix_ms: u64,
147    pub observed_at_unix_ms: u64,
148    pub finished_at_unix_ms: Option<u64>,
149    pub expires_at_unix_ms: Option<u64>,
150    pub activity_seq: Option<u64>,
151    /// Activity covered by the last full retained-output snapshot.
152    #[serde(default)]
153    pub output_activity_seq: Option<u64>,
154    /// Metadata advanced without a complete output capture; a bound retry is needed.
155    #[serde(default = "default_output_pending")]
156    pub output_pending: bool,
157    pub sequence: u64,
158    pub state_sequence: u64,
159    pub output_sequence: u64,
160    pub output_revision: u64,
161    pub retained_start: u64,
162    pub retained_end: u64,
163    pub retained_bytes: usize,
164    pub output_truncated: bool,
165    pub runtime_output_truncated: bool,
166    pub screen_truncated: bool,
167    pub screen_retained_bytes: usize,
168    pub output_coverage: String,
169    pub dropped_output_file: Option<PathBuf>,
170    pub unknown_reason: Option<String>,
171}
172
173#[derive(Clone, Debug, Serialize)]
174pub struct CatalogSnapshot {
175    pub sequence: u64,
176    pub cursor: String,
177    pub executions: Vec<ExecutionRecord>,
178}
179
180#[derive(Clone, Debug, Serialize)]
181pub struct HistoryChanges {
182    pub sequence: u64,
183    pub after_sequence: u64,
184    pub cursor: String,
185    pub executions: Vec<ExecutionRecord>,
186    pub missing_ids: Vec<String>,
187    pub resync_required: bool,
188}
189
190// Pagination continuity and retention are independent properties of this wire response.
191#[expect(clippy::struct_excessive_bools, reason = "independent pagination metadata flags")]
192#[derive(Clone, Debug, Serialize)]
193pub struct OutputPage {
194    pub execution_id: String,
195    pub source: OutputSource,
196    pub text: String,
197    pub next_cursor: String,
198    pub has_more: bool,
199    pub gap: bool,
200    pub resync_required: bool,
201    pub retained_start: u64,
202    pub retained_end: u64,
203    pub output_truncated: bool,
204    pub output_coverage: String,
205}
206
207/// File locks serialize adapters; immutable output revisions make index updates atomic.
208#[derive(Clone, Debug)]
209pub struct ExecutionCatalog {
210    root: PathBuf,
211    limits: HistoryLimits,
212}
213
214impl ExecutionCatalog {
215    pub const fn new(root: PathBuf, limits: HistoryLimits) -> Self {
216        Self { root, limits }
217    }
218
219    pub fn record(
220        &self,
221        scope: &ExecutionScope,
222        capture: ExecutionCapture,
223    ) -> HistoryResult<ExecutionRecord> {
224        let now = capture.observed_at_unix_ms;
225        storage::transaction(self, scope, now, move |transaction| {
226            update::record(transaction, &capture)
227        })
228    }
229
230    /// Update a known execution without recreating an expired or foreign record.
231    pub fn update_existing(
232        &self,
233        scope: &ExecutionScope,
234        capture: ExecutionCapture,
235    ) -> HistoryResult<Option<ExecutionRecord>> {
236        validate_id(&capture.execution_id)?;
237        let now = capture.observed_at_unix_ms;
238        storage::transaction(self, scope, now, move |transaction| {
239            let Some(previous) = transaction.find(&capture.execution_id) else {
240                return Ok(None);
241            };
242            if previous
243                .execution_token
244                .as_ref()
245                .zip(capture.execution_token.as_ref())
246                .is_none_or(|(previous, current)| previous != current)
247            {
248                return Err(HistoryError::IdentityConflict);
249            }
250            update::record(transaction, &capture).map(Some)
251        })
252    }
253
254    pub fn get(
255        &self,
256        scope: &ExecutionScope,
257        execution_id: &str,
258        now: u64,
259    ) -> HistoryResult<Option<ExecutionRecord>> {
260        validate_id(execution_id)?;
261        storage::transaction(self, scope, now, |transaction| {
262            Ok(transaction.find(execution_id).cloned())
263        })
264    }
265
266    pub fn list(&self, scope: &ExecutionScope, now: u64) -> HistoryResult<CatalogSnapshot> {
267        storage::transaction(self, scope, now, |transaction| {
268            let mut executions = transaction.index.executions.clone();
269            executions.sort_by_key(|record| std::cmp::Reverse(record.sequence));
270            Ok(CatalogSnapshot {
271                sequence: transaction.index.sequence,
272                cursor: paging::event_cursor(transaction)?,
273                executions,
274            })
275        })
276    }
277
278    pub fn read(
279        &self,
280        scope: &ExecutionScope,
281        execution_id: &str,
282        source: OutputSource,
283        cursor: Option<&str>,
284        limit: usize,
285        now: u64,
286    ) -> HistoryResult<OutputPage> {
287        validate_id(execution_id)?;
288        if !(4..=MAX_PAGE_BYTES).contains(&limit) {
289            return Err(HistoryError::InvalidInput(format!(
290                "page limit must be between 4 and {MAX_PAGE_BYTES} bytes"
291            )));
292        }
293        storage::transaction(self, scope, now, |transaction| {
294            paging::read(transaction, execution_id, source, cursor, limit)
295        })
296    }
297
298    pub fn changes(
299        &self,
300        scope: &ExecutionScope,
301        cursor: &str,
302        execution_ids: &[String],
303        now: u64,
304    ) -> HistoryResult<HistoryChanges> {
305        if execution_ids.len() > MAX_WATCHED_EXECUTIONS {
306            return Err(HistoryError::InvalidInput(
307                "at most 32 execution IDs may be watched".into(),
308            ));
309        }
310        for id in execution_ids {
311            validate_id(id)?;
312        }
313        storage::transaction(self, scope, now, |transaction| {
314            paging::changes(transaction, cursor, execution_ids)
315        })
316    }
317
318    pub fn mark_unknown(
319        &self,
320        scope: &ExecutionScope,
321        execution_id: &str,
322        reason: &str,
323        now: u64,
324    ) -> HistoryResult<Option<ExecutionRecord>> {
325        validate_id(execution_id)?;
326        storage::transaction(self, scope, now, |transaction| {
327            Ok(update::mark_unknown(transaction, execution_id, reason))
328        })
329    }
330}
331
332fn default_output_pending() -> bool {
333    true
334}
335
336fn validate_id(id: &str) -> HistoryResult<()> {
337    if id.is_empty() || id.len() > 512 || id.contains('\0') {
338        return Err(HistoryError::InvalidInput("execution ID is empty or too long".into()));
339    }
340    Ok(())
341}
winx_code_agent/state/execution_history.rs

winx_code_agent/state/
execution_history.rs
