1use std::fs;
2use std::path::Path;
3
4use anyhow::Result;
5
6use crate::WorkflowEventEnvelope;
7
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub struct EventWindow {
10 pub events: Vec<WorkflowEventEnvelope>,
11 pub oldest_seq: Option<u64>,
12 pub newest_seq: Option<u64>,
13 pub total_count: usize,
14 pub has_older: bool,
15 pub has_newer: bool,
16}
17
18const DEFAULT_LIMIT: usize = 200;
19const MAX_LIMIT: usize = 1000;
20const DEFAULT_TAIL: usize = 100;
21
22pub fn event_seq_from_id(event_id: &str) -> u64 {
23 event_id
24 .rsplit_once('-')
25 .and_then(|(_, seq)| seq.parse::<u64>().ok())
26 .unwrap_or(0)
27}
28
29pub fn infer_run_status(events: &[WorkflowEventEnvelope]) -> String {
30 for ev in events.iter().rev() {
31 match ev.event_type.as_str() {
32 "runCanceled" => return "cancelled".to_string(),
33 "runSucceeded" => return "succeeded".to_string(),
34 "runFailed" => return "failed".to_string(),
35 "runStarted" => return "running".to_string(),
36 "runCreated" => return "pending".to_string(),
37 _ => {}
38 }
39 }
40 "pending".to_string()
41}
42
43pub fn read_run_events_pure(run_dir: &Path) -> Result<Option<Vec<WorkflowEventEnvelope>>> {
44 let file = run_dir.join("events.ndjson");
45 let raw = match fs::read_to_string(&file) {
46 Ok(raw) => raw,
47 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(None),
48 Err(err) => return Err(err.into()),
49 };
50 let mut events = Vec::new();
51 for line in raw.lines() {
52 if line.trim().is_empty() {
53 continue;
54 }
55 let parsed = serde_json::from_str::<WorkflowEventEnvelope>(line);
56 match parsed {
57 Ok(ev) => events.push(ev),
58 Err(_) => return Ok(None),
59 }
60 }
61 Ok(Some(events))
62}
63
64pub fn read_event_window(events: &[WorkflowEventEnvelope], opts: EventWindowOpts) -> EventWindow {
65 let total = events.len();
66 if total == 0 {
67 return EventWindow {
68 events: Vec::new(),
69 oldest_seq: None,
70 newest_seq: None,
71 total_count: 0,
72 has_older: false,
73 has_newer: false,
74 };
75 }
76 let limit = clamp_limit(opts.limit);
77 if let Some(after_seq) = opts.after_seq {
78 let idx = events
79 .iter()
80 .position(|e| event_seq_from_id(&e.event_id) > after_seq)
81 .unwrap_or(total);
82 let slice = events[idx..std::cmp::min(idx + limit, total)].to_vec();
83 return EventWindow {
84 oldest_seq: slice.first().map(|e| event_seq_from_id(&e.event_id)),
85 newest_seq: slice.last().map(|e| event_seq_from_id(&e.event_id)),
86 has_older: idx > 0,
87 has_newer: idx + slice.len() < total,
88 total_count: total,
89 events: slice,
90 };
91 }
92 if let Some(before_seq) = opts.before_seq {
93 let end_idx = events
94 .iter()
95 .position(|e| event_seq_from_id(&e.event_id) >= before_seq)
96 .unwrap_or(total);
97 let start_idx = end_idx.saturating_sub(limit);
98 let slice = events[start_idx..end_idx].to_vec();
99 return EventWindow {
100 oldest_seq: slice.first().map(|e| event_seq_from_id(&e.event_id)),
101 newest_seq: slice.last().map(|e| event_seq_from_id(&e.event_id)),
102 has_older: start_idx > 0,
103 has_newer: end_idx < total,
104 total_count: total,
105 events: slice,
106 };
107 }
108 let start_idx = total.saturating_sub(opts.tail.unwrap_or(DEFAULT_TAIL).clamp(1, MAX_LIMIT));
109 let slice = events[start_idx..].to_vec();
110 EventWindow {
111 oldest_seq: slice.first().map(|e| event_seq_from_id(&e.event_id)),
112 newest_seq: slice.last().map(|e| event_seq_from_id(&e.event_id)),
113 has_older: start_idx > 0,
114 has_newer: false,
115 total_count: total,
116 events: slice,
117 }
118}
119
120#[derive(Debug, Clone, Copy, Default)]
121pub struct EventWindowOpts {
122 pub tail: Option<usize>,
123 pub before_seq: Option<u64>,
124 pub after_seq: Option<u64>,
125 pub limit: Option<usize>,
126}
127
128fn clamp_limit(raw: Option<usize>) -> usize {
129 match raw {
130 Some(0) | None => DEFAULT_LIMIT,
131 Some(v) => v.min(MAX_LIMIT),
132 }
133}
134
135#[cfg(test)]
136mod tests {
137 use super::*;
138
139 fn ev(event_id: &str, event_type: &str) -> WorkflowEventEnvelope {
140 WorkflowEventEnvelope {
141 event_id: event_id.to_string(),
142 run_id: "run-1".to_string(),
143 timestamp: 1,
144 schema_version: 1,
145 actor: crate::WorkflowActor::System,
146 event_type: event_type.to_string(),
147 payload: serde_json::json!({}),
148 payload_hash: None,
149 }
150 }
151
152 #[test]
153 fn infer_run_status_prefers_latest_terminal_event() {
154 let events = vec![ev("run-1-1", "runCreated"), ev("run-1-2", "runCanceled")];
155 assert_eq!(infer_run_status(&events), "cancelled");
156 }
157
158 #[test]
159 fn read_event_window_supports_tail_and_after_seq() {
160 let events = vec![
161 ev("run-1-1", "runCreated"),
162 ev("run-1-2", "runStarted"),
163 ev("run-1-3", "nodeWaiting"),
164 ev("run-1-4", "runSucceeded"),
165 ];
166 let tail = read_event_window(
167 &events,
168 EventWindowOpts {
169 tail: Some(2),
170 before_seq: None,
171 after_seq: None,
172 limit: None,
173 },
174 );
175 assert_eq!(tail.events.len(), 2);
176 assert_eq!(tail.oldest_seq, Some(3));
177 assert_eq!(tail.newest_seq, Some(4));
178
179 let after = read_event_window(
180 &events,
181 EventWindowOpts {
182 tail: None,
183 before_seq: None,
184 after_seq: Some(2),
185 limit: Some(2),
186 },
187 );
188 assert_eq!(after.events.len(), 2);
189 assert_eq!(after.oldest_seq, Some(3));
190 assert_eq!(after.newest_seq, Some(4));
191 }
192}