1use serde::{Deserialize, Serialize};
7
8use crate::runtime::kernel::CancellationReason;
9use crate::runtime::session::SessionEvent;
10
11#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
12pub struct SignalDeliveryDisposedRecord {
13 pub turn: u32,
14 pub operation_id: String,
15 pub delivery_id: String,
16 pub attempt: u32,
17 pub signal_id: String,
18 pub disposition: String,
19 pub queue_depth: u32,
20}
21
22#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
23pub struct ProcessRecord {
24 pub turn: u32,
25 pub agent_id: String,
26 pub parent_session_id: String,
27 pub state: String,
28}
29
30#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
31pub struct SuspendRecord {
32 pub turn: u32,
33 pub reason: String,
34 pub pending_calls: Vec<String>,
35}
36
37#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
38pub struct BudgetExceededRecord {
39 pub turn: u32,
40 pub operation_id: String,
41 pub reservation_id: Option<String>,
42 pub budget: String,
43}
44
45#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
46pub struct BudgetUsageRecord {
47 pub turn: u32,
48 pub operation_id: String,
49 pub reservation_id: String,
50 pub tokens: u64,
51 pub subagents: u32,
52 pub rounds: u32,
53}
54
55#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
56pub struct CancellationRecord {
57 pub turn: u32,
58 pub operation_id: String,
59 pub reason: CancellationReason,
60 pub pending_call_ids: Vec<String>,
61}
62
63#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
65pub struct OsSnapshot {
66 pub last_suspend: Option<SuspendRecord>,
67 pub last_resumed_turn: Option<u32>,
68 pub process_by_agent: Vec<ProcessRecord>,
69 pub budget_exceeded: Vec<BudgetExceededRecord>,
70 pub budget_usage_reported: Vec<BudgetUsageRecord>,
71 pub cancellations: Vec<CancellationRecord>,
72 pub signals: Vec<SignalDeliveryDisposedRecord>,
73 pub page_out_count: u32,
74 pub page_in_count: u32,
75 pub tool_gated_count: u32,
76 #[serde(default)]
77 pub memory_written_count: u32,
78 #[serde(default)]
79 pub memory_queried_count: u32,
80 #[serde(default)]
81 pub memory_validation_failed_count: u32,
82 #[serde(default)]
83 pub memory_retrieval_result_count: u32,
84}
85
86pub fn rebuild_os_snapshot_from_events(events: &[SessionEvent]) -> OsSnapshot {
88 let mut snap = OsSnapshot::default();
89 let mut process_index: std::collections::HashMap<String, usize> =
90 std::collections::HashMap::new();
91
92 for event in events {
93 if !event.is_kernel_os_event()
94 && !matches!(event, SessionEvent::MemoryRetrievalResult { .. })
95 {
96 continue;
97 }
98
99 match event {
100 SessionEvent::Suspended {
101 turn,
102 reason,
103 pending_calls,
104 ..
105 } => {
106 snap.last_suspend = Some(SuspendRecord {
107 turn: *turn,
108 reason: reason.clone(),
109 pending_calls: pending_calls.clone(),
110 });
111 }
112 SessionEvent::Resumed { turn, .. } => {
113 snap.last_resumed_turn = Some(*turn);
114 }
115 SessionEvent::ToolGated { .. } => {
116 snap.tool_gated_count += 1;
117 }
118 SessionEvent::AgentProcessChanged {
119 turn,
120 agent_id,
121 parent_session_id,
122 state,
123 ..
124 } => {
125 let record = ProcessRecord {
126 turn: *turn,
127 agent_id: agent_id.clone(),
128 parent_session_id: parent_session_id.clone(),
129 state: state.clone(),
130 };
131 if let Some(idx) = process_index.get(agent_id) {
132 snap.process_by_agent[*idx] = record;
133 } else {
134 process_index.insert(agent_id.clone(), snap.process_by_agent.len());
135 snap.process_by_agent.push(record);
136 }
137 }
138 SessionEvent::BudgetExceeded {
139 turn,
140 operation_id,
141 reservation_id,
142 budget,
143 ..
144 } => {
145 snap.budget_exceeded.push(BudgetExceededRecord {
146 turn: *turn,
147 operation_id: operation_id.clone(),
148 reservation_id: reservation_id.clone(),
149 budget: budget.clone(),
150 });
151 }
152 SessionEvent::BudgetUsageReported {
153 turn,
154 operation_id,
155 reservation_id,
156 tokens,
157 subagents,
158 rounds,
159 } => {
160 snap.budget_usage_reported.push(BudgetUsageRecord {
161 turn: *turn,
162 operation_id: operation_id.clone(),
163 reservation_id: reservation_id.clone(),
164 tokens: *tokens,
165 subagents: *subagents,
166 rounds: *rounds,
167 });
168 }
169 SessionEvent::OperationCancelled {
170 turn,
171 operation_id,
172 reason,
173 pending_call_ids,
174 } => {
175 snap.cancellations.push(CancellationRecord {
176 turn: *turn,
177 operation_id: operation_id.clone(),
178 reason: *reason,
179 pending_call_ids: pending_call_ids.clone(),
180 });
181 }
182 SessionEvent::SignalDeliveryDisposed {
183 turn,
184 operation_id,
185 delivery_id,
186 attempt,
187 signal_id,
188 disposition,
189 queue_depth,
190 ..
191 } => {
192 snap.signals.push(SignalDeliveryDisposedRecord {
193 turn: *turn,
194 operation_id: operation_id.clone(),
195 delivery_id: delivery_id.clone(),
196 attempt: *attempt,
197 signal_id: signal_id.clone(),
198 disposition: disposition.clone(),
199 queue_depth: *queue_depth,
200 });
201 }
202 SessionEvent::PageOut { .. } => {
203 snap.page_out_count += 1;
204 }
205 SessionEvent::PageIn { .. } => {
206 snap.page_in_count += 1;
207 }
208 SessionEvent::MemoryWritten { .. } => {
209 snap.memory_written_count += 1;
210 }
211 SessionEvent::MemoryQueried { .. } => {
212 snap.memory_queried_count += 1;
213 }
214 SessionEvent::MemoryValidationFailed { .. } => {
215 snap.memory_validation_failed_count += 1;
216 }
217 SessionEvent::MemoryRetrievalResult { .. } => {
218 snap.memory_retrieval_result_count += 1;
219 }
220 _ => {}
221 }
222 }
223
224 snap
225}
226
227#[cfg(test)]
228mod tests {
229 use super::*;
230
231 #[test]
232 fn rebuild_tracks_process_and_signals() {
233 let events = vec![
234 SessionEvent::AgentProcessChanged {
235 turn: 1,
236 agent_id: "child-1".into(),
237 parent_session_id: "parent".into(),
238 role: "worker".into(),
239 isolation: "shared".into(),
240 context_inheritance: "none".into(),
241 state: "running".into(),
242 permitted_capability_ids: vec![],
243 result_termination: None,
244 },
245 SessionEvent::SignalDeliveryDisposed {
246 turn: 2,
247 operation_id: "op".into(),
248 delivery_id: "delivery".into(),
249 attempt: 1,
250 signal_id: "sig-a".into(),
251 disposition: "queue".into(),
252 queue_depth: 1,
253 },
254 SessionEvent::Suspended {
255 turn: 3,
256 reason: "ask_user".into(),
257 pending_calls: vec!["c1".into()],
258 },
259 SessionEvent::AgentProcessChanged {
260 turn: 4,
261 agent_id: "child-1".into(),
262 parent_session_id: "parent".into(),
263 role: "worker".into(),
264 isolation: "shared".into(),
265 context_inheritance: "none".into(),
266 state: "joined".into(),
267 permitted_capability_ids: vec![],
268 result_termination: Some("completed".into()),
269 },
270 ];
271 let snap = rebuild_os_snapshot_from_events(&events);
272 assert_eq!(snap.process_by_agent.len(), 1);
273 assert_eq!(snap.process_by_agent[0].state, "joined");
274 assert_eq!(snap.signals.len(), 1);
275 assert_eq!(
276 snap.last_suspend.as_ref().map(|s| s.reason.as_str()),
277 Some("ask_user")
278 );
279 }
280
281 fn load_fixture(name: &str) -> String {
282 let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
283 .join("../../tests/fixtures/session")
284 .join(name);
285 std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("read {}: {}", path.display(), e))
286 }
287
288 fn assert_golden(events_file: &str, snapshot_file: &str) {
289 let events: Vec<SessionEvent> =
290 serde_json::from_str(&load_fixture(events_file)).expect("events json");
291 let snap = rebuild_os_snapshot_from_events(&events);
292 let expected: OsSnapshot =
293 serde_json::from_str(&load_fixture(snapshot_file)).expect("snapshot json");
294 assert_eq!(snap, expected);
295 }
296
297 #[test]
298 fn golden_os_snapshot_spawn_lifecycle_fixture() {
299 assert_golden(
300 "events_spawn_lifecycle.json",
301 "os_snapshot_spawn_lifecycle.json",
302 );
303 }
304
305 #[test]
306 fn golden_os_snapshot_ask_user_fixture() {
307 assert_golden("events_ask_user.json", "os_snapshot_ask_user.json");
308 }
309}