1use std::collections::BTreeMap;
7
8use serde::{Deserialize, Serialize};
9
10use super::event::{EventType, SessionEvent};
11
12#[derive(Debug, Clone, Serialize, Deserialize)]
14pub struct FileAccess {
15 pub file_path: String,
16 pub agent_instance_id: String,
17 pub timestamp: String,
18 #[serde(skip_serializing_if = "Option::is_none")]
19 pub digest: Option<String>,
20 #[serde(default, skip_serializing_if = "Option::is_none")]
22 pub operation: Option<String>,
23 #[serde(default, skip_serializing_if = "Option::is_none")]
24 pub additions: Option<u32>,
25 #[serde(default, skip_serializing_if = "Option::is_none")]
26 pub deletions: Option<u32>,
27 #[serde(default, skip_serializing_if = "Option::is_none")]
43 pub source: Option<String>,
44}
45
46#[derive(Debug, Clone, Serialize, Deserialize)]
48pub struct PortAccess {
49 pub port: u16,
50 pub agent_instance_id: String,
51 pub timestamp: String,
52 #[serde(skip_serializing_if = "Option::is_none")]
53 pub protocol: Option<String>,
54}
55
56#[derive(Debug, Clone, Serialize, Deserialize)]
58pub struct NetworkConnection {
59 pub destination: String,
60 #[serde(skip_serializing_if = "Option::is_none")]
61 pub port: Option<u16>,
62 pub agent_instance_id: String,
63 pub timestamp: String,
64}
65
66#[derive(Debug, Clone, Serialize, Deserialize)]
68pub struct ProcessExecution {
69 pub process_name: String,
70 pub agent_instance_id: String,
71 pub started_at: String,
72 #[serde(skip_serializing_if = "Option::is_none")]
73 pub exit_code: Option<i32>,
74 #[serde(skip_serializing_if = "Option::is_none")]
75 pub duration_ms: Option<u64>,
76 #[serde(default, skip_serializing_if = "Option::is_none")]
78 pub command: Option<String>,
79 #[serde(default, skip_serializing_if = "Option::is_none")]
82 pub source: Option<String>,
83}
84
85#[derive(Debug, Clone, Serialize, Deserialize)]
87pub struct ToolInvocation {
88 pub tool_name: String,
89 pub agent_instance_id: String,
90 pub timestamp: String,
91 #[serde(skip_serializing_if = "Option::is_none")]
92 pub duration_ms: Option<u64>,
93}
94
95#[derive(Debug, Clone, Default, Serialize, Deserialize)]
97pub struct SideEffects {
98 pub files_read: Vec<FileAccess>,
99 pub files_written: Vec<FileAccess>,
100 pub ports_opened: Vec<PortAccess>,
101 pub network_connections: Vec<NetworkConnection>,
102 pub processes: Vec<ProcessExecution>,
103 pub tool_invocations: Vec<ToolInvocation>,
104}
105
106impl SideEffects {
107 pub fn from_events(events: &[SessionEvent]) -> Self {
109 let mut se = SideEffects::default();
110
111 let mut started_processes: BTreeMap<(String, String), usize> = BTreeMap::new();
114
115 for event in events {
116 match &event.event_type {
117 EventType::AgentReadFile { file_path, digest } => {
118 se.files_read.push(FileAccess {
119 file_path: file_path.clone(),
120 agent_instance_id: event.agent_instance_id.clone(),
121 timestamp: event.timestamp.clone(),
122 digest: digest.clone(),
123 operation: None,
124 additions: None,
125 deletions: None,
126 source: Some(source_from_meta(event, "hook")),
127 });
128 }
129
130 EventType::AgentWroteFile {
131 file_path,
132 digest,
133 operation,
134 additions,
135 deletions,
136 } => {
137 se.files_written.push(FileAccess {
138 file_path: file_path.clone(),
139 agent_instance_id: event.agent_instance_id.clone(),
140 timestamp: event.timestamp.clone(),
141 digest: digest.clone(),
142 operation: operation.clone(),
143 additions: *additions,
144 deletions: *deletions,
145 source: Some(source_from_meta(event, "hook")),
146 });
147 }
148
149 EventType::AgentOpenedPort { port, protocol } => {
150 se.ports_opened.push(PortAccess {
151 port: *port,
152 agent_instance_id: event.agent_instance_id.clone(),
153 timestamp: event.timestamp.clone(),
154 protocol: protocol.clone(),
155 });
156 }
157
158 EventType::AgentConnectedNetwork { destination, port } => {
159 se.network_connections.push(NetworkConnection {
160 destination: destination.clone(),
161 port: *port,
162 agent_instance_id: event.agent_instance_id.clone(),
163 timestamp: event.timestamp.clone(),
164 });
165 }
166
167 EventType::AgentStartedProcess {
168 process_name,
169 pid: _,
170 command,
171 } => {
172 let idx = se.processes.len();
173 se.processes.push(ProcessExecution {
174 process_name: process_name.clone(),
175 agent_instance_id: event.agent_instance_id.clone(),
176 started_at: event.timestamp.clone(),
177 exit_code: None,
178 duration_ms: None,
179 command: command.clone(),
180 source: Some(source_from_meta(event, "hook")),
181 });
182 started_processes
183 .insert((event.agent_instance_id.clone(), process_name.clone()), idx);
184 }
185
186 EventType::AgentCompletedProcess {
187 process_name,
188 exit_code,
189 duration_ms,
190 command,
191 } => {
192 let key = (event.agent_instance_id.clone(), process_name.clone());
193 if let Some(&idx) = started_processes.get(&key) {
194 if let Some(proc) = se.processes.get_mut(idx) {
195 proc.exit_code = *exit_code;
196 proc.duration_ms = *duration_ms;
197 if proc.command.is_none() {
198 proc.command = command.clone();
199 }
200 }
201 } else {
202 se.processes.push(ProcessExecution {
203 process_name: process_name.clone(),
204 agent_instance_id: event.agent_instance_id.clone(),
205 started_at: event.timestamp.clone(),
206 exit_code: *exit_code,
207 duration_ms: *duration_ms,
208 command: command.clone(),
209 source: Some(source_from_meta(event, "hook")),
210 });
211 }
212 }
213
214 EventType::AgentCalledTool {
215 tool_name,
216 duration_ms,
217 ..
218 } => {
219 se.tool_invocations.push(ToolInvocation {
220 tool_name: tool_name.clone(),
221 agent_instance_id: event.agent_instance_id.clone(),
222 timestamp: event.timestamp.clone(),
223 duration_ms: *duration_ms,
224 });
225
226 promote_mcp_called_tool(event, tool_name, &mut se);
243 }
244
245 _ => {}
246 }
247 }
248
249 se
250 }
251
252 pub fn summary(&self) -> SideEffectSummary {
254 SideEffectSummary {
255 files_read: self.files_read.len() as u32,
256 files_written: self.files_written.len() as u32,
257 ports_opened: self.ports_opened.len() as u32,
258 network_connections: self.network_connections.len() as u32,
259 processes: self.processes.len() as u32,
260 tool_invocations: self.tool_invocations.len() as u32,
261 }
262 }
263}
264
265#[derive(Debug, Clone, Serialize, Deserialize)]
267pub struct SideEffectSummary {
268 pub files_read: u32,
269 pub files_written: u32,
270 pub ports_opened: u32,
271 pub network_connections: u32,
272 pub processes: u32,
273 pub tool_invocations: u32,
274}
275
276fn meta_string(event: &SessionEvent, dotted_path: &str) -> Option<String> {
297 let mut cur = event.meta.as_ref()?;
298 for segment in dotted_path.split('.') {
299 cur = cur.get(segment)?;
300 }
301 cur.as_str().map(|s| s.to_string())
302}
303
304fn first_meta_string(event: &SessionEvent, paths: &[&str]) -> Option<String> {
306 for path in paths {
307 if let Some(v) = meta_string(event, path) {
308 if !v.is_empty() {
309 return Some(v);
310 }
311 }
312 }
313 None
314}
315
316fn source_from_meta(event: &SessionEvent, default: &str) -> String {
339 event
340 .meta
341 .as_ref()
342 .and_then(|m| m.get("source"))
343 .and_then(|v| v.as_str())
344 .filter(|s| !s.is_empty())
345 .map(|s| s.to_string())
346 .unwrap_or_else(|| default.to_string())
347}
348
349#[derive(Debug, Clone, Copy, PartialEq, Eq)]
354enum ToolCategory {
355 Read,
356 Write,
357 Process,
358 Unknown,
359}
360
361fn classify_tool(tool_name: &str) -> ToolCategory {
362 let t = tool_name.to_lowercase();
363 if t.contains("bash")
365 || t.contains("shell")
366 || t.contains("exec")
367 || t.contains("run_command")
368 || t.contains("ran_command")
369 {
370 return ToolCategory::Process;
371 }
372 if t.contains("write")
375 || t.contains("edit")
376 || t.contains("create_file")
377 || t.contains("modify")
378 || t.contains("patch")
379 || t.contains("save_file")
380 || t.contains("delete_file")
381 || t.contains("remove_file")
382 || t.contains("rename_file")
383 {
384 return ToolCategory::Write;
385 }
386 if t.contains("read")
388 || t.contains("view_file")
389 || t.contains("cat_file")
390 || t.contains("open_file")
391 || t.contains("get_file_contents")
392 {
393 return ToolCategory::Read;
394 }
395 ToolCategory::Unknown
396}
397
398fn promote_mcp_called_tool(event: &SessionEvent, tool_name: &str, se: &mut SideEffects) {
399 let category = classify_tool(tool_name);
400
401 let file_path = first_meta_string(
404 event,
405 &[
406 "tool_input.file_path",
407 "tool_input.path",
408 "tool_input.notebook_path",
409 "tool_input.target_file",
410 "file_path",
411 "path",
412 ],
413 );
414
415 let command = first_meta_string(
417 event,
418 &["tool_input.command", "command", "tool_input.cmd", "cmd"],
419 );
420
421 match (category, file_path, command) {
422 (ToolCategory::Read, Some(p), _) => {
423 se.files_read.push(FileAccess {
424 file_path: p,
425 agent_instance_id: event.agent_instance_id.clone(),
426 timestamp: event.timestamp.clone(),
427 digest: None,
428 operation: None,
429 additions: None,
430 deletions: None,
431 source: Some("mcp".into()),
432 });
433 }
434 (ToolCategory::Write, Some(p), _) => {
435 se.files_written.push(FileAccess {
436 file_path: p,
437 agent_instance_id: event.agent_instance_id.clone(),
438 timestamp: event.timestamp.clone(),
439 digest: None,
440 operation: None,
441 additions: None,
442 deletions: None,
443 source: Some("mcp".into()),
444 });
445 }
446 (ToolCategory::Process, _, Some(cmd)) => {
447 let short = cmd.chars().take(120).collect::<String>();
450 se.processes.push(ProcessExecution {
451 process_name: short,
452 agent_instance_id: event.agent_instance_id.clone(),
453 started_at: event.timestamp.clone(),
454 exit_code: None,
455 duration_ms: None,
456 command: Some(cmd),
457 source: Some("mcp".into()),
458 });
459 }
460 (ToolCategory::Unknown, Some(p), _) => {
461 se.files_written.push(FileAccess {
467 file_path: p,
468 agent_instance_id: event.agent_instance_id.clone(),
469 timestamp: event.timestamp.clone(),
470 digest: None,
471 operation: None,
472 additions: None,
473 deletions: None,
474 source: Some("mcp".into()),
475 });
476 }
477 _ => {
478 }
483 }
484}
485
486#[cfg(test)]
487mod tests {
488 use super::*;
489 use crate::session::event::*;
490
491 fn evt(event_type: EventType) -> SessionEvent {
492 SessionEvent {
493 session_id: "ssn_001".into(),
494 event_id: generate_event_id(),
495 timestamp: "2026-04-05T08:00:00Z".into(),
496 sequence_no: 0,
497 trace_id: "t".into(),
498 span_id: "s".into(),
499 parent_span_id: None,
500 agent_id: "agent://test".into(),
501 agent_instance_id: "ai_1".into(),
502 agent_name: "test".into(),
503 agent_role: None,
504 host_id: "h".into(),
505 tool_runtime_id: None,
506 event_type,
507 artifact_ref: None,
508 meta: None,
509 }
510 }
511
512 #[test]
513 fn aggregates_file_and_tool_events() {
514 let events = vec![
515 evt(EventType::AgentReadFile {
516 file_path: "src/main.rs".into(),
517 digest: None,
518 }),
519 evt(EventType::AgentWroteFile {
520 file_path: "src/lib.rs".into(),
521 digest: Some("sha256:abc".into()),
522 operation: Some("modified".into()),
523 additions: Some(10),
524 deletions: Some(3),
525 }),
526 evt(EventType::AgentCalledTool {
527 tool_name: "read_file".into(),
528 tool_input_digest: None,
529 tool_output_digest: None,
530 duration_ms: Some(10),
531 }),
532 evt(EventType::AgentCalledTool {
533 tool_name: "write_file".into(),
534 tool_input_digest: None,
535 tool_output_digest: None,
536 duration_ms: None,
537 }),
538 ];
539
540 let se = SideEffects::from_events(&events);
541 assert_eq!(se.files_read.len(), 1);
542 assert_eq!(se.files_written.len(), 1);
543 assert_eq!(se.tool_invocations.len(), 2);
544 let summary = se.summary();
545 assert_eq!(summary.tool_invocations, 2);
546 }
547
548 #[test]
549 fn matches_process_start_and_complete() {
550 let events = vec![
551 evt(EventType::AgentStartedProcess {
552 process_name: "npm test".into(),
553 pid: Some(1234),
554 command: Some("npm test --runInBand".into()),
555 }),
556 evt(EventType::AgentCompletedProcess {
557 process_name: "npm test".into(),
558 exit_code: Some(0),
559 duration_ms: Some(5000),
560 command: None,
561 }),
562 ];
563
564 let se = SideEffects::from_events(&events);
565 assert_eq!(se.processes.len(), 1);
566 assert_eq!(se.processes[0].exit_code, Some(0));
567 assert_eq!(se.processes[0].duration_ms, Some(5000));
568 }
569
570 fn called_tool_with_meta(tool_name: &str, meta: serde_json::Value) -> SessionEvent {
573 let mut e = evt(EventType::AgentCalledTool {
574 tool_name: tool_name.into(),
575 tool_input_digest: None,
576 tool_output_digest: None,
577 duration_ms: None,
578 });
579 e.meta = Some(meta);
580 e
581 }
582
583 #[test]
584 fn hook_file_events_carry_source_hook() {
585 let events = vec![
589 evt(EventType::AgentReadFile {
590 file_path: "src/a.rs".into(),
591 digest: None,
592 }),
593 evt(EventType::AgentWroteFile {
594 file_path: "src/b.rs".into(),
595 digest: None,
596 operation: None,
597 additions: None,
598 deletions: None,
599 }),
600 ];
601 let se = SideEffects::from_events(&events);
602 assert_eq!(se.files_read[0].source.as_deref(), Some("hook"));
603 assert_eq!(se.files_written[0].source.as_deref(), Some("hook"));
604 }
605
606 #[test]
607 fn mcp_called_tool_with_file_path_promotes_to_files_written() {
608 let events = vec![called_tool_with_meta(
614 "Edit",
615 serde_json::json!({
616 "source": "mcp-bridge",
617 "tool_input": { "file_path": "src/api/receipt.ts" },
618 }),
619 )];
620 let se = SideEffects::from_events(&events);
621 assert_eq!(
622 se.files_written.len(),
623 1,
624 "Edit with file_path must promote to files_written"
625 );
626 assert_eq!(se.files_written[0].file_path, "src/api/receipt.ts");
627 assert_eq!(se.files_written[0].source.as_deref(), Some("mcp"));
628 assert_eq!(se.tool_invocations.len(), 1);
631 }
632
633 #[test]
634 fn mcp_read_tool_promotes_to_files_read() {
635 let events = vec![called_tool_with_meta(
636 "Read",
637 serde_json::json!({ "tool_input": { "file_path": "package.json" } }),
638 )];
639 let se = SideEffects::from_events(&events);
640 assert_eq!(se.files_read.len(), 1);
641 assert_eq!(se.files_read[0].file_path, "package.json");
642 assert_eq!(se.files_read[0].source.as_deref(), Some("mcp"));
643 }
644
645 #[test]
646 fn mcp_bash_tool_promotes_to_processes() {
647 let events = vec![called_tool_with_meta(
648 "Bash",
649 serde_json::json!({ "tool_input": { "command": "bun test --run" } }),
650 )];
651 let se = SideEffects::from_events(&events);
652 assert_eq!(se.processes.len(), 1);
653 assert_eq!(se.processes[0].command.as_deref(), Some("bun test --run"));
654 assert_eq!(se.processes[0].source.as_deref(), Some("mcp"));
655 }
656
657 #[test]
658 fn mcp_unknown_tool_with_path_defaults_to_files_written() {
659 let events = vec![called_tool_with_meta(
664 "mcp__weird-vendor__do_thing",
665 serde_json::json!({ "tool_input": { "file_path": "config.toml" } }),
666 )];
667 let se = SideEffects::from_events(&events);
668 assert_eq!(se.files_written.len(), 1);
669 assert_eq!(se.files_written[0].file_path, "config.toml");
670 assert_eq!(se.files_written[0].source.as_deref(), Some("mcp"));
671 }
672
673 #[test]
674 fn mcp_called_tool_without_meta_does_not_promote() {
675 let events = vec![called_tool_with_meta(
680 "ls",
681 serde_json::json!({"source": "mcp-bridge"}),
682 )];
683 let se = SideEffects::from_events(&events);
684 assert_eq!(se.files_read.len(), 0);
685 assert_eq!(se.files_written.len(), 0);
686 assert_eq!(se.processes.len(), 0);
687 assert_eq!(se.tool_invocations.len(), 1);
688 }
689
690 #[test]
691 fn mcp_promotion_handles_alt_path_field_names() {
692 for path_field in &[
695 "tool_input.path",
696 "tool_input.target_file",
697 "file_path",
698 "path",
699 ] {
700 let mut meta_obj = serde_json::Map::new();
701 let parts: Vec<&str> = path_field.split('.').collect();
703 if parts.len() == 1 {
704 meta_obj.insert(parts[0].into(), serde_json::json!("x.txt"));
705 } else {
706 let inner = serde_json::json!({ parts[1]: "x.txt" });
707 meta_obj.insert(parts[0].into(), inner);
708 }
709 let events = vec![called_tool_with_meta(
710 "Edit",
711 serde_json::Value::Object(meta_obj),
712 )];
713 let se = SideEffects::from_events(&events);
714 assert_eq!(
715 se.files_written.len(),
716 1,
717 "expected promotion via {} but got nothing",
718 path_field,
719 );
720 }
721 }
722
723 fn evt_with_meta(et: EventType, meta: serde_json::Value) -> SessionEvent {
733 let mut e = evt(et);
734 e.meta = Some(meta);
735 e
736 }
737
738 #[test]
739 fn source_from_meta_preserves_session_event_cli() {
740 let events = vec![evt_with_meta(
744 EventType::AgentWroteFile {
745 file_path: "src/x.rs".into(),
746 digest: None,
747 operation: None,
748 additions: None,
749 deletions: None,
750 },
751 serde_json::json!({"source": "session-event-cli"}),
752 )];
753 let se = SideEffects::from_events(&events);
754 assert_eq!(
755 se.files_written[0].source.as_deref(),
756 Some("session-event-cli"),
757 "session-event-cli must be preserved verbatim, not downgraded to hook"
758 );
759 }
760
761 #[test]
762 fn source_from_meta_preserves_daemon_atime() {
763 let events = vec![evt_with_meta(
767 EventType::AgentReadFile {
768 file_path: "src/x.rs".into(),
769 digest: None,
770 },
771 serde_json::json!({"source": "daemon-atime"}),
772 )];
773 let se = SideEffects::from_events(&events);
774 assert_eq!(
775 se.files_read[0].source.as_deref(),
776 Some("daemon-atime"),
777 "daemon-atime must be preserved verbatim, not downgraded to hook"
778 );
779 }
780
781 #[test]
782 fn source_from_meta_preserves_arbitrary_unknown_label() {
783 let events = vec![evt_with_meta(
788 EventType::AgentWroteFile {
789 file_path: "x".into(),
790 digest: None,
791 operation: None,
792 additions: None,
793 deletions: None,
794 },
795 serde_json::json!({"source": "future-bridge-v2"}),
796 )];
797 let se = SideEffects::from_events(&events);
798 assert_eq!(
799 se.files_written[0].source.as_deref(),
800 Some("future-bridge-v2")
801 );
802 }
803
804 #[test]
805 fn source_from_meta_falls_back_when_meta_source_absent() {
806 let events = vec![evt(EventType::AgentWroteFile {
812 file_path: "x".into(),
813 digest: None,
814 operation: None,
815 additions: None,
816 deletions: None,
817 })];
818 let se = SideEffects::from_events(&events);
819 assert_eq!(se.files_written[0].source.as_deref(), Some("hook"));
820 }
821
822 #[test]
823 fn source_from_meta_treats_empty_string_as_absent() {
824 let events = vec![evt_with_meta(
827 EventType::AgentReadFile {
828 file_path: "x".into(),
829 digest: None,
830 },
831 serde_json::json!({"source": ""}),
832 )];
833 let se = SideEffects::from_events(&events);
834 assert_eq!(se.files_read[0].source.as_deref(), Some("hook"));
835 }
836}