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    /// Foreground records a new launch must reconcile first: still running or
279    /// with output not yet captured.
280    pub fn pending_foreground(
281        &self,
282        scope: &ExecutionScope,
283        now: u64,
284    ) -> HistoryResult<Vec<ExecutionRecord>> {
285        // Unlike `list`, this clones only the matching records and mints no
286        // event cursor: it runs before every foreground launch.
287        storage::transaction(self, scope, now, |transaction| {
288            let mut pending = transaction
289                .index
290                .executions
291                .iter()
292                .filter(|record| {
293                    record.background_id.is_none()
294                        && (record.process_status == ExecutionProcessStatus::Running
295                            || record.output_pending)
296                })
297                .cloned()
298                .collect::<Vec<_>>();
299            pending.sort_by_key(|record| std::cmp::Reverse(record.sequence));
300            Ok(pending)
301        })
302    }
303
304    pub fn read(
305        &self,
306        scope: &ExecutionScope,
307        execution_id: &str,
308        source: OutputSource,
309        cursor: Option<&str>,
310        limit: usize,
311        now: u64,
312    ) -> HistoryResult<OutputPage> {
313        validate_id(execution_id)?;
314        if !(4..=MAX_PAGE_BYTES).contains(&limit) {
315            return Err(HistoryError::InvalidInput(format!(
316                "page limit must be between 4 and {MAX_PAGE_BYTES} bytes"
317            )));
318        }
319        storage::transaction(self, scope, now, |transaction| {
320            paging::read(transaction, execution_id, source, cursor, limit)
321        })
322    }
323
324    pub fn changes(
325        &self,
326        scope: &ExecutionScope,
327        cursor: &str,
328        execution_ids: &[String],
329        now: u64,
330    ) -> HistoryResult<HistoryChanges> {
331        if execution_ids.len() > MAX_WATCHED_EXECUTIONS {
332            return Err(HistoryError::InvalidInput(
333                "at most 32 execution IDs may be watched".into(),
334            ));
335        }
336        for id in execution_ids {
337            validate_id(id)?;
338        }
339        storage::transaction(self, scope, now, |transaction| {
340            paging::changes(transaction, cursor, execution_ids)
341        })
342    }
343
344    pub fn mark_unknown(
345        &self,
346        scope: &ExecutionScope,
347        execution_id: &str,
348        reason: &str,
349        now: u64,
350    ) -> HistoryResult<Option<ExecutionRecord>> {
351        validate_id(execution_id)?;
352        storage::transaction(self, scope, now, |transaction| {
353            Ok(update::mark_unknown(transaction, execution_id, reason))
354        })
355    }
356}
357
358fn default_output_pending() -> bool {
359    true
360}
361
362fn validate_id(id: &str) -> HistoryResult<()> {
363    if id.is_empty() || id.len() > 512 || id.contains('\0') {
364        return Err(HistoryError::InvalidInput("execution ID is empty or too long".into()));
365    }
366    Ok(())
367}
winx_code_agent/state/execution_history.rs

winx_code_agent/state/
execution_history.rs
