1use std::time::{Instant, SystemTime, UNIX_EPOCH};
2
3use murmur_artifact::{ContainmentClass, LockOrigin, TraceCapture};
4use serde::Serialize;
5use serde_json::Value;
6use tokio::{
7 fs::File,
8 io::{AsyncWriteExt, BufWriter},
9};
10
11use crate::{
12 agent::{DriverUsage, OccupancyBasis},
13 containment::ScopeReport,
14 formation::FormationId,
15 gateway_credential::CredentialChange,
16 inference_retry::{InferenceRetryRecord, RetryNote, RetryStop},
17 lanes::TaskLane,
18 origin::{TaskProvenance, TrustClass},
19 trace_blobs::BlobStore,
20 types::InstalledArtifactSummary,
21 workdir_fs::{WorkdirRoot, DEFAULT_FILE_MODE},
22};
23
24pub(crate) const TRACE_FILE_NAME: &str = "trace.jsonl";
26
27pub(crate) struct TraceWriter {
28 writer: BufWriter<File>,
29 session_id: String,
30 capsule_name: String,
31 capsule_version: String,
32 model: String,
33 capabilities: Vec<String>,
34 effective_grants: ScopeReport,
40 capture: TraceCapture,
44 blobs: BlobStore,
47 system_prompt_source: &'static str,
52 system_prompt_sha256: Option<String>,
56 credential_source: &'static str,
59 gateways: Vec<SessionGateway>,
62 control: Option<SessionControl>,
65 inference_choices: Vec<SessionInferenceChoice>,
68 tool_refresh: Option<&'static str>,
71 runtime_artifacts: Vec<RuntimeArtifact>,
74 host_facts: Vec<HostFactsRecord>,
77 resumed_from: Option<String>,
81 context_id: Option<String>,
84 spawned_by: Option<String>,
88 spawn_delegation_id: Option<String>,
89 formation_id: Option<FormationId>,
92 formation_member: Option<(String, Vec<String>)>,
95 session_start_time: Instant,
96 session_event_id: String,
101 session_started: bool,
102 session_ended: bool,
103 task_event_id: Option<String>,
106 turn_event_id: Option<String>,
110 total_turns: u32,
112 total_input_tokens: u64,
113 total_output_tokens: u64,
114 total_tool_calls: u32,
115 total_shell_calls: u32,
116 task_turns: u32,
118 task_input_tokens: u64,
119 task_output_tokens: u64,
120 task_tool_calls: u32,
121 task_shell_calls: u32,
122 task_agent_turns: u32,
127 task_hook_inferences: u32,
130 task_start_instant: Option<Instant>,
131 pub(crate) active_task_id: Option<String>,
132 task_failures_written: u64,
135 last_task_failure: Option<String>,
137 max_session_tokens: Option<u64>,
139 machine_tokens_per_day: Option<u64>,
140 session_status: Option<std::sync::Arc<crate::session_status::SessionStatus>>,
144}
145
146#[derive(Debug, Clone)]
151pub(crate) struct InferenceOrigin {
152 pub(crate) source: String,
154 pub(crate) model: String,
156}
157
158#[derive(Debug, Clone, PartialEq)]
160pub(crate) struct InferenceCallSettings {
161 pub(crate) max_output_tokens: u32,
164 pub(crate) input_budget: Option<u32>,
167 pub(crate) driver_config: Option<Value>,
170}
171
172pub(crate) fn recorded_driver_config(
180 env: &[(String, String)],
181 redact: impl Fn(&str) -> String,
182) -> Option<Value> {
183 fn redact_value(value: Value, redact: &impl Fn(&str) -> String) -> Value {
184 match value {
185 Value::String(text) => Value::String(redact(&text)),
186 Value::Number(number) => {
187 let text = number.to_string();
188 if redact(&text) == text {
189 Value::Number(number)
190 } else {
191 Value::String(crate::gateway_credential::REDACTED.to_string())
192 }
193 }
194 Value::Array(items) => Value::Array(
195 items
196 .into_iter()
197 .map(|item| redact_value(item, redact))
198 .collect(),
199 ),
200 Value::Object(entries) => Value::Object(
201 entries
202 .into_iter()
203 .map(|(key, item)| (redact(&key), redact_value(item, redact)))
204 .collect(),
205 ),
206 other @ (Value::Bool(_) | Value::Null) => other,
207 }
208 }
209 let (_, raw) = env
210 .iter()
211 .find(|(key, _)| key == "MURMUR_INFERENCE_DRIVER_CONFIG")?;
212 Some(match serde_json::from_str::<Value>(raw) {
213 Ok(value) => redact_value(value, &redact),
214 Err(_) => Value::String(redact(raw)),
215 })
216}
217
218pub(crate) const COMPACTION_DECLINED_NO_HOOK_REPLACEMENT: &str = "no_hook_replacement";
221
222pub(crate) const COMPACTION_DECLINED_UNRESOLVED_TOOL_CALL: &str = "unresolved_tool_call";
225
226pub(crate) const SEED_OUTCOME_SEEDED: &str = "seeded";
228
229pub(crate) const SEED_OUTCOME_TRIMMED: &str = "trimmed";
232
233pub(crate) const SEED_OUTCOME_COMPACTED: &str = "compacted";
236
237pub(crate) const SEED_OUTCOME_REJECTED: &str = "rejected";
239
240pub(crate) const SEED_REJECTED_MESSAGE_OVER_BUDGET: &str = "message_over_budget";
243
244pub(crate) const SEED_REJECTED_OVERFLOW_OVER_LIMIT: &str = "overflow_over_limit";
247
248pub(crate) const SEED_REJECTED_NO_BUDGET: &str = "no_budget";
251
252pub(crate) const SEED_REJECTED_UNSUPPORTED_TRANSPORT: &str = "unsupported_transport";
255
256pub(crate) const RETENTION_STORE_SESSIONS: &str = "sessions";
258
259pub(crate) const RETENTION_STORE_RECORDS: &str = "records";
261
262pub(crate) const RETENTION_REASON_MAX_SESSIONS: &str = "max_sessions";
264
265pub(crate) const RETENTION_REASON_MAX_AGE: &str = "max_age";
268
269pub(crate) const RETENTION_REASON_MAX_MESSAGES: &str = "max_messages";
271
272pub(crate) fn new_event_id() -> String {
280 format!("evt_{}", uuid::Uuid::now_v7().simple())
281}
282
283#[derive(Serialize)]
297struct SessionStartEvent {
298 event_type: &'static str,
299 event_id: String,
300 parent_id: Option<String>,
301 session_id: String,
302 timestamp: u64,
303 capsule_name: String,
304 capsule_version: String,
305 model: String,
306 max_turns: Option<u32>,
308 model_limits: Option<crate::model_limits::ResolvedModelLimits>,
311 max_session_tokens: Option<u64>,
314 machine_tokens_per_day: Option<u64>,
317 deadline: Option<SessionStartDeadline>,
320 capabilities: Vec<String>,
321 tools_declared: Vec<String>,
322 containment_declared: ContainmentClass,
324 containment_achieved: ContainmentClass,
329 workdir_exec: bool,
337 userns_grant: Option<crate::sealed::UsernsGrant>,
349 effective_grants: ScopeReport,
358 system_prompt_source: &'static str,
366 system_prompt_sha256: Option<String>,
375 credential_source: &'static str,
378 gateways: Vec<SessionGateway>,
382 #[serde(skip_serializing_if = "Option::is_none")]
385 control: Option<SessionControl>,
386 #[serde(skip_serializing_if = "Vec::is_empty")]
390 inference_choices: Vec<SessionInferenceChoice>,
391 tool_refresh: Option<&'static str>,
394 runtime_artifacts: Vec<RuntimeArtifact>,
398 host_facts: Vec<HostFactsRecord>,
402 resumed_from: Option<String>,
409 context_id: Option<String>,
415 #[serde(skip_serializing_if = "Option::is_none")]
419 spawned_by: Option<String>,
420 #[serde(skip_serializing_if = "Option::is_none")]
423 delegation_id: Option<String>,
424 #[serde(skip_serializing_if = "Option::is_none")]
428 formation_id: Option<FormationId>,
429 #[serde(skip_serializing_if = "Option::is_none")]
432 formation_member: Option<String>,
433 #[serde(skip_serializing_if = "Option::is_none")]
436 formation_callees: Option<Vec<String>>,
437}
438
439#[derive(Serialize)]
441struct SessionStartDeadline {
442 at: String,
444 secs: u64,
447 notice_secs: u64,
450 source: &'static str,
452}
453
454#[derive(Serialize)]
455struct InferenceEvent {
456 event_type: &'static str,
457 event_id: String,
458 parent_id: Option<String>,
459 session_id: String,
460 timestamp: u64,
461 turn: u32,
462 task_id: Option<String>,
463 #[serde(skip_serializing_if = "Option::is_none")]
476 input_tokens: Option<u64>,
477 #[serde(skip_serializing_if = "Option::is_none")]
481 output_tokens: Option<u64>,
482 decision: String,
483 #[serde(skip_serializing_if = "Option::is_none")]
490 stop_reason: Option<String>,
491 tool_name: Option<String>,
492 #[serde(skip_serializing_if = "Option::is_none")]
495 input_tokens_actual: Option<u64>,
496 #[serde(skip_serializing_if = "Option::is_none")]
497 output_tokens_actual: Option<u64>,
498 #[serde(skip_serializing_if = "Option::is_none")]
499 cached_tokens: Option<u64>,
500 #[serde(skip_serializing_if = "Option::is_none")]
501 cache_write_tokens: Option<u64>,
502 #[serde(skip_serializing_if = "Option::is_none")]
505 thinking_tokens: Option<u64>,
506 #[serde(skip_serializing_if = "Option::is_none")]
511 origin: Option<String>,
512 #[serde(skip_serializing_if = "Option::is_none")]
515 driver_choice: Option<String>,
516 #[serde(skip_serializing_if = "Option::is_none")]
521 model: Option<String>,
522 #[serde(skip_serializing_if = "Option::is_none")]
527 max_output_tokens: Option<u32>,
528 #[serde(skip_serializing_if = "Option::is_none")]
532 input_budget: Option<Option<u32>>,
533 #[serde(skip_serializing_if = "Option::is_none")]
538 driver_config: Option<Option<Value>>,
539 #[serde(skip_serializing_if = "Vec::is_empty")]
544 message_ids: Vec<String>,
545 #[serde(skip_serializing_if = "Option::is_none")]
553 system_sha: Option<String>,
554 #[serde(skip_serializing_if = "Option::is_none")]
557 tools_sha: Option<String>,
558 #[serde(skip_serializing_if = "Option::is_none")]
561 response_sha: Option<String>,
562 #[serde(skip_serializing_if = "Vec::is_empty")]
570 message_shas: Vec<String>,
571 #[serde(skip_serializing_if = "Option::is_none")]
575 error_code: Option<String>,
576 #[serde(skip_serializing_if = "Option::is_none")]
580 error: Option<String>,
581 #[serde(skip_serializing_if = "Option::is_none")]
585 provider_status: Option<u16>,
586 #[serde(skip_serializing_if = "Option::is_none")]
589 attempts: Option<u32>,
590 #[serde(skip_serializing_if = "Option::is_none")]
593 retry_stopped: Option<&'static str>,
594}
595
596pub(crate) struct WireCapture {
606 system: Vec<u8>,
609 tools: Vec<u8>,
611 messages: Vec<Vec<u8>>,
615 response: Vec<u8>,
617}
618
619impl WireCapture {
620 pub(crate) fn from_driver_payload(payload: &Value, response: &str) -> Self {
626 Self {
627 system: payload
628 .get("system")
629 .and_then(Value::as_str)
630 .unwrap_or_default()
631 .as_bytes()
632 .to_vec(),
633 tools: payload
634 .get("tools")
635 .map(|tools| serde_json::to_vec(tools).unwrap_or_default())
636 .unwrap_or_default(),
637 messages: payload
638 .get("messages")
639 .and_then(Value::as_array)
640 .map(|messages| {
641 messages
642 .iter()
643 .map(|message| serde_json::to_vec(message).unwrap_or_default())
644 .collect()
645 })
646 .unwrap_or_default(),
647 response: response.as_bytes().to_vec(),
648 }
649 }
650}
651
652#[derive(Serialize)]
653struct ToolCallEvent {
654 event_type: &'static str,
655 event_id: String,
656 parent_id: Option<String>,
657 session_id: String,
658 timestamp: u64,
659 turn: u32,
660 task_id: Option<String>,
661 tool_name: String,
662 tool_call_id: Option<String>,
666 input: Value,
667 input_bytes: u64,
668 #[serde(skip_serializing_if = "Option::is_none")]
669 output: Option<String>,
670 output_bytes: u64,
671 duration_ms: u64,
672 status: String,
673 #[serde(skip_serializing_if = "Option::is_none")]
679 error: Option<String>,
680 #[serde(skip_serializing_if = "Option::is_none")]
684 state_effect: Option<String>,
685 #[serde(skip_serializing_if = "Option::is_none")]
690 resource_id: Option<String>,
691}
692
693#[derive(Serialize)]
694struct SkillCallEvent {
695 event_type: &'static str,
696 event_id: String,
697 parent_id: Option<String>,
698 session_id: String,
699 timestamp: u64,
700 turn: u32,
701 task_id: Option<String>,
702 skill_name: String,
703 output_bytes: u64,
704 duration_ms: u64,
705 status: String,
706 origin: &'static str,
708 trust: &'static str,
711}
712
713#[derive(Serialize)]
714struct ShellEvent {
715 event_type: &'static str,
716 event_id: String,
717 parent_id: Option<String>,
718 session_id: String,
719 timestamp: u64,
720 turn: u32,
721 task_id: Option<String>,
722 binary: String,
726 command: String,
727 exit_code: i32,
728 stdout_bytes: u64,
729 stderr_bytes: u64,
730 duration_ms: u64,
731 #[serde(skip_serializing_if = "Option::is_none")]
737 resource_limit: Option<String>,
738}
739
740#[derive(Serialize)]
743struct ShellDetachedEvent {
744 event_type: &'static str,
745 event_id: String,
746 parent_id: Option<String>,
747 session_id: String,
748 timestamp: u64,
749 turn: u32,
750 task_id: Option<String>,
751 work_id: String,
752 binary: String,
753 command: String,
754 grace_ms: u64,
756}
757
758#[derive(Serialize)]
761struct ShellCompletedEvent {
762 event_type: &'static str,
763 event_id: String,
764 parent_id: Option<String>,
765 session_id: String,
766 timestamp: u64,
767 work_id: String,
768 binary: String,
769 command: String,
770 exit_code: i32,
771 duration_ms: u64,
772 output_path: String,
775 output_bytes: u64,
776 #[serde(skip_serializing_if = "Option::is_none")]
778 resource_limit: Option<String>,
779 status: String,
782 #[serde(skip_serializing_if = "Option::is_none")]
786 completion_task_id: Option<String>,
787}
788
789#[derive(Serialize)]
791struct BackgroundCommandEvent<'a> {
792 event_type: &'static str,
793 event_id: String,
794 parent_id: Option<String>,
795 session_id: String,
796 timestamp: u64,
797 #[serde(skip_serializing_if = "Option::is_none")]
799 task_id: Option<String>,
800 work_id: &'a str,
801 operation: &'static str,
803 state: &'static str,
805 #[serde(skip_serializing_if = "Option::is_none")]
807 exit_code: Option<i32>,
808 #[serde(skip_serializing_if = "Option::is_none")]
810 waited_ms: Option<u64>,
811 output_bytes: u64,
813}
814
815#[derive(Serialize)]
824struct ShellAbandonedEvent {
825 event_type: &'static str,
826 event_id: String,
827 parent_id: Option<String>,
828 session_id: String,
829 timestamp: u64,
830 work_id: String,
831 binary: String,
832 command: String,
833 running_ms: u64,
836 #[serde(skip_serializing_if = "Option::is_none")]
839 exit_code: Option<i32>,
840 #[serde(skip_serializing_if = "Option::is_none")]
844 output_path: Option<String>,
845 #[serde(skip_serializing_if = "Option::is_none")]
847 output_bytes: Option<u64>,
848}
849
850#[derive(Serialize)]
862struct ShellLostEvent {
863 event_type: &'static str,
864 event_id: String,
865 #[serde(skip_serializing_if = "Option::is_none")]
868 parent_id: Option<String>,
869 session_id: String,
871 timestamp: u64,
872 work_id: String,
873 binary: String,
874 command: String,
875 detached_at_ms: u64,
877 reconciled_by_session: String,
879 reconciled_task_id: String,
882}
883
884#[derive(Serialize)]
890struct ShellDetachUnrecordedEvent {
891 event_type: &'static str,
892 event_id: String,
893 parent_id: Option<String>,
894 session_id: String,
895 timestamp: u64,
896 turn: u32,
897 task_id: Option<String>,
898 work_id: String,
899 binary: String,
900 reason: String,
902}
903
904#[derive(Serialize)]
905struct CompactionEvent {
906 event_type: &'static str,
907 event_id: String,
908 parent_id: Option<String>,
909 session_id: String,
910 timestamp: u64,
911 turn: u32,
912 task_id: Option<String>,
913 tokens_before: u64,
916 tokens_before_source: OccupancyBasis,
918 tokens_after: u64,
921}
922
923#[derive(Serialize)]
927struct CompactionDeclinedEvent {
928 event_type: &'static str,
929 event_id: String,
930 parent_id: Option<String>,
931 session_id: String,
932 timestamp: u64,
933 turn: u32,
934 task_id: Option<String>,
935 tokens: u64,
938 tokens_source: OccupancyBasis,
940 reason: String,
943}
944
945#[derive(Serialize)]
952struct ContextSeedEvent {
953 event_type: &'static str,
954 event_id: String,
955 parent_id: Option<String>,
956 session_id: String,
957 timestamp: u64,
958 task_id: Option<String>,
959 hook_name: String,
961 tokens: u64,
964 proposed_tokens: u64,
967 budget_tokens: u64,
970 outcome: String,
973 #[serde(skip_serializing_if = "Option::is_none")]
976 reason: Option<String>,
977 message_ids: Vec<String>,
981}
982
983#[derive(Serialize)]
991struct RetentionEvent {
992 event_type: &'static str,
993 event_id: String,
994 parent_id: Option<String>,
995 session_id: String,
996 timestamp: u64,
997 store: String,
999 reason: String,
1002 removed: u32,
1006 targets: Vec<String>,
1008 #[serde(skip_serializing_if = "Option::is_none")]
1011 messages_dropped: Option<u64>,
1012}
1013
1014#[derive(Serialize)]
1015struct A2aTaskReceivedEvent {
1016 event_type: &'static str,
1017 event_id: String,
1018 parent_id: Option<String>,
1019 session_id: String,
1020 timestamp: u64,
1021 task_id: String,
1022 context_id: String,
1023 message_id: String,
1024 traceparent_from_caller: Option<String>,
1025 #[serde(skip_serializing_if = "Option::is_none")]
1028 caller_member: Option<String>,
1029}
1030
1031#[derive(Serialize)]
1032struct A2aSendEvent {
1033 event_type: &'static str,
1034 event_id: String,
1035 parent_id: Option<String>,
1036 session_id: String,
1037 timestamp: u64,
1038 peer_url: String,
1039 message_id: String,
1040 task_id: String,
1041 context_id: String,
1042 traceparent: Option<String>,
1043 trust: String,
1046}
1047
1048#[derive(Serialize)]
1051struct ArtifactPulledEvent {
1052 event_type: &'static str,
1053 event_id: String,
1054 parent_id: Option<String>,
1055 session_id: String,
1056 timestamp: u64,
1057 name: String,
1058 version: String,
1059 runtime: &'static str,
1061 origin: &'static str,
1064 session: Option<String>,
1067 trust: &'static str,
1068 fetched_from: String,
1071 #[serde(skip_serializing_if = "Option::is_none")]
1074 owner_id: Option<u64>,
1075}
1076
1077#[derive(Serialize)]
1078struct SessionEndEvent {
1079 event_type: &'static str,
1080 event_id: String,
1081 parent_id: Option<String>,
1082 session_id: String,
1083 timestamp: u64,
1084 total_turns: u32,
1085 total_input_tokens: u64,
1086 total_output_tokens: u64,
1087 total_tool_calls: u32,
1088 total_shell_calls: u32,
1089 duration_ms: u64,
1090 exit_status: String,
1091}
1092
1093#[derive(Serialize)]
1094struct TaskStartEvent {
1095 event_type: &'static str,
1096 event_id: String,
1097 parent_id: Option<String>,
1098 session_id: String,
1099 timestamp: u64,
1100 task_id: String,
1101 context_id: String,
1102 source: String,
1105 origin: String,
1106 trust: String,
1107 lane: String,
1110 message_parts_bytes: u64,
1111 #[serde(skip_serializing_if = "Vec::is_empty")]
1114 part_kinds: Vec<&'static str>,
1115 #[serde(skip_serializing_if = "Vec::is_empty")]
1117 reference_task_ids: Vec<String>,
1118}
1119
1120#[derive(Serialize)]
1121struct TaskEndEvent {
1122 event_type: &'static str,
1123 event_id: String,
1124 parent_id: Option<String>,
1125 session_id: String,
1126 timestamp: u64,
1127 task_id: String,
1128 exit_status: String,
1129 duration_ms: u64,
1130 turns: u32,
1131 input_tokens: u64,
1132 output_tokens: u64,
1133 tool_calls: u32,
1134 shell_calls: u32,
1135 reopen_count: u32,
1138 #[serde(skip_serializing_if = "Vec::is_empty")]
1141 running_commands: Vec<String>,
1142 #[serde(skip_serializing_if = "Option::is_none")]
1144 keep_working: Option<KeepWorkingSummary>,
1145 #[serde(skip_serializing_if = "Option::is_none")]
1147 done_check: Option<DoneCheckSummary>,
1148}
1149
1150#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1153pub(crate) struct DoneCheckSummary {
1154 pub(crate) runs: u32,
1156 pub(crate) reopens: u32,
1158 pub(crate) limit: u32,
1160 pub(crate) met: Option<bool>,
1163 pub(crate) stopped_by: Option<&'static str>,
1166}
1167
1168#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1171pub(crate) struct KeepWorkingSummary {
1172 pub(crate) used: u32,
1174 pub(crate) limit: u32,
1176 pub(crate) stopped_by: Option<&'static str>,
1180}
1181
1182#[derive(Serialize)]
1189struct TaskCanceledEvent {
1190 event_type: &'static str,
1191 event_id: String,
1192 parent_id: Option<String>,
1193 session_id: String,
1194 timestamp: u64,
1195 task_id: String,
1196 #[serde(skip_serializing_if = "Option::is_none")]
1198 turn: Option<u32>,
1199 phase: String,
1202 detached_work_ids: Vec<String>,
1205 delegation_ids: Vec<String>,
1207}
1208
1209pub(crate) const TASK_REJECTED_SESSION_ENDED: &str = "session_ended";
1212
1213pub(crate) const TASK_REJECTED_SESSION_STOPPED: &str = "session_stopped";
1215
1216#[derive(Serialize)]
1223struct TaskRejectedEvent {
1224 event_type: &'static str,
1225 event_id: String,
1226 parent_id: Option<String>,
1227 session_id: String,
1228 timestamp: u64,
1229 task_id: String,
1230 context_id: String,
1231 source: String,
1234 cause: String,
1236 reason: String,
1238}
1239
1240pub(crate) const TASK_FAILED_DRIVER_ERROR: &str = "driver_error";
1243
1244pub(crate) const TASK_FAILED_CREDENTIAL_REJECTED: &str = "credential_rejected";
1247
1248pub(crate) const TASK_FAILED_MALFORMED_RESPONSE: &str = "malformed_response";
1251
1252pub(crate) const TASK_FAILED_COMPACTION_HOOK: &str = "compaction_hook";
1254
1255pub(crate) const TASK_FAILED_INPUT_TIMEOUT: &str = "input_timeout";
1257
1258pub(crate) const TASK_FAILED_REOPEN_BUDGET_EXHAUSTED: &str = "reopen_budget_exhausted";
1261
1262pub(crate) const TASK_FAILED_RUNTIME_ERROR: &str = "runtime_error";
1265
1266pub(crate) const TASK_FAILED_WORKDIR_PATH_REFUSED: &str = "workdir_path_refused";
1269
1270pub(crate) const INFERENCE_ERROR_CREDENTIAL_REJECTED: &str = TASK_FAILED_CREDENTIAL_REJECTED;
1274
1275pub(crate) const INFERENCE_ERROR_DRIVER_ERROR: &str = TASK_FAILED_DRIVER_ERROR;
1279
1280pub(crate) const INFERENCE_ERROR_DRIVER_FAILED: &str = "driver_failed";
1283
1284pub(crate) const INFERENCE_ERROR_MALFORMED_RESPONSE: &str = TASK_FAILED_MALFORMED_RESPONSE;
1287
1288pub(crate) const INFERENCE_ERROR_HARNESS_AUTH: &str = "harness_auth";
1290
1291pub(crate) const INFERENCE_ERROR_HARNESS_QUOTA: &str = "harness_quota";
1293
1294pub(crate) const INFERENCE_ERROR_HARNESS_ERROR: &str = "harness_error";
1297
1298pub(crate) const INFERENCE_ERROR_HARNESS_OTHER: &str = "harness_other";
1300
1301pub(crate) const INFERENCE_ERROR_HARNESS_INACTIVE: &str = "harness_inactive";
1303
1304pub(crate) struct FailedInference<'a> {
1306 pub(crate) turn: u32,
1307 pub(crate) error_code: &'static str,
1309 pub(crate) error: &'a str,
1311 pub(crate) provider_status: Option<u16>,
1313 pub(crate) usage: Option<&'a DriverUsage>,
1315 pub(crate) origin: Option<&'a InferenceOrigin>,
1317 pub(crate) choice: Option<&'a crate::driver_choice::DriverChoice>,
1319 pub(crate) message_ids: Vec<String>,
1321 pub(crate) tool_name: Option<String>,
1322 pub(crate) spent: Option<(Option<u64>, Option<u64>)>,
1325 pub(crate) settings: Option<&'a InferenceCallSettings>,
1327 pub(crate) retry: RetryNote,
1329}
1330
1331struct InferenceLine<'a> {
1334 turn: u32,
1335 input_tokens: Option<u64>,
1336 output_tokens: Option<u64>,
1337 decision: String,
1338 stop_reason: Option<&'a str>,
1339 tool_name: Option<String>,
1340 origin: Option<&'a InferenceOrigin>,
1341 usage: Option<&'a DriverUsage>,
1342 message_ids: Vec<String>,
1343 wire: Option<&'a WireCapture>,
1344 choice: Option<&'a crate::driver_choice::DriverChoice>,
1345 failure: Option<(&'static str, String, Option<u16>)>,
1347 counted: bool,
1349 settings: Option<&'a InferenceCallSettings>,
1351 retry: RetryNote,
1352}
1353
1354pub(crate) const MAX_TASK_FAILURE_REASON_BYTES: usize = 2_000;
1357
1358#[derive(Serialize)]
1362struct WorkdirPathRefusedEvent<'a> {
1363 event_type: &'static str,
1364 event_id: String,
1365 parent_id: Option<String>,
1366 session_id: String,
1367 timestamp: u64,
1368 #[serde(skip_serializing_if = "Option::is_none")]
1369 task_id: Option<String>,
1370 path: &'a str,
1372 operation: crate::workdir_fs::WorkdirOperation,
1373 refused_at: &'a str,
1376 reason: crate::workdir_fs::RefusalReason,
1377}
1378
1379#[derive(Serialize)]
1387struct TaskFailedEvent {
1388 event_type: &'static str,
1389 event_id: String,
1390 parent_id: Option<String>,
1391 session_id: String,
1392 timestamp: u64,
1393 #[serde(skip_serializing_if = "Option::is_none")]
1395 task_id: Option<String>,
1396 #[serde(skip_serializing_if = "Option::is_none")]
1399 turn: Option<u32>,
1400 cause: String,
1401 reason: String,
1403}
1404
1405pub(crate) const REOPEN_CONTEXT_CONTINUED: &str = "continued";
1408
1409pub(crate) const REOPEN_CONTEXT_RESTARTED: &str = "restarted";
1412
1413#[derive(Serialize)]
1418struct TaskReopenedEvent {
1419 event_type: &'static str,
1420 event_id: String,
1421 parent_id: Option<String>,
1422 session_id: String,
1423 timestamp: u64,
1424 task_id: String,
1425 hook_name: String,
1427 reason: String,
1429 reopen_number: u32,
1431 attempt_context: &'static str,
1433 turns_remaining: Option<u32>,
1437}
1438
1439#[derive(Serialize)]
1444struct TaskContinuedEvent<'a> {
1445 event_type: &'static str,
1446 event_id: String,
1447 parent_id: Option<String>,
1448 session_id: String,
1449 timestamp: u64,
1450 task_id: String,
1451 continuation_number: u32,
1453 member_calls: &'a [String],
1455 delegations: &'a [String],
1458 waited_ms: u64,
1460 turns_remaining: Option<u32>,
1464}
1465
1466#[derive(Serialize)]
1469struct KeepWorkingEvent {
1470 event_type: &'static str,
1471 event_id: String,
1472 parent_id: Option<String>,
1473 session_id: String,
1474 timestamp: u64,
1475 task_id: String,
1476 number: u32,
1478 limit: u32,
1480 turns_remaining: Option<u32>,
1483 message_bytes: u64,
1485}
1486
1487#[derive(Serialize)]
1491struct DoneCheckEvent<'a> {
1492 event_type: &'static str,
1493 event_id: String,
1494 parent_id: Option<String>,
1495 session_id: String,
1496 timestamp: u64,
1497 #[serde(flatten)]
1498 run: &'a DoneCheckRun<'a>,
1499}
1500
1501#[derive(Debug, Serialize)]
1503pub(crate) struct DoneCheckRun<'a> {
1504 pub(crate) task_id: &'a str,
1505 pub(crate) number: u32,
1507 pub(crate) command: &'a str,
1509 pub(crate) outcome: &'static str,
1511 pub(crate) exit_code: Option<i32>,
1513 pub(crate) duration_ms: u64,
1514 pub(crate) stdout_bytes: u64,
1516 pub(crate) stderr_bytes: u64,
1517 pub(crate) truncated: bool,
1518 pub(crate) resource_limit: Option<&'a str>,
1520 pub(crate) reopened: bool,
1522 pub(crate) reopens_left: u32,
1524 pub(crate) turns_remaining: Option<u32>,
1526}
1527
1528#[derive(Serialize)]
1531struct InferenceRetryEvent<'a> {
1532 event_type: &'static str,
1533 event_id: String,
1534 parent_id: Option<String>,
1535 session_id: String,
1536 timestamp: u64,
1537 turn: u32,
1538 task_id: Option<String>,
1539 attempt: u32,
1540 max_retries: u32,
1541 reason: &'static str,
1542 #[serde(skip_serializing_if = "Option::is_none")]
1543 provider_status: Option<u16>,
1544 error: &'a str,
1545 delay_ms: u64,
1546 #[serde(skip_serializing_if = "Option::is_none")]
1547 retry_after_ms: Option<u64>,
1548 charged_tokens: u64,
1549 #[serde(skip_serializing_if = "Option::is_none")]
1550 origin: Option<&'a str>,
1551}
1552
1553#[derive(Serialize)]
1556struct SpendCeilingReachedEvent<'a> {
1557 event_type: &'static str,
1558 event_id: String,
1559 parent_id: Option<String>,
1560 session_id: String,
1561 timestamp: u64,
1562 turn: u32,
1563 task_id: Option<String>,
1564 #[serde(flatten)]
1565 refusal: &'a crate::spend::SpendRefusal,
1566 #[serde(skip_serializing_if = "Option::is_none")]
1568 origin: Option<&'a str>,
1569}
1570
1571#[derive(Serialize)]
1576pub(crate) struct HarnessStart {
1577 pub driver: String,
1579 pub driver_version: String,
1580 pub harness: String,
1582 pub binary: String,
1584 pub binary_source: String,
1586 pub harness_version: Option<String>,
1588 pub version_tested: bool,
1590 pub harness_session_id: String,
1592 pub session_mode: String,
1593 pub args_count: usize,
1596 pub env_names: Vec<String>,
1598 pub account_env_unset: Vec<String>,
1601 pub files: Vec<String>,
1603 pub bridge_tools: Vec<String>,
1605 pub stdin_bytes: usize,
1607 pub keep_stdin_open: bool,
1608}
1609
1610#[derive(Serialize)]
1611struct HarnessStartEvent {
1612 event_type: &'static str,
1613 event_id: String,
1614 parent_id: Option<String>,
1615 session_id: String,
1616 timestamp: u64,
1617 task_id: Option<String>,
1618 #[serde(flatten)]
1619 start: HarnessStart,
1620}
1621
1622#[derive(Serialize)]
1625struct HarnessWarningEvent {
1626 event_type: &'static str,
1627 event_id: String,
1628 parent_id: Option<String>,
1629 session_id: String,
1630 timestamp: u64,
1631 task_id: Option<String>,
1632 code: String,
1633 message: String,
1634}
1635
1636#[derive(Serialize)]
1639struct HarnessSessionEvent {
1640 event_type: &'static str,
1641 event_id: String,
1642 parent_id: Option<String>,
1643 session_id: String,
1644 timestamp: u64,
1645 task_id: Option<String>,
1646 harness_session_id: String,
1647 auth: String,
1649 model: Option<String>,
1651}
1652
1653#[derive(Serialize)]
1657struct HarnessSessionForgottenEvent {
1658 event_type: &'static str,
1659 event_id: String,
1660 parent_id: Option<String>,
1661 session_id: String,
1662 timestamp: u64,
1663 task_id: Option<String>,
1664 context_id: String,
1666 harness_session_id: String,
1668 requested_by: String,
1670}
1671
1672#[derive(Serialize)]
1673struct HarnessRetryEvent {
1674 event_type: &'static str,
1675 event_id: String,
1676 parent_id: Option<String>,
1677 session_id: String,
1678 timestamp: u64,
1679 task_id: Option<String>,
1680 attempt: u32,
1681 reason: String,
1682}
1683
1684#[derive(Serialize)]
1685struct HarnessNoteEvent {
1686 event_type: &'static str,
1687 event_id: String,
1688 parent_id: Option<String>,
1689 session_id: String,
1690 timestamp: u64,
1691 task_id: Option<String>,
1692 text: String,
1693}
1694
1695#[derive(Serialize)]
1696struct HarnessFailedEvent {
1697 event_type: &'static str,
1698 event_id: String,
1699 parent_id: Option<String>,
1700 session_id: String,
1701 timestamp: u64,
1702 task_id: Option<String>,
1703 kind: String,
1705 message: String,
1706 source: String,
1708}
1709
1710#[derive(Serialize)]
1712struct HarnessInterruptEvent {
1713 event_type: &'static str,
1714 event_id: String,
1715 parent_id: Option<String>,
1716 session_id: String,
1717 timestamp: u64,
1718 task_id: Option<String>,
1719 method: String,
1721 delivered: bool,
1723 grace_ms: u64,
1726}
1727
1728#[derive(Serialize)]
1730pub(crate) struct HarnessExit {
1731 pub code: Option<i32>,
1733 pub signal: Option<i32>,
1735 pub cause: String,
1738 pub killed: bool,
1740 pub duration_ms: u64,
1742}
1743
1744#[derive(Serialize)]
1745struct HarnessExitEvent {
1746 event_type: &'static str,
1747 event_id: String,
1748 parent_id: Option<String>,
1749 session_id: String,
1750 timestamp: u64,
1751 task_id: Option<String>,
1752 #[serde(flatten)]
1753 exit: HarnessExit,
1754}
1755
1756#[derive(Serialize)]
1760struct CallDeniedEvent {
1761 event_type: &'static str,
1762 event_id: String,
1763 parent_id: Option<String>,
1764 session_id: String,
1765 timestamp: u64,
1766 turn: u32,
1767 event: String,
1770 hook_name: String,
1772 target: String,
1775 reason: String,
1777}
1778
1779#[derive(Serialize)]
1783struct ToolInputRefusedEvent {
1784 event_type: &'static str,
1785 event_id: String,
1786 parent_id: Option<String>,
1787 session_id: String,
1788 timestamp: u64,
1789 turn: u32,
1790 tool_name: String,
1791 tool_call_id: Option<String>,
1794 missing: Vec<String>,
1796 reason: String,
1798}
1799
1800#[derive(Serialize)]
1805struct ProtectedPathDeniedEvent {
1806 event_type: &'static str,
1807 event_id: String,
1808 parent_id: Option<String>,
1809 session_id: String,
1810 timestamp: u64,
1811 turn: u32,
1812 call: String,
1814 target: String,
1816 path: String,
1819 rule: String,
1821 signal: String,
1824 reason: String,
1826}
1827
1828#[derive(Serialize)]
1829struct HookDispatchErrorEvent {
1830 event_type: &'static str,
1831 event_id: String,
1832 parent_id: Option<String>,
1833 session_id: String,
1834 timestamp: u64,
1835 hook_name: String,
1836 event: String,
1837 arm: String,
1838}
1839
1840#[derive(Serialize)]
1844struct FormationEndedEvent {
1845 event_type: &'static str,
1846 event_id: String,
1847 parent_id: Option<String>,
1848 session_id: String,
1849 timestamp: u64,
1850 formation_id: String,
1851}
1852
1853#[derive(Serialize)]
1855struct SessionStatusEvent<'a> {
1856 event_type: &'static str,
1857 event_id: String,
1858 parent_id: Option<String>,
1859 session_id: String,
1860 timestamp: u64,
1861 #[serde(skip_serializing_if = "Option::is_none")]
1863 task_id: Option<String>,
1864 status: &'a crate::session_status::StatusReport,
1865}
1866
1867#[derive(Serialize)]
1869struct DeadlineNoticeEvent<'a> {
1870 event_type: &'static str,
1871 event_id: String,
1872 parent_id: Option<String>,
1873 session_id: String,
1874 timestamp: u64,
1875 #[serde(skip_serializing_if = "Option::is_none")]
1877 task_id: Option<String>,
1878 tool_name: &'a str,
1880 deadline_at: &'a str,
1882 left_secs: u64,
1884 notice_secs: u64,
1886}
1887
1888#[derive(Serialize)]
1891struct DeadlineReachedEvent {
1892 event_type: &'static str,
1893 event_id: String,
1894 parent_id: Option<String>,
1895 session_id: String,
1896 timestamp: u64,
1897 deadline_at: String,
1899 deadline_secs: u64,
1901 source: &'static str,
1903}
1904
1905#[derive(Serialize)]
1912struct SpawnerEndedEvent {
1913 event_type: &'static str,
1914 event_id: String,
1915 parent_id: Option<String>,
1916 session_id: String,
1917 timestamp: u64,
1918 #[serde(skip_serializing_if = "Option::is_none")]
1919 spawned_by: Option<String>,
1920 #[serde(skip_serializing_if = "Option::is_none")]
1921 delegation_id: Option<String>,
1922}
1923
1924#[derive(Serialize)]
1925struct ResourceListEvent {
1926 event_type: &'static str,
1927 event_id: String,
1928 parent_id: Option<String>,
1929 session_id: String,
1930 timestamp: u64,
1931 root: String,
1934 entry_count: usize,
1935 total_bytes: u64,
1936 generation: u64,
1938 containment_achieved: ContainmentClass,
1939 outcome: String,
1941 reason: Option<String>,
1943}
1944
1945#[derive(Serialize)]
1946struct ResourceReadEvent {
1947 event_type: &'static str,
1948 event_id: String,
1949 parent_id: Option<String>,
1950 session_id: String,
1951 timestamp: u64,
1952 path: String,
1955 outcome: String,
1956 bytes: Option<u64>,
1958 sha256: Option<String>,
1960 generation: u64,
1961 containment_achieved: ContainmentClass,
1962 reason: Option<String>,
1963}
1964
1965#[derive(Serialize)]
1967struct PeerHandleMintEvent {
1968 event_type: &'static str,
1969 event_id: String,
1970 parent_id: Option<String>,
1971 session_id: String,
1972 timestamp: u64,
1973 handle_id: Option<String>,
1976 path: String,
1979 audience: String,
1980 expires_at_ms: Option<u64>,
1982 outcome: String,
1983 reason: Option<String>,
1984}
1985
1986#[derive(Serialize)]
1989struct PeerHandleRedeemEvent {
1990 event_type: &'static str,
1991 event_id: String,
1992 parent_id: Option<String>,
1993 session_id: String,
1994 timestamp: u64,
1995 handle_id: String,
1996 path: Option<String>,
1999 generation: u64,
2002 audience_asserted: Option<String>,
2004 bytes: Option<u64>,
2006 sha256: Option<String>,
2008 outcome: String,
2009 reason: Option<String>,
2010}
2011
2012#[derive(Serialize)]
2014struct PeerFileFetchEvent {
2015 event_type: &'static str,
2016 event_id: String,
2017 parent_id: Option<String>,
2018 session_id: String,
2019 timestamp: u64,
2020 peer: String,
2021 handle_id: String,
2022 stored_path: Option<String>,
2025 bytes: Option<u64>,
2026 sha256: Option<String>,
2027 outcome: String,
2028 reason: Option<String>,
2029}
2030
2031#[derive(Serialize)]
2042struct DelegationStartEvent {
2043 event_type: &'static str,
2044 event_id: String,
2045 parent_id: Option<String>,
2046 session_id: String,
2047 timestamp: u64,
2048 delegation_id: String,
2051 capsule: String,
2052 version: String,
2053 child_session_id: String,
2055 child_workdir: String,
2057 child_records_dir: String,
2060}
2061
2062impl DelegationStartEvent {
2063 #[allow(clippy::too_many_arguments)]
2066 fn new(
2067 session_id: &str,
2068 session_event_id: &str,
2069 delegation_id: &str,
2070 capsule: &str,
2071 version: &str,
2072 child_session_id: &str,
2073 child_workdir: &str,
2074 child_records_dir: &str,
2075 ) -> Self {
2076 Self {
2077 event_type: "delegation_start",
2078 event_id: new_event_id(),
2079 parent_id: Some(session_event_id.to_string()),
2080 session_id: session_id.to_string(),
2081 timestamp: timestamp_ms(),
2082 delegation_id: delegation_id.to_string(),
2083 capsule: capsule.to_string(),
2084 version: version.to_string(),
2085 child_session_id: child_session_id.to_string(),
2086 child_workdir: child_workdir.to_string(),
2087 child_records_dir: child_records_dir.to_string(),
2088 }
2089 }
2090}
2091
2092#[derive(Serialize)]
2098struct MemberCallStartEvent {
2099 event_type: &'static str,
2100 event_id: String,
2101 parent_id: Option<String>,
2102 session_id: String,
2103 timestamp: u64,
2104 task_id: String,
2106 call_id: String,
2108 member: String,
2110 member_task_id: String,
2112}
2113
2114#[derive(Serialize)]
2120struct MemberCallBusyEvent {
2121 event_type: &'static str,
2122 event_id: String,
2123 parent_id: Option<String>,
2124 session_id: String,
2125 timestamp: u64,
2126 task_id: String,
2128 call_id: String,
2130 member: String,
2132 offer: u32,
2134 waited_ms: u64,
2136 message: String,
2138}
2139
2140#[derive(Serialize)]
2146struct MemberCallEvent {
2147 event_type: &'static str,
2148 event_id: String,
2149 parent_id: Option<String>,
2150 session_id: String,
2151 timestamp: u64,
2152 task_id: String,
2153 call_id: String,
2154 member: String,
2155 #[serde(skip_serializing_if = "Option::is_none")]
2157 member_task_id: Option<String>,
2158 status: String,
2161 duration_ms: u64,
2162 output: String,
2164 truncated: bool,
2165 delivered: bool,
2168 #[serde(skip_serializing_if = "Vec::is_empty")]
2171 no_answer_below: Vec<NoAnswerBelow>,
2172}
2173
2174#[derive(Serialize)]
2176struct NoAnswerBelow {
2177 member: String,
2178 status: String,
2180}
2181
2182#[derive(Serialize)]
2193struct DelegationEvent {
2194 event_type: &'static str,
2195 event_id: String,
2196 parent_id: Option<String>,
2197 session_id: String,
2198 timestamp: u64,
2199 capsule: String,
2201 version: String,
2202 delegation_id: Option<String>,
2204 child_session_id: Option<String>,
2206 duration_ms: u64,
2209 outcome: String,
2219 reason: Option<String>,
2222}
2223
2224impl DelegationEvent {
2225 #[allow(clippy::too_many_arguments)]
2228 fn new(
2229 session_id: &str,
2230 session_event_id: &str,
2231 capsule: &str,
2232 version: &str,
2233 delegation_id: Option<String>,
2234 child_session_id: Option<String>,
2235 duration_ms: u64,
2236 outcome: &str,
2237 reason: Option<String>,
2238 ) -> Self {
2239 Self {
2240 event_type: "delegation",
2241 event_id: new_event_id(),
2242 parent_id: Some(session_event_id.to_string()),
2243 session_id: session_id.to_string(),
2244 timestamp: timestamp_ms(),
2245 capsule: capsule.to_string(),
2246 version: version.to_string(),
2247 delegation_id,
2248 child_session_id,
2249 duration_ms,
2250 outcome: outcome.to_string(),
2251 reason,
2252 }
2253 }
2254}
2255
2256impl TraceWriter {
2259 #[allow(clippy::too_many_arguments)]
2260 pub(crate) async fn open(
2261 records_root: &WorkdirRoot,
2262 session_id: String,
2263 capsule_name: String,
2264 capsule_version: String,
2265 model: String,
2266 capabilities: Vec<String>,
2267 effective_grants: ScopeReport,
2268 capture: TraceCapture,
2269 system_prompt: Option<String>,
2270 system_prompt_overridden: bool,
2271 resumed_from: Option<String>,
2272 context_id: Option<String>,
2273 spawned_by: Option<String>,
2274 spawn_delegation_id: Option<String>,
2275 ) -> std::io::Result<Self> {
2276 let file = File::from_std(records_root.open_append(TRACE_FILE_NAME, DEFAULT_FILE_MODE)?);
2277 let system_prompt_source = match (system_prompt_overridden, system_prompt.is_some()) {
2280 (true, _) => "cli",
2281 (false, true) => "manifest",
2282 (false, false) => "none",
2283 };
2284 let system_prompt_sha256 = system_prompt
2285 .as_ref()
2286 .map(|prompt| murmur_artifact::sha256_hex(prompt.as_bytes()));
2287 let blobs = BlobStore::new(records_root);
2288 if let (true, Some(prompt), Some(sha)) = (
2292 capture.captures_content(),
2293 system_prompt.as_ref(),
2294 system_prompt_sha256.as_deref(),
2295 ) {
2296 blobs.put(sha, prompt.as_bytes()).await?;
2297 }
2298 Ok(Self {
2299 writer: BufWriter::new(file),
2300 session_id,
2301 capsule_name,
2302 capsule_version,
2303 model,
2304 capabilities,
2305 effective_grants,
2306 capture,
2307 blobs,
2308 system_prompt_source,
2309 system_prompt_sha256,
2310 credential_source: "none",
2311 gateways: Vec::new(),
2312 control: None,
2313 inference_choices: Vec::new(),
2314 tool_refresh: None,
2315 runtime_artifacts: Vec::new(),
2316 host_facts: Vec::new(),
2317 resumed_from,
2318 context_id,
2319 spawned_by,
2320 spawn_delegation_id,
2321 formation_id: None,
2322 formation_member: None,
2323 session_start_time: Instant::now(),
2324 session_event_id: new_event_id(),
2325 session_started: false,
2326 session_ended: false,
2327 task_event_id: None,
2328 turn_event_id: None,
2329 total_turns: 0,
2330 total_input_tokens: 0,
2331 total_output_tokens: 0,
2332 total_tool_calls: 0,
2333 total_shell_calls: 0,
2334 task_turns: 0,
2335 task_input_tokens: 0,
2336 task_output_tokens: 0,
2337 task_tool_calls: 0,
2338 task_shell_calls: 0,
2339 task_agent_turns: 0,
2340 task_hook_inferences: 0,
2341 task_start_instant: None,
2342 active_task_id: None,
2343 task_failures_written: 0,
2344 last_task_failure: None,
2345 max_session_tokens: None,
2346 machine_tokens_per_day: None,
2347 session_status: None,
2348 })
2349 }
2350
2351 pub(crate) fn set_session_status(
2354 &mut self,
2355 status: std::sync::Arc<crate::session_status::SessionStatus>,
2356 ) {
2357 self.session_status = Some(status);
2358 }
2359
2360 pub(crate) fn set_spend_ceilings(
2362 &mut self,
2363 max_session_tokens: Option<u64>,
2364 machine_tokens_per_day: Option<u64>,
2365 ) {
2366 self.max_session_tokens = max_session_tokens;
2367 self.machine_tokens_per_day = machine_tokens_per_day;
2368 }
2369
2370 pub(crate) fn session_event_id(&self) -> &str {
2374 &self.session_event_id
2375 }
2376
2377 pub(crate) fn captures_wire(&self) -> bool {
2381 self.capture.captures_hashes()
2382 }
2383
2384 fn session_parent(&self) -> Option<String> {
2389 self.session_started.then(|| self.session_event_id.clone())
2390 }
2391
2392 fn task_parent(&self) -> Option<String> {
2395 self.task_event_id.clone().or_else(|| self.session_parent())
2396 }
2397
2398 fn turn_parent(&self) -> Option<String> {
2402 self.turn_event_id.clone().or_else(|| self.task_parent())
2403 }
2404
2405 pub(crate) fn set_credential_source(&mut self, source: &'static str) {
2408 self.credential_source = source;
2409 }
2410
2411 pub(crate) fn set_gateways(&mut self, gateways: Vec<SessionGateway>) {
2413 self.gateways = gateways;
2414 }
2415
2416 pub(crate) fn set_control(&mut self, control: Option<SessionControl>) {
2418 self.control = control;
2419 }
2420
2421 pub(crate) fn set_inference_choices(&mut self, choices: Vec<SessionInferenceChoice>) {
2424 self.inference_choices = choices;
2425 }
2426
2427 pub(crate) fn set_runtime_artifacts(&mut self, installed: &[InstalledArtifactSummary]) {
2430 self.runtime_artifacts = installed
2431 .iter()
2432 .filter_map(|artifact| match &artifact.origin {
2433 LockOrigin::Runtime { session } => Some(RuntimeArtifact {
2434 name: artifact.name.clone(),
2435 version: artifact.version.clone(),
2436 origin: artifact.origin.as_str(),
2437 session: session.clone(),
2438 trust: crate::origin::artifact_trust(&artifact.origin).as_str(),
2439 }),
2440 LockOrigin::Operator => None,
2441 })
2442 .collect();
2443 }
2444
2445 pub(crate) fn set_host_facts(&mut self, uses: &[crate::host_facts::HostFactsUse]) {
2447 self.host_facts = uses.iter().map(HostFactsRecord::of).collect();
2448 }
2449
2450 pub(crate) async fn write_control_applied(
2457 &mut self,
2458 turn: u32,
2459 name: &str,
2460 value: Value,
2461 change_id: &str,
2462 ) -> std::io::Result<()> {
2463 let event = ControlAppliedEvent {
2464 event_type: "control_applied",
2465 event_id: new_event_id(),
2466 parent_id: self.task_parent(),
2467 session_id: self.session_id.clone(),
2468 timestamp: timestamp_ms(),
2469 turn,
2470 task_id: self.active_task_id.clone(),
2471 name,
2472 value,
2473 change_id,
2474 };
2475 self.write_event(&event).await
2476 }
2477
2478 pub(crate) fn set_formation_id(&mut self, formation_id: Option<&FormationId>) {
2480 self.formation_id = formation_id.cloned();
2481 }
2482
2483 pub(crate) fn set_formation_member(
2486 &mut self,
2487 member: Option<&crate::formation_credentials::FormationMember>,
2488 ) {
2489 self.formation_member = member.map(|member| {
2490 (
2491 member.name().to_string(),
2492 member.callees().map(str::to_string).collect(),
2493 )
2494 });
2495 }
2496
2497 pub(crate) fn set_tool_refresh(&mut self, tool_refresh: Option<&'static str>) {
2500 self.tool_refresh = tool_refresh;
2501 }
2502
2503 pub(crate) async fn write_tools_refreshed(
2511 &mut self,
2512 turn: u32,
2513 trigger: &str,
2514 added: &[String],
2515 removed: &[String],
2516 tools: &[String],
2517 ) -> std::io::Result<()> {
2518 let event = ToolsRefreshedEvent {
2519 event_type: "tools_refreshed",
2520 event_id: new_event_id(),
2521 parent_id: self.task_parent(),
2522 session_id: self.session_id.clone(),
2523 timestamp: timestamp_ms(),
2524 turn,
2525 task_id: self.active_task_id.clone(),
2526 trigger,
2527 added,
2528 removed,
2529 tools,
2530 };
2531 self.write_event(&event).await
2532 }
2533
2534 pub(crate) async fn write_session_start(
2535 &mut self,
2536 max_turns: Option<u32>,
2537 model_limits: Option<crate::model_limits::ResolvedModelLimits>,
2538 tools_declared: Vec<String>,
2539 ) -> std::io::Result<()> {
2540 let event = SessionStartEvent {
2541 event_type: "session_start",
2542 event_id: self.session_event_id.clone(),
2543 parent_id: None,
2544 session_id: self.session_id.clone(),
2545 timestamp: timestamp_ms(),
2546 capsule_name: self.capsule_name.clone(),
2547 capsule_version: self.capsule_version.clone(),
2548 model: self.model.clone(),
2549 max_turns,
2550 model_limits,
2551 max_session_tokens: self.max_session_tokens,
2552 machine_tokens_per_day: self.machine_tokens_per_day,
2553 deadline: self.session_status.as_ref().and_then(|status| {
2554 status.deadline().map(|deadline| SessionStartDeadline {
2555 at: deadline.at_rfc3339(),
2556 secs: deadline.secs(),
2557 notice_secs: status.notice_secs(),
2558 source: deadline.source().as_str(),
2559 })
2560 }),
2561 capabilities: self.capabilities.clone(),
2562 tools_declared,
2563 containment_declared: self.effective_grants.declared_containment,
2564 containment_achieved: self.effective_grants.achieved_containment,
2565 workdir_exec: self.effective_grants.workdir_exec,
2566 userns_grant: self.effective_grants.userns_grant,
2567 effective_grants: self.effective_grants.clone(),
2568 system_prompt_source: self.system_prompt_source,
2569 system_prompt_sha256: self.system_prompt_sha256.clone(),
2570 credential_source: self.credential_source,
2571 gateways: self.gateways.clone(),
2572 control: self.control.clone(),
2573 inference_choices: self.inference_choices.clone(),
2574 tool_refresh: self.tool_refresh,
2575 runtime_artifacts: self.runtime_artifacts.clone(),
2576 host_facts: self.host_facts.clone(),
2577 resumed_from: self.resumed_from.clone(),
2578 context_id: self.context_id.clone(),
2579 spawned_by: self.spawned_by.clone(),
2580 delegation_id: self.spawn_delegation_id.clone(),
2581 formation_id: self.formation_id.clone(),
2582 formation_member: self.formation_member.as_ref().map(|(name, _)| name.clone()),
2583 formation_callees: self
2584 .formation_member
2585 .as_ref()
2586 .map(|(_, callees)| callees.clone()),
2587 };
2588 self.write_event(&event).await?;
2589 self.session_started = true;
2590 Ok(())
2591 }
2592
2593 #[allow(clippy::too_many_arguments)]
2606 pub(crate) async fn write_inference(
2607 &mut self,
2608 turn: u32,
2609 input_tokens: Option<u64>,
2610 output_tokens: Option<u64>,
2611 decision: String,
2612 stop_reason: Option<&str>,
2613 tool_name: Option<String>,
2614 origin: Option<&InferenceOrigin>,
2615 usage: Option<&DriverUsage>,
2616 message_ids: Vec<String>,
2617 wire: Option<&WireCapture>,
2618 settings: Option<&InferenceCallSettings>,
2619 ) -> std::io::Result<()> {
2620 self.write_inference_event(InferenceLine {
2621 turn,
2622 input_tokens,
2623 output_tokens,
2624 decision,
2625 stop_reason,
2626 tool_name,
2627 origin,
2628 usage,
2629 message_ids,
2630 wire,
2631 choice: None,
2632 failure: None,
2633 counted: true,
2634 settings,
2635 retry: RetryNote::default(),
2636 })
2637 .await
2638 }
2639
2640 #[allow(clippy::too_many_arguments)]
2643 pub(crate) async fn write_hook_inference(
2644 &mut self,
2645 turn: u32,
2646 input_tokens: Option<u64>,
2647 output_tokens: Option<u64>,
2648 decision: String,
2649 origin: &InferenceOrigin,
2650 usage: Option<&DriverUsage>,
2651 settings: Option<&InferenceCallSettings>,
2652 retry: RetryNote,
2653 ) -> std::io::Result<()> {
2654 self.write_inference_event(InferenceLine {
2655 turn,
2656 input_tokens,
2657 output_tokens,
2658 decision,
2659 stop_reason: None,
2660 tool_name: None,
2661 origin: Some(origin),
2662 usage,
2663 message_ids: Vec::new(),
2664 wire: None,
2665 choice: None,
2666 failure: None,
2667 counted: true,
2668 settings,
2669 retry,
2670 })
2671 .await
2672 }
2673
2674 #[allow(clippy::too_many_arguments)]
2678 pub(crate) async fn write_turn_inference(
2679 &mut self,
2680 turn: u32,
2681 input_tokens: Option<u64>,
2682 output_tokens: Option<u64>,
2683 decision: String,
2684 stop_reason: Option<&str>,
2685 tool_name: Option<String>,
2686 usage: Option<&DriverUsage>,
2687 message_ids: Vec<String>,
2688 wire: Option<&WireCapture>,
2689 choice: Option<&crate::driver_choice::DriverChoice>,
2690 settings: Option<&InferenceCallSettings>,
2691 retry: RetryNote,
2692 ) -> std::io::Result<()> {
2693 self.write_inference_event(InferenceLine {
2694 turn,
2695 input_tokens,
2696 output_tokens,
2697 decision,
2698 stop_reason,
2699 tool_name,
2700 origin: None,
2701 usage,
2702 message_ids,
2703 wire,
2704 choice,
2705 failure: None,
2706 counted: true,
2707 settings,
2708 retry,
2709 })
2710 .await
2711 }
2712
2713 pub(crate) async fn write_failed_inference(
2721 &mut self,
2722 failed: FailedInference<'_>,
2723 ) -> std::io::Result<()> {
2724 let error =
2725 crate::agent::truncate_on_char_boundary(failed.error, MAX_TASK_FAILURE_REASON_BYTES)
2726 .to_string();
2727 let (input_tokens, output_tokens) = failed.spent.unwrap_or((None, None));
2728 self.write_inference_event(InferenceLine {
2729 turn: failed.turn,
2730 input_tokens,
2731 output_tokens,
2732 decision: "error".to_string(),
2733 stop_reason: Some("error"),
2734 tool_name: failed.tool_name,
2735 origin: failed.origin,
2736 usage: failed.usage,
2737 message_ids: failed.message_ids,
2738 wire: None,
2739 choice: failed.choice,
2740 failure: Some((failed.error_code, error, failed.provider_status)),
2741 counted: failed.spent.is_some(),
2742 settings: failed.settings,
2743 retry: failed.retry,
2744 })
2745 .await
2746 }
2747
2748 async fn write_inference_event(&mut self, line: InferenceLine<'_>) -> std::io::Result<()> {
2749 let InferenceLine {
2750 turn,
2751 input_tokens,
2752 output_tokens,
2753 decision,
2754 stop_reason,
2755 tool_name,
2756 origin,
2757 usage,
2758 message_ids,
2759 wire,
2760 choice,
2761 failure,
2762 counted,
2763 settings,
2764 retry,
2765 } = line;
2766 let (error_code, error, provider_status) = match failure {
2767 Some((code, error, status)) => (Some(code.to_string()), Some(error), status),
2768 None => (None, None, None),
2769 };
2770 let event_id = new_event_id();
2771 let parent_id = if origin.is_none() {
2775 let parent = self.task_parent();
2776 self.turn_event_id = Some(event_id.clone());
2777 parent
2778 } else {
2779 self.turn_parent()
2780 };
2781 let (system_sha, tools_sha, response_sha, message_shas) =
2782 match wire.filter(|_| self.capture.captures_hashes()) {
2783 Some(wire) => self.record_wire(wire).await?,
2784 None => (None, None, None, Vec::new()),
2785 };
2786 let event = InferenceEvent {
2787 event_type: "inference",
2788 event_id,
2789 parent_id,
2790 session_id: self.session_id.clone(),
2791 timestamp: timestamp_ms(),
2792 turn,
2793 task_id: self.active_task_id.clone(),
2794 input_tokens,
2795 output_tokens,
2796 decision,
2797 stop_reason: stop_reason.map(str::to_string),
2798 tool_name,
2799 input_tokens_actual: usage.and_then(|u| u.input_tokens),
2800 output_tokens_actual: usage.and_then(|u| u.output_tokens),
2801 cached_tokens: usage.and_then(|u| u.cached_tokens),
2802 cache_write_tokens: usage.and_then(|u| u.cache_write_tokens),
2803 thinking_tokens: usage.and_then(|u| u.thinking_tokens),
2804 origin: origin.map(|o| o.source.clone()),
2805 driver_choice: choice.map(|choice| choice.name.clone()),
2806 model: origin
2807 .map(|o| o.model.clone())
2808 .or_else(|| choice.map(|choice| choice.model.clone())),
2809 max_output_tokens: settings.map(|settings| settings.max_output_tokens),
2810 input_budget: settings.map(|settings| settings.input_budget),
2811 driver_config: settings.map(|settings| settings.driver_config.clone()),
2812 message_ids,
2813 system_sha,
2814 tools_sha,
2815 response_sha,
2816 message_shas,
2817 error_code,
2818 error,
2819 provider_status,
2820 attempts: retry.attempts,
2821 retry_stopped: retry.stopped.map(RetryStop::wire_name),
2822 };
2823 self.write_event(&event).await?;
2824 if origin.is_none() {
2825 self.task_agent_turns = self.task_agent_turns.saturating_add(1);
2826 }
2827 if counted {
2828 let (input, output) = (input_tokens.unwrap_or(0), output_tokens.unwrap_or(0));
2832 self.total_turns = self.total_turns.saturating_add(1);
2833 self.total_input_tokens = self.total_input_tokens.saturating_add(input);
2834 self.total_output_tokens = self.total_output_tokens.saturating_add(output);
2835 self.task_turns = self.task_turns.saturating_add(1);
2836 if origin.is_some() {
2837 self.task_hook_inferences = self.task_hook_inferences.saturating_add(1);
2838 }
2839 self.task_input_tokens = self.task_input_tokens.saturating_add(input);
2840 self.task_output_tokens = self.task_output_tokens.saturating_add(output);
2841 }
2842 if let Some(status) = &self.session_status {
2845 status.record_inference(self.task_turns, input_tokens, output_tokens, usage);
2846 }
2847 Ok(())
2848 }
2849
2850 async fn record_wire(
2857 &mut self,
2858 wire: &WireCapture,
2859 ) -> std::io::Result<(Option<String>, Option<String>, Option<String>, Vec<String>)> {
2860 let system_sha = murmur_artifact::sha256_hex(&wire.system);
2861 let tools_sha = murmur_artifact::sha256_hex(&wire.tools);
2862 let response_sha = murmur_artifact::sha256_hex(&wire.response);
2863 let message_shas: Vec<String> = wire
2864 .messages
2865 .iter()
2866 .map(|message| murmur_artifact::sha256_hex(message))
2867 .collect();
2868
2869 if self.capture.captures_content() {
2870 self.blobs.put(&system_sha, &wire.system).await?;
2874 self.blobs.put(&tools_sha, &wire.tools).await?;
2875 self.blobs.put(&response_sha, &wire.response).await?;
2876 for (sha, body) in message_shas.iter().zip(&wire.messages) {
2877 self.blobs.put(sha, body).await?;
2878 }
2879 }
2880
2881 Ok((
2882 Some(system_sha),
2883 Some(tools_sha),
2884 Some(response_sha),
2885 message_shas,
2886 ))
2887 }
2888
2889 #[allow(clippy::too_many_arguments)]
2890 pub(crate) async fn write_tool_call(
2891 &mut self,
2892 turn: u32,
2893 tool_name: String,
2894 tool_call_id: Option<String>,
2895 mut input: Value,
2896 input_bytes: u64,
2897 output: &str,
2898 output_bytes: u64,
2899 duration_ms: u64,
2900 status: String,
2901 state_effect: Option<String>,
2902 resource_id: Option<String>,
2903 error: Option<&str>,
2904 ) -> std::io::Result<()> {
2905 crate::peer_handoff::redact_handles_in_json(&mut input);
2910 let event = ToolCallEvent {
2911 event_type: "tool_call",
2912 event_id: new_event_id(),
2913 parent_id: self.turn_parent(),
2914 session_id: self.session_id.clone(),
2915 timestamp: timestamp_ms(),
2916 turn,
2917 task_id: self.active_task_id.clone(),
2918 tool_name,
2919 tool_call_id: tool_call_id.filter(|id| !id.is_empty()),
2922 input,
2923 input_bytes,
2924 output: self
2925 .capture
2926 .captures_content()
2927 .then(|| crate::peer_handoff::redact_handle_tokens(output).into_owned()),
2928 output_bytes,
2929 duration_ms,
2930 status,
2931 error: error.map(|error| {
2932 let redacted = crate::peer_handoff::redact_handle_tokens(error);
2933 crate::agent::truncate_on_char_boundary(&redacted, MAX_TASK_FAILURE_REASON_BYTES)
2934 .to_string()
2935 }),
2936 state_effect,
2937 resource_id,
2938 };
2939 self.write_event(&event).await?;
2940 self.total_tool_calls = self.total_tool_calls.saturating_add(1);
2941 self.task_tool_calls = self.task_tool_calls.saturating_add(1);
2942 Ok(())
2943 }
2944
2945 pub(crate) async fn write_skill_call(
2951 &mut self,
2952 turn: u32,
2953 skill_name: String,
2954 origin: &LockOrigin,
2955 output_bytes: u64,
2956 duration_ms: u64,
2957 status: String,
2958 ) -> std::io::Result<()> {
2959 let event = SkillCallEvent {
2960 event_type: "skill_call",
2961 event_id: new_event_id(),
2962 parent_id: self.turn_parent(),
2963 session_id: self.session_id.clone(),
2964 timestamp: timestamp_ms(),
2965 turn,
2966 task_id: self.active_task_id.clone(),
2967 skill_name,
2968 output_bytes,
2969 duration_ms,
2970 status,
2971 origin: origin.as_str(),
2972 trust: crate::origin::artifact_trust(origin).as_str(),
2973 };
2974 self.write_event(&event).await
2975 }
2977
2978 #[allow(clippy::too_many_arguments)]
2981 pub(crate) async fn write_shell(
2982 &mut self,
2983 turn: u32,
2984 binary: String,
2985 command: String,
2986 exit_code: i32,
2987 stdout_bytes: u64,
2988 stderr_bytes: u64,
2989 duration_ms: u64,
2990 resource_limit: Option<String>,
2991 ) -> std::io::Result<()> {
2992 let event = ShellEvent {
2993 event_type: "shell",
2994 event_id: new_event_id(),
2995 parent_id: self.turn_parent(),
2996 session_id: self.session_id.clone(),
2997 timestamp: timestamp_ms(),
2998 turn,
2999 task_id: self.active_task_id.clone(),
3000 binary,
3001 command,
3002 exit_code,
3003 stdout_bytes,
3004 stderr_bytes,
3005 duration_ms,
3006 resource_limit,
3007 };
3008 self.write_event(&event).await?;
3009 self.total_shell_calls = self.total_shell_calls.saturating_add(1);
3010 self.task_shell_calls = self.task_shell_calls.saturating_add(1);
3011 Ok(())
3012 }
3013
3014 pub(crate) async fn write_shell_detached(
3021 &mut self,
3022 turn: u32,
3023 work_id: &str,
3024 binary: &str,
3025 command: &str,
3026 grace_ms: u64,
3027 ) -> std::io::Result<()> {
3028 let event = ShellDetachedEvent {
3029 event_type: "shell_detached",
3030 event_id: new_event_id(),
3031 parent_id: self.turn_parent(),
3032 session_id: self.session_id.clone(),
3033 timestamp: timestamp_ms(),
3034 turn,
3035 task_id: self.active_task_id.clone(),
3036 work_id: work_id.to_string(),
3037 binary: binary.to_string(),
3038 command: command.to_string(),
3039 grace_ms,
3040 };
3041 self.write_event(&event).await?;
3042 self.total_shell_calls = self.total_shell_calls.saturating_add(1);
3043 self.task_shell_calls = self.task_shell_calls.saturating_add(1);
3044 Ok(())
3045 }
3046
3047 #[allow(clippy::too_many_arguments)]
3052 pub(crate) async fn write_shell_completed(
3053 &mut self,
3054 work_id: &str,
3055 binary: &str,
3056 command: &str,
3057 exit_code: i32,
3058 duration_ms: u64,
3059 output_path: &str,
3060 output_bytes: u64,
3061 resource_limit: Option<String>,
3062 status: &str,
3063 completion_task_id: Option<&str>,
3064 ) -> std::io::Result<()> {
3065 let event = ShellCompletedEvent {
3066 event_type: "shell_completed",
3067 event_id: new_event_id(),
3068 parent_id: self.session_parent(),
3069 session_id: self.session_id.clone(),
3070 timestamp: timestamp_ms(),
3071 work_id: work_id.to_string(),
3072 binary: binary.to_string(),
3073 command: command.to_string(),
3074 exit_code,
3075 duration_ms,
3076 output_path: output_path.to_string(),
3077 output_bytes,
3078 resource_limit,
3079 status: status.to_string(),
3080 completion_task_id: completion_task_id.map(str::to_string),
3081 };
3082 self.write_event(&event).await
3083 }
3084
3085 pub(crate) async fn write_shell_abandoned(
3092 &mut self,
3093 work_id: &str,
3094 binary: &str,
3095 command: &str,
3096 running_ms: u64,
3097 recovered: Option<(i32, &str, u64)>,
3098 ) -> std::io::Result<()> {
3099 let event = ShellAbandonedEvent {
3100 event_type: "shell_abandoned",
3101 event_id: new_event_id(),
3102 parent_id: self.session_parent(),
3103 session_id: self.session_id.clone(),
3104 timestamp: timestamp_ms(),
3105 work_id: work_id.to_string(),
3106 binary: binary.to_string(),
3107 command: command.to_string(),
3108 running_ms,
3109 exit_code: recovered.map(|(exit_code, _, _)| exit_code),
3110 output_path: recovered.map(|(_, path, _)| path.to_string()),
3111 output_bytes: recovered.map(|(_, _, bytes)| bytes),
3112 };
3113 self.write_event(&event).await
3114 }
3115
3116 pub(crate) async fn write_shell_detach_unrecorded(
3118 &mut self,
3119 turn: u32,
3120 work_id: &str,
3121 binary: &str,
3122 reason: &str,
3123 ) -> std::io::Result<()> {
3124 let event = ShellDetachUnrecordedEvent {
3125 event_type: "shell_detach_unrecorded",
3126 event_id: new_event_id(),
3127 parent_id: self.turn_parent(),
3128 session_id: self.session_id.clone(),
3129 timestamp: timestamp_ms(),
3130 turn,
3131 task_id: self.active_task_id.clone(),
3132 work_id: work_id.to_string(),
3133 binary: binary.to_string(),
3134 reason: reason.to_string(),
3135 };
3136 self.write_event(&event).await
3137 }
3138
3139 pub(crate) async fn write_a2a_task_received(
3140 &mut self,
3141 task_id: &str,
3142 context_id: &str,
3143 message_id: &str,
3144 traceparent_from_caller: Option<&str>,
3145 caller_member: Option<&str>,
3146 ) -> std::io::Result<()> {
3147 let event = A2aTaskReceivedEvent {
3148 event_type: "a2a_task_received",
3149 event_id: new_event_id(),
3150 parent_id: self.session_parent(),
3151 session_id: self.session_id.clone(),
3152 timestamp: timestamp_ms(),
3153 task_id: task_id.to_string(),
3154 context_id: context_id.to_string(),
3155 message_id: message_id.to_string(),
3156 traceparent_from_caller: traceparent_from_caller.map(str::to_string),
3157 caller_member: caller_member.map(str::to_string),
3158 };
3159 self.write_event(&event).await
3160 }
3161
3162 pub(crate) async fn write_a2a_send(
3163 &mut self,
3164 peer_url: &str,
3165 message_id: &str,
3166 task_id: &str,
3167 context_id: &str,
3168 traceparent: Option<&str>,
3169 trust: TrustClass,
3170 ) -> std::io::Result<()> {
3171 let event = A2aSendEvent {
3172 event_type: "a2a_send",
3173 event_id: new_event_id(),
3174 parent_id: self.session_parent(),
3175 session_id: self.session_id.clone(),
3176 timestamp: timestamp_ms(),
3177 peer_url: peer_url.to_string(),
3178 message_id: message_id.to_string(),
3179 task_id: task_id.to_string(),
3180 context_id: context_id.to_string(),
3181 traceparent: traceparent.map(str::to_string),
3182 trust: trust.as_str().to_string(),
3183 };
3184 self.write_event(&event).await
3185 }
3186
3187 pub(crate) async fn write_artifact_pulled(
3193 &mut self,
3194 pulled: &crate::runtime::PulledArtifact,
3195 ) -> std::io::Result<()> {
3196 let artifact = &pulled.summary;
3197 let session = match &artifact.origin {
3198 LockOrigin::Runtime { session } => Some(session.clone()),
3199 LockOrigin::Operator => None,
3200 };
3201 let event = ArtifactPulledEvent {
3202 event_type: "artifact_pulled",
3203 event_id: new_event_id(),
3204 parent_id: self.session_parent(),
3205 session_id: self.session_id.clone(),
3206 timestamp: timestamp_ms(),
3207 name: artifact.name.clone(),
3208 version: artifact.version.clone(),
3209 runtime: artifact.runtime.as_str(),
3210 origin: artifact.origin.as_str(),
3211 session,
3212 trust: crate::origin::artifact_trust(&artifact.origin).as_str(),
3213 fetched_from: pulled.fetched_from.clone(),
3214 owner_id: pulled.owner_id,
3215 };
3216 self.write_event(&event).await
3217 }
3218
3219 #[allow(clippy::too_many_arguments)]
3222 pub(crate) async fn write_task_start(
3223 &mut self,
3224 task_id: &str,
3225 context_id: &str,
3226 source: &str,
3227 provenance: TaskProvenance,
3228 message_parts_bytes: u64,
3229 part_kinds: &[&'static str],
3230 reference_task_ids: &[String],
3231 ) -> std::io::Result<()> {
3232 self.task_turns = 0;
3234 self.task_input_tokens = 0;
3235 self.task_output_tokens = 0;
3236 self.task_tool_calls = 0;
3237 self.task_shell_calls = 0;
3238 self.task_agent_turns = 0;
3239 self.task_hook_inferences = 0;
3240 let started = Instant::now();
3241 self.task_start_instant = Some(started);
3242 self.active_task_id = Some(task_id.to_string());
3243 if let Some(status) = &self.session_status {
3244 status.task_started(task_id, started);
3245 }
3246
3247 let event_id = new_event_id();
3248 self.task_event_id = Some(event_id.clone());
3249 self.turn_event_id = None;
3250
3251 let event = TaskStartEvent {
3252 event_type: "task_start",
3253 event_id: event_id.clone(),
3254 parent_id: self.session_parent(),
3255 session_id: self.session_id.clone(),
3256 timestamp: timestamp_ms(),
3257 task_id: task_id.to_string(),
3258 context_id: context_id.to_string(),
3259 source: source.to_string(),
3260 origin: provenance.origin().as_str().to_string(),
3261 trust: provenance.trust().as_str().to_string(),
3262 lane: TaskLane::for_origin(provenance.origin())
3265 .as_str()
3266 .to_string(),
3267 message_parts_bytes,
3268 part_kinds: part_kinds.to_vec(),
3269 reference_task_ids: reference_task_ids.to_vec(),
3270 };
3271 self.write_event(&event).await
3272 }
3273
3274 pub(crate) async fn write_task_end(
3275 &mut self,
3276 task_id: &str,
3277 exit_status: &str,
3278 reopen_count: u32,
3279 running_commands: Vec<String>,
3280 keep_working: Option<KeepWorkingSummary>,
3281 done_check: Option<DoneCheckSummary>,
3282 ) -> std::io::Result<()> {
3283 let duration_ms = self
3284 .task_start_instant
3285 .take()
3286 .map(|t| t.elapsed().as_millis().try_into().unwrap_or(u64::MAX))
3287 .unwrap_or(0);
3288
3289 let event = TaskEndEvent {
3290 event_type: "task_end",
3291 event_id: new_event_id(),
3292 parent_id: self.task_parent(),
3293 session_id: self.session_id.clone(),
3294 timestamp: timestamp_ms(),
3295 task_id: task_id.to_string(),
3296 exit_status: exit_status.to_string(),
3297 duration_ms,
3298 turns: self.task_turns,
3299 input_tokens: self.task_input_tokens,
3300 output_tokens: self.task_output_tokens,
3301 tool_calls: self.task_tool_calls,
3302 shell_calls: self.task_shell_calls,
3303 reopen_count,
3304 running_commands,
3305 keep_working,
3306 done_check,
3307 };
3308 self.active_task_id = None;
3309 self.task_event_id = None;
3310 self.turn_event_id = None;
3311 self.task_agent_turns = 0;
3312 if let Some(status) = &self.session_status {
3313 status.task_ended();
3314 }
3315 self.write_event(&event).await
3316 }
3317
3318 pub(crate) async fn write_task_canceled(
3324 &mut self,
3325 task_id: &str,
3326 turn: Option<u32>,
3327 phase: &str,
3328 detached_work_ids: Vec<String>,
3329 delegation_ids: Vec<String>,
3330 ) -> std::io::Result<()> {
3331 let event = TaskCanceledEvent {
3332 event_type: "task_canceled",
3333 event_id: new_event_id(),
3334 parent_id: self.task_parent(),
3335 session_id: self.session_id.clone(),
3336 timestamp: timestamp_ms(),
3337 task_id: task_id.to_string(),
3338 turn,
3339 phase: phase.to_string(),
3340 detached_work_ids,
3341 delegation_ids,
3342 };
3343 self.write_event(&event).await
3344 }
3345
3346 pub(crate) async fn write_task_rejected(
3352 &mut self,
3353 task_id: &str,
3354 context_id: &str,
3355 source: &str,
3356 cause: &str,
3357 reason: &str,
3358 ) -> std::io::Result<()> {
3359 let event = TaskRejectedEvent {
3360 event_type: "task_rejected",
3361 event_id: new_event_id(),
3362 parent_id: self.task_parent(),
3363 session_id: self.session_id.clone(),
3364 timestamp: timestamp_ms(),
3365 task_id: task_id.to_string(),
3366 context_id: context_id.to_string(),
3367 source: source.to_string(),
3368 cause: cause.to_string(),
3369 reason: reason.to_string(),
3370 };
3371 self.write_event(&event).await
3372 }
3373
3374 pub(crate) async fn write_task_failed(
3380 &mut self,
3381 turn: Option<u32>,
3382 cause: &str,
3383 reason: &str,
3384 ) -> std::io::Result<()> {
3385 let reason = crate::agent::truncate_on_char_boundary(reason, MAX_TASK_FAILURE_REASON_BYTES)
3386 .to_string();
3387 let event = TaskFailedEvent {
3388 event_type: "task_failed",
3389 event_id: new_event_id(),
3390 parent_id: self.task_parent(),
3391 session_id: self.session_id.clone(),
3392 timestamp: timestamp_ms(),
3393 task_id: self.active_task_id.clone(),
3394 turn,
3395 cause: cause.to_string(),
3396 reason: reason.clone(),
3397 };
3398 self.task_failures_written += 1;
3399 self.last_task_failure = Some(reason);
3400 self.write_event(&event).await
3401 }
3402
3403 pub(crate) async fn write_workdir_path_refused(
3405 &mut self,
3406 refusal: &crate::workdir_fs::WorkdirRefusal,
3407 ) -> std::io::Result<()> {
3408 let event = WorkdirPathRefusedEvent {
3409 event_type: "workdir_path_refused",
3410 event_id: new_event_id(),
3411 parent_id: self.task_parent(),
3412 session_id: self.session_id.clone(),
3413 timestamp: timestamp_ms(),
3414 task_id: self.active_task_id.clone(),
3415 path: &refusal.path,
3416 operation: refusal.operation,
3417 refused_at: &refusal.refused_at,
3418 reason: refusal.reason,
3419 };
3420 self.write_event(&event).await
3421 }
3422
3423 pub(crate) fn last_task_failure(&self) -> Option<&str> {
3425 self.last_task_failure.as_deref()
3426 }
3427
3428 pub(crate) fn task_failures_written(&self) -> u64 {
3431 self.task_failures_written
3432 }
3433
3434 pub(crate) async fn write_task_reopened(
3438 &mut self,
3439 task_id: &str,
3440 hook_name: &str,
3441 reason: &str,
3442 reopen_number: u32,
3443 attempt_context: &'static str,
3444 turns_remaining: Option<u32>,
3445 ) -> std::io::Result<()> {
3446 let event = TaskReopenedEvent {
3447 event_type: "task_reopened",
3448 event_id: new_event_id(),
3449 parent_id: self.task_parent(),
3450 session_id: self.session_id.clone(),
3451 timestamp: timestamp_ms(),
3452 task_id: task_id.to_string(),
3453 hook_name: hook_name.to_string(),
3454 reason: reason.to_string(),
3455 reopen_number,
3456 attempt_context,
3457 turns_remaining,
3458 };
3459 self.write_event(&event).await
3460 }
3461
3462 pub(crate) async fn write_task_continued(
3466 &mut self,
3467 task_id: &str,
3468 continuation_number: u32,
3469 member_calls: &[String],
3470 delegations: &[String],
3471 waited_ms: u64,
3472 turns_remaining: Option<u32>,
3473 ) -> std::io::Result<()> {
3474 let event = TaskContinuedEvent {
3475 event_type: "task_continued",
3476 event_id: new_event_id(),
3477 parent_id: self.task_parent(),
3478 session_id: self.session_id.clone(),
3479 timestamp: timestamp_ms(),
3480 task_id: task_id.to_string(),
3481 continuation_number,
3482 member_calls,
3483 delegations,
3484 waited_ms,
3485 turns_remaining,
3486 };
3487 self.write_event(&event).await
3488 }
3489
3490 pub(crate) async fn write_keep_working(
3493 &mut self,
3494 task_id: &str,
3495 number: u32,
3496 limit: u32,
3497 turns_remaining: Option<u32>,
3498 message_bytes: u64,
3499 ) -> std::io::Result<()> {
3500 let event = KeepWorkingEvent {
3501 event_type: "keep_working",
3502 event_id: new_event_id(),
3503 parent_id: self.task_parent(),
3504 session_id: self.session_id.clone(),
3505 timestamp: timestamp_ms(),
3506 task_id: task_id.to_string(),
3507 number,
3508 limit,
3509 turns_remaining,
3510 message_bytes,
3511 };
3512 if let Some(status) = &self.session_status {
3513 status.keep_working_sent(number);
3514 }
3515 self.write_event(&event).await
3516 }
3517
3518 pub(crate) async fn write_done_check(
3521 &mut self,
3522 run: &DoneCheckRun<'_>,
3523 reopens: u32,
3524 ) -> std::io::Result<()> {
3525 let event = DoneCheckEvent {
3526 event_type: "done_check",
3527 event_id: new_event_id(),
3528 parent_id: self.task_parent(),
3529 session_id: self.session_id.clone(),
3530 timestamp: timestamp_ms(),
3531 run,
3532 };
3533 if run.reopened {
3534 if let Some(status) = &self.session_status {
3535 status.done_check_reopened(reopens);
3536 }
3537 }
3538 self.write_event(&event).await
3539 }
3540
3541 pub(crate) fn task_turns(&self) -> u32 {
3551 self.task_turns
3552 }
3553
3554 pub(crate) fn task_hook_inferences(&self) -> u32 {
3556 self.task_hook_inferences
3557 }
3558
3559 pub(crate) fn task_agent_turns(&self) -> u32 {
3563 self.task_agent_turns
3564 }
3565
3566 pub(crate) async fn write_compaction(
3567 &mut self,
3568 turn: u32,
3569 tokens_before: u64,
3570 tokens_before_source: OccupancyBasis,
3571 tokens_after: u64,
3572 ) -> std::io::Result<()> {
3573 let event = CompactionEvent {
3574 event_type: "compaction",
3575 event_id: new_event_id(),
3576 parent_id: self.turn_parent(),
3577 session_id: self.session_id.clone(),
3578 timestamp: timestamp_ms(),
3579 turn,
3580 task_id: self.active_task_id.clone(),
3581 tokens_before,
3582 tokens_before_source,
3583 tokens_after,
3584 };
3585 self.write_event(&event).await
3586 }
3587
3588 pub(crate) async fn write_compaction_declined(
3594 &mut self,
3595 turn: u32,
3596 tokens: u64,
3597 tokens_source: OccupancyBasis,
3598 reason: &str,
3599 ) -> std::io::Result<()> {
3600 let event = CompactionDeclinedEvent {
3601 event_type: "compaction_declined",
3602 event_id: new_event_id(),
3603 parent_id: self.turn_parent(),
3604 session_id: self.session_id.clone(),
3605 timestamp: timestamp_ms(),
3606 turn,
3607 task_id: self.active_task_id.clone(),
3608 tokens,
3609 tokens_source,
3610 reason: reason.to_string(),
3611 };
3612 self.write_event(&event).await
3613 }
3614
3615 #[allow(clippy::too_many_arguments)]
3622 pub(crate) async fn write_context_seed(
3623 &mut self,
3624 hook_name: &str,
3625 tokens: u64,
3626 proposed_tokens: u64,
3627 budget_tokens: u64,
3628 outcome: &str,
3629 reason: Option<&str>,
3630 message_ids: Vec<String>,
3631 ) -> std::io::Result<()> {
3632 let event = ContextSeedEvent {
3633 event_type: "context_seed",
3634 event_id: new_event_id(),
3635 parent_id: self.task_parent(),
3636 session_id: self.session_id.clone(),
3637 timestamp: timestamp_ms(),
3638 task_id: self.active_task_id.clone(),
3639 hook_name: hook_name.to_string(),
3640 tokens,
3641 proposed_tokens,
3642 budget_tokens,
3643 outcome: outcome.to_string(),
3644 reason: reason.map(str::to_string),
3645 message_ids,
3646 };
3647 self.write_event(&event).await
3648 }
3649
3650 pub(crate) async fn write_retention(
3656 &mut self,
3657 store: &str,
3658 reason: &str,
3659 targets: Vec<String>,
3660 messages_dropped: Option<u64>,
3661 ) -> std::io::Result<()> {
3662 let event = RetentionEvent {
3663 event_type: "retention",
3664 event_id: new_event_id(),
3665 parent_id: self.session_parent(),
3666 session_id: self.session_id.clone(),
3667 timestamp: timestamp_ms(),
3668 store: store.to_string(),
3669 reason: reason.to_string(),
3670 removed: u32::try_from(targets.len()).unwrap_or(u32::MAX),
3671 targets,
3672 messages_dropped,
3673 };
3674 self.write_event(&event).await
3675 }
3676
3677 pub(crate) async fn write_call_denied(
3681 &mut self,
3682 turn: u32,
3683 event: &str,
3684 hook_name: &str,
3685 target: &str,
3686 reason: &str,
3687 ) -> std::io::Result<()> {
3688 let event = CallDeniedEvent {
3689 event_type: "call_denied",
3690 event_id: new_event_id(),
3691 parent_id: self.turn_parent(),
3692 session_id: self.session_id.clone(),
3693 timestamp: timestamp_ms(),
3694 turn,
3695 event: event.to_string(),
3696 hook_name: hook_name.to_string(),
3697 target: target.to_string(),
3698 reason: reason.to_string(),
3699 };
3700 self.write_event(&event).await
3701 }
3702
3703 pub(crate) async fn write_tool_input_refused(
3708 &mut self,
3709 turn: u32,
3710 tool_name: &str,
3711 tool_call_id: Option<&str>,
3712 missing: &[String],
3713 reason: &str,
3714 ) -> std::io::Result<()> {
3715 let event = ToolInputRefusedEvent {
3716 event_type: "tool_input_refused",
3717 event_id: new_event_id(),
3718 parent_id: self.turn_parent(),
3719 session_id: self.session_id.clone(),
3720 timestamp: timestamp_ms(),
3721 turn,
3722 tool_name: tool_name.to_string(),
3723 tool_call_id: tool_call_id.filter(|id| !id.is_empty()).map(str::to_string),
3724 missing: missing.to_vec(),
3725 reason: reason.to_string(),
3726 };
3727 self.write_event(&event).await
3728 }
3729
3730 pub(crate) async fn write_inference_retry(
3735 &mut self,
3736 turn: u32,
3737 retry: &InferenceRetryRecord,
3738 ) -> std::io::Result<()> {
3739 let event = InferenceRetryEvent {
3740 event_type: "inference_retry",
3741 event_id: new_event_id(),
3742 parent_id: self.turn_parent(),
3743 session_id: self.session_id.clone(),
3744 timestamp: timestamp_ms(),
3745 turn,
3746 task_id: self.active_task_id.clone(),
3747 attempt: retry.attempt,
3748 max_retries: retry.max_retries,
3749 reason: retry.reason.wire_name(),
3750 provider_status: retry.provider_status,
3751 error: crate::agent::truncate_on_char_boundary(
3752 &retry.error,
3753 MAX_TASK_FAILURE_REASON_BYTES,
3754 ),
3755 delay_ms: retry.delay_ms,
3756 retry_after_ms: retry.retry_after_ms,
3757 charged_tokens: retry.charged_tokens,
3758 origin: retry.origin.as_deref(),
3759 };
3760 self.write_event(&event).await
3761 }
3762
3763 pub(crate) async fn write_spend_ceiling_reached(
3766 &mut self,
3767 turn: u32,
3768 refusal: &crate::spend::SpendRefusal,
3769 origin: Option<&str>,
3770 ) -> std::io::Result<()> {
3771 let event = SpendCeilingReachedEvent {
3772 event_type: "spend_ceiling_reached",
3773 event_id: new_event_id(),
3774 parent_id: self.turn_parent(),
3775 session_id: self.session_id.clone(),
3776 timestamp: timestamp_ms(),
3777 turn,
3778 task_id: self.active_task_id.clone(),
3779 refusal,
3780 origin,
3781 };
3782 self.write_event(&event).await
3783 }
3784
3785 #[allow(clippy::too_many_arguments)]
3789 pub(crate) async fn write_protected_path_denied(
3790 &mut self,
3791 turn: u32,
3792 call: &str,
3793 target: &str,
3794 path: &str,
3795 rule: &str,
3796 signal: &str,
3797 reason: &str,
3798 ) -> std::io::Result<()> {
3799 let event = ProtectedPathDeniedEvent {
3800 event_type: "protected_path_denied",
3801 event_id: new_event_id(),
3802 parent_id: self.turn_parent(),
3803 session_id: self.session_id.clone(),
3804 timestamp: timestamp_ms(),
3805 turn,
3806 call: call.to_string(),
3807 target: target.to_string(),
3808 path: path.to_string(),
3809 rule: rule.to_string(),
3810 signal: signal.to_string(),
3811 reason: reason.to_string(),
3812 };
3813 self.write_event(&event).await
3814 }
3815
3816 pub(crate) async fn write_hook_dispatch_error(
3822 &mut self,
3823 hook_name: &str,
3824 event: &str,
3825 arm: &str,
3826 ) -> std::io::Result<()> {
3827 let event = HookDispatchErrorEvent {
3828 event_type: "hook_dispatch_error",
3829 event_id: new_event_id(),
3830 parent_id: self.session_parent(),
3831 session_id: self.session_id.clone(),
3832 timestamp: timestamp_ms(),
3833 hook_name: hook_name.to_string(),
3834 event: event.to_string(),
3835 arm: arm.to_string(),
3836 };
3837 self.write_event(&event).await
3838 }
3839
3840 pub(crate) async fn write_harness_start(&mut self, start: HarnessStart) -> std::io::Result<()> {
3850 let event = HarnessStartEvent {
3851 event_type: "harness_start",
3852 event_id: new_event_id(),
3853 parent_id: self.task_parent(),
3854 session_id: self.session_id.clone(),
3855 timestamp: timestamp_ms(),
3856 task_id: self.active_task_id.clone(),
3857 start,
3858 };
3859 self.write_event(&event).await
3860 }
3861
3862 pub(crate) async fn write_harness_warning(
3864 &mut self,
3865 code: &str,
3866 message: &str,
3867 ) -> std::io::Result<()> {
3868 let event = HarnessWarningEvent {
3869 event_type: "harness_warning",
3870 event_id: new_event_id(),
3871 parent_id: self.task_parent(),
3872 session_id: self.session_id.clone(),
3873 timestamp: timestamp_ms(),
3874 task_id: self.active_task_id.clone(),
3875 code: code.to_string(),
3876 message: message.to_string(),
3877 };
3878 self.write_event(&event).await
3879 }
3880
3881 pub(crate) async fn write_harness_session(
3883 &mut self,
3884 harness_session_id: &str,
3885 auth: &str,
3886 model: Option<&str>,
3887 ) -> std::io::Result<()> {
3888 let event = HarnessSessionEvent {
3889 event_type: "harness_session",
3890 event_id: new_event_id(),
3891 parent_id: self.task_parent(),
3892 session_id: self.session_id.clone(),
3893 timestamp: timestamp_ms(),
3894 task_id: self.active_task_id.clone(),
3895 harness_session_id: harness_session_id.to_string(),
3896 auth: auth.to_string(),
3897 model: model.map(str::to_string),
3898 };
3899 self.write_event(&event).await
3900 }
3901
3902 pub(crate) async fn write_harness_session_forgotten(
3904 &mut self,
3905 context_id: &str,
3906 harness_session_id: &str,
3907 requested_by: &str,
3908 ) -> std::io::Result<()> {
3909 let event = HarnessSessionForgottenEvent {
3910 event_type: "harness_session_forgotten",
3911 event_id: new_event_id(),
3912 parent_id: self.task_parent(),
3913 session_id: self.session_id.clone(),
3914 timestamp: timestamp_ms(),
3915 task_id: self.active_task_id.clone(),
3916 context_id: context_id.to_string(),
3917 harness_session_id: harness_session_id.to_string(),
3918 requested_by: requested_by.to_string(),
3919 };
3920 self.write_event(&event).await
3921 }
3922
3923 pub(crate) async fn write_harness_retry(
3925 &mut self,
3926 attempt: u32,
3927 reason: &str,
3928 ) -> std::io::Result<()> {
3929 let event = HarnessRetryEvent {
3930 event_type: "harness_retry",
3931 event_id: new_event_id(),
3932 parent_id: self.task_parent(),
3933 session_id: self.session_id.clone(),
3934 timestamp: timestamp_ms(),
3935 task_id: self.active_task_id.clone(),
3936 attempt,
3937 reason: reason.to_string(),
3938 };
3939 self.write_event(&event).await
3940 }
3941
3942 pub(crate) async fn write_harness_note(&mut self, text: &str) -> std::io::Result<()> {
3945 let event = HarnessNoteEvent {
3946 event_type: "harness_note",
3947 event_id: new_event_id(),
3948 parent_id: self.task_parent(),
3949 session_id: self.session_id.clone(),
3950 timestamp: timestamp_ms(),
3951 task_id: self.active_task_id.clone(),
3952 text: text.to_string(),
3953 };
3954 self.write_event(&event).await
3955 }
3956
3957 pub(crate) async fn write_harness_failed(
3960 &mut self,
3961 kind: &str,
3962 message: &str,
3963 source: &str,
3964 ) -> std::io::Result<()> {
3965 let event = HarnessFailedEvent {
3966 event_type: "harness_failed",
3967 event_id: new_event_id(),
3968 parent_id: self.task_parent(),
3969 session_id: self.session_id.clone(),
3970 timestamp: timestamp_ms(),
3971 task_id: self.active_task_id.clone(),
3972 kind: kind.to_string(),
3973 message: message.to_string(),
3974 source: source.to_string(),
3975 };
3976 self.write_event(&event).await
3977 }
3978
3979 pub(crate) async fn write_harness_interrupt(
3982 &mut self,
3983 method: &str,
3984 delivered: bool,
3985 grace_ms: u64,
3986 ) -> std::io::Result<()> {
3987 let event = HarnessInterruptEvent {
3988 event_type: "harness_interrupt",
3989 event_id: new_event_id(),
3990 parent_id: self.task_parent(),
3991 session_id: self.session_id.clone(),
3992 timestamp: timestamp_ms(),
3993 task_id: self.active_task_id.clone(),
3994 method: method.to_string(),
3995 delivered,
3996 grace_ms,
3997 };
3998 self.write_event(&event).await
3999 }
4000
4001 pub(crate) async fn write_harness_exit(&mut self, exit: HarnessExit) -> std::io::Result<()> {
4003 let event = HarnessExitEvent {
4004 event_type: "harness_exit",
4005 event_id: new_event_id(),
4006 parent_id: self.task_parent(),
4007 session_id: self.session_id.clone(),
4008 timestamp: timestamp_ms(),
4009 task_id: self.active_task_id.clone(),
4010 exit,
4011 };
4012 self.write_event(&event).await
4013 }
4014
4015 pub(crate) async fn write_session_end(&mut self, exit_status: &str) -> std::io::Result<()> {
4016 let duration_ms = self
4017 .session_start_time
4018 .elapsed()
4019 .as_millis()
4020 .try_into()
4021 .unwrap_or(u64::MAX);
4022 let event = SessionEndEvent {
4023 event_type: "session_end",
4024 event_id: new_event_id(),
4025 parent_id: self.session_parent(),
4026 session_id: self.session_id.clone(),
4027 timestamp: timestamp_ms(),
4028 total_turns: self.total_turns,
4029 total_input_tokens: self.total_input_tokens,
4030 total_output_tokens: self.total_output_tokens,
4031 total_tool_calls: self.total_tool_calls,
4032 total_shell_calls: self.total_shell_calls,
4033 duration_ms,
4034 exit_status: exit_status.to_string(),
4035 };
4036 self.write_event(&event).await?;
4037 self.session_ended = true;
4038 Ok(())
4039 }
4040
4041 pub(crate) async fn write_session_end_if_not_ended(
4044 &mut self,
4045 exit_status: &str,
4046 ) -> std::io::Result<()> {
4047 if self.session_started && !self.session_ended {
4048 self.write_session_end(exit_status).await?;
4049 }
4050 Ok(())
4051 }
4052
4053 pub(crate) async fn flush(&mut self) -> std::io::Result<()> {
4054 self.writer.flush().await
4055 }
4056
4057 async fn write_event(&mut self, event: &impl Serialize) -> std::io::Result<()> {
4065 let mut line = serde_json::to_string(event).map_err(std::io::Error::other)?;
4066 line.push('\n');
4067 self.writer.write_all(line.as_bytes()).await?;
4068 self.writer.flush().await
4069 }
4070}
4071
4072#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
4074pub(crate) struct SessionGateway {
4075 pub(crate) artifact: String,
4077 pub(crate) host: String,
4079 pub(crate) credential_source: &'static str,
4081 pub(crate) metered: bool,
4084}
4085
4086#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
4089pub(crate) struct RuntimeArtifact {
4090 name: String,
4091 version: String,
4092 origin: &'static str,
4094 session: String,
4096 trust: &'static str,
4098}
4099
4100#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
4102pub(crate) struct HostFactsRecord {
4103 binary: String,
4105 source: &'static str,
4107 #[serde(skip_serializing_if = "Option::is_none")]
4110 reason: Option<&'static str>,
4111 #[serde(skip_serializing_if = "Option::is_none")]
4113 changed_path: Option<String>,
4114}
4115
4116impl HostFactsRecord {
4117 fn of(used: &crate::host_facts::HostFactsUse) -> Self {
4118 use crate::host_facts::HostFactsSource;
4119
4120 let binary = used.binary.display().to_string();
4121 match &used.source {
4122 HostFactsSource::Kept => Self {
4123 binary,
4124 source: "kept",
4125 reason: None,
4126 changed_path: None,
4127 },
4128 HostFactsSource::Derived {
4129 reason,
4130 changed_path,
4131 } => Self {
4132 binary,
4133 source: "derived",
4134 reason: Some(reason.wire_name()),
4135 changed_path: changed_path.as_ref().map(|path| path.display().to_string()),
4136 },
4137 }
4138 }
4139}
4140
4141#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
4143pub(crate) struct SessionControl {
4144 pub(crate) settings: Vec<&'static str>,
4146 pub(crate) secrets: Vec<String>,
4148 #[serde(skip_serializing_if = "Vec::is_empty")]
4151 pub(crate) agent_settings: Vec<&'static str>,
4152}
4153
4154#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
4157pub(crate) struct SessionInferenceChoice {
4158 pub(crate) name: String,
4159 pub(crate) driver: String,
4160 pub(crate) model: String,
4161 pub(crate) credential_source: &'static str,
4164 pub(crate) available: bool,
4167 pub(crate) model_limits: Option<crate::model_limits::ResolvedModelLimits>,
4169}
4170
4171pub(crate) struct ControlChange<'a> {
4175 pub(crate) event_id: String,
4176 pub(crate) principal: crate::control_plane::Principal,
4178 pub(crate) token_id: Option<&'a str>,
4181 pub(crate) kind: &'static str,
4183 pub(crate) name: &'a str,
4184 pub(crate) action: &'static str,
4186 pub(crate) previous: Option<Value>,
4188 pub(crate) value: Option<Value>,
4189 pub(crate) applies_from: Option<&'static str>,
4191 pub(crate) replaced: Option<bool>,
4193}
4194
4195#[derive(Serialize)]
4198struct ControlChangeEvent<'a> {
4199 event_type: &'static str,
4200 event_id: String,
4201 parent_id: Option<String>,
4202 session_id: String,
4203 timestamp: u64,
4204 principal: &'static str,
4205 #[serde(skip_serializing_if = "Option::is_none")]
4206 token_id: Option<&'a str>,
4207 kind: &'static str,
4208 name: &'a str,
4209 action: &'static str,
4210 #[serde(skip_serializing_if = "Option::is_none")]
4211 previous: Option<Value>,
4212 #[serde(skip_serializing_if = "Option::is_none")]
4213 value: Option<Value>,
4214 #[serde(skip_serializing_if = "Option::is_none")]
4215 applies_from: Option<&'static str>,
4216 #[serde(skip_serializing_if = "Option::is_none")]
4217 replaced: Option<bool>,
4218}
4219
4220#[derive(Serialize)]
4224struct ControlRefusedEvent<'a> {
4225 event_type: &'static str,
4226 event_id: String,
4227 parent_id: Option<String>,
4228 session_id: String,
4229 timestamp: u64,
4230 principal: &'static str,
4231 status: u16,
4232 reason: &'static str,
4233 #[serde(skip_serializing_if = "Option::is_none")]
4234 kind: Option<&'static str>,
4235 #[serde(skip_serializing_if = "Option::is_none")]
4236 name: Option<&'a str>,
4237 #[serde(skip_serializing_if = "Option::is_none")]
4238 token_id: Option<&'a str>,
4239}
4240
4241#[derive(Serialize)]
4243struct ControlAppliedEvent<'a> {
4244 event_type: &'static str,
4245 event_id: String,
4246 parent_id: Option<String>,
4247 session_id: String,
4248 timestamp: u64,
4249 turn: u32,
4250 task_id: Option<String>,
4251 name: &'a str,
4252 value: Value,
4253 change_id: &'a str,
4254}
4255
4256#[derive(Serialize)]
4258struct ToolsRefreshedEvent<'a> {
4259 event_type: &'static str,
4260 event_id: String,
4261 parent_id: Option<String>,
4262 session_id: String,
4263 timestamp: u64,
4264 turn: u32,
4265 task_id: Option<String>,
4266 trigger: &'a str,
4267 added: &'a [String],
4268 removed: &'a [String],
4269 tools: &'a [String],
4270}
4271
4272#[derive(Serialize)]
4275struct GatewayCredentialEvent<'a> {
4276 event_type: &'static str,
4277 event_id: String,
4278 parent_id: Option<String>,
4279 session_id: String,
4280 timestamp: u64,
4281 artifact: &'a str,
4282 source: &'static str,
4283 #[serde(skip_serializing_if = "Option::is_none")]
4284 credential: Option<&'a str>,
4285 change: &'static str,
4286 #[serde(skip_serializing_if = "Option::is_none")]
4287 trigger: Option<&'static str>,
4288 #[serde(skip_serializing_if = "Option::is_none")]
4289 status: Option<u16>,
4290 #[serde(skip_serializing_if = "Option::is_none")]
4291 retried: Option<bool>,
4292 #[serde(skip_serializing_if = "Option::is_none")]
4293 reason: Option<&'static str>,
4294}
4295
4296#[derive(Serialize)]
4300struct InferenceCredentialEvent<'a> {
4301 event_type: &'static str,
4302 event_id: String,
4303 parent_id: Option<String>,
4304 session_id: String,
4305 timestamp: u64,
4306 source: &'static str,
4307 #[serde(skip_serializing_if = "Option::is_none")]
4308 credential: Option<&'a str>,
4309 change: &'static str,
4310 #[serde(skip_serializing_if = "Option::is_none")]
4311 trigger: Option<&'static str>,
4312 #[serde(skip_serializing_if = "Option::is_none")]
4313 status: Option<u16>,
4314 #[serde(skip_serializing_if = "Option::is_none")]
4315 retried: Option<bool>,
4316 #[serde(skip_serializing_if = "Option::is_none")]
4317 reason: Option<&'static str>,
4318}
4319
4320#[allow(clippy::type_complexity)]
4323fn credential_change_fields(
4324 change: CredentialChange,
4325) -> (
4326 &'static str,
4327 Option<&'static str>,
4328 Option<u16>,
4329 Option<bool>,
4330 Option<&'static str>,
4331) {
4332 match change {
4333 CredentialChange::Rotated { trigger } => ("rotated", Some(trigger), None, None, None),
4334 CredentialChange::Rejected { status, retried } => {
4335 ("rejected", None, Some(status), Some(retried), None)
4336 }
4337 CredentialChange::Unreadable { reason } => ("unreadable", None, None, None, Some(reason)),
4338 }
4339}
4340
4341pub struct ResourceTraceAppender {
4356 file: tokio::sync::Mutex<File>,
4358 session_id: String,
4359 session_event_id: String,
4364}
4365
4366impl ResourceTraceAppender {
4367 pub async fn open(
4368 records_root: &WorkdirRoot,
4369 session_id: String,
4370 session_event_id: String,
4371 ) -> std::io::Result<Self> {
4372 let file = File::from_std(records_root.open_append(TRACE_FILE_NAME, DEFAULT_FILE_MODE)?);
4373 Ok(Self {
4374 file: tokio::sync::Mutex::new(file),
4375 session_id,
4376 session_event_id,
4377 })
4378 }
4379
4380 #[allow(clippy::too_many_arguments)]
4384 pub(crate) async fn write_resource_list(
4385 &self,
4386 root: &str,
4387 entry_count: usize,
4388 total_bytes: u64,
4389 generation: u64,
4390 containment_achieved: ContainmentClass,
4391 outcome: &str,
4392 reason: Option<String>,
4393 ) {
4394 let event = ResourceListEvent {
4395 event_type: "resource_list",
4396 event_id: new_event_id(),
4397 parent_id: Some(self.session_event_id.clone()),
4398 session_id: self.session_id.clone(),
4399 timestamp: timestamp_ms(),
4400 root: root.to_string(),
4401 entry_count,
4402 total_bytes,
4403 generation,
4404 containment_achieved,
4405 outcome: outcome.to_string(),
4406 reason,
4407 };
4408 self.append(&event).await;
4409 }
4410
4411 #[allow(clippy::too_many_arguments)]
4414 pub(crate) async fn write_resource_read(
4415 &self,
4416 path: &str,
4417 outcome: &str,
4418 bytes: Option<u64>,
4419 sha256: Option<String>,
4420 generation: u64,
4421 containment_achieved: ContainmentClass,
4422 reason: Option<String>,
4423 ) {
4424 let event = ResourceReadEvent {
4425 event_type: "resource_read",
4426 event_id: new_event_id(),
4427 parent_id: Some(self.session_event_id.clone()),
4428 session_id: self.session_id.clone(),
4429 timestamp: timestamp_ms(),
4430 path: path.to_string(),
4431 outcome: outcome.to_string(),
4432 bytes,
4433 sha256,
4434 generation,
4435 containment_achieved,
4436 reason,
4437 };
4438 self.append(&event).await;
4439 }
4440
4441 pub(crate) async fn write_formation_ended(&self, formation_id: &crate::FormationId) {
4444 let event = FormationEndedEvent {
4445 event_type: "formation_ended",
4446 event_id: new_event_id(),
4447 parent_id: Some(self.session_event_id.clone()),
4448 session_id: self.session_id.clone(),
4449 timestamp: timestamp_ms(),
4450 formation_id: formation_id.as_str().to_string(),
4451 };
4452 self.append(&event).await;
4453 }
4454
4455 pub(crate) async fn write_session_status(
4458 &self,
4459 task_id: Option<String>,
4460 status: &crate::session_status::StatusReport,
4461 ) {
4462 let event = SessionStatusEvent {
4463 event_type: "session_status",
4464 event_id: new_event_id(),
4465 parent_id: Some(self.session_event_id.clone()),
4466 session_id: self.session_id.clone(),
4467 timestamp: timestamp_ms(),
4468 task_id,
4469 status,
4470 };
4471 self.append(&event).await;
4472 }
4473
4474 #[allow(clippy::too_many_arguments)]
4477 pub(crate) async fn write_background_command(
4478 &self,
4479 task_id: Option<String>,
4480 work_id: &str,
4481 operation: &'static str,
4482 state: &'static str,
4483 exit_code: Option<i32>,
4484 waited_ms: Option<u64>,
4485 output_bytes: u64,
4486 ) {
4487 let event = BackgroundCommandEvent {
4488 event_type: "background_command",
4489 event_id: new_event_id(),
4490 parent_id: Some(self.session_event_id.clone()),
4491 session_id: self.session_id.clone(),
4492 timestamp: timestamp_ms(),
4493 task_id,
4494 work_id,
4495 operation,
4496 state,
4497 exit_code,
4498 waited_ms,
4499 output_bytes,
4500 };
4501 self.append(&event).await;
4502 }
4503
4504 pub(crate) async fn write_deadline_notice(
4506 &self,
4507 task_id: Option<String>,
4508 tool_name: &str,
4509 notice: &crate::session_status::DeadlineNotice,
4510 ) {
4511 let event = DeadlineNoticeEvent {
4512 event_type: "deadline_notice",
4513 event_id: new_event_id(),
4514 parent_id: Some(self.session_event_id.clone()),
4515 session_id: self.session_id.clone(),
4516 timestamp: timestamp_ms(),
4517 task_id,
4518 tool_name,
4519 deadline_at: ¬ice.at,
4520 left_secs: notice.left_secs,
4521 notice_secs: notice.notice_secs,
4522 };
4523 self.append(&event).await;
4524 }
4525
4526 pub(crate) async fn write_deadline_reached(
4529 &self,
4530 deadline: &crate::session_status::SessionDeadline,
4531 ) {
4532 let event = DeadlineReachedEvent {
4533 event_type: "deadline_reached",
4534 event_id: new_event_id(),
4535 parent_id: Some(self.session_event_id.clone()),
4536 session_id: self.session_id.clone(),
4537 timestamp: timestamp_ms(),
4538 deadline_at: deadline.at_rfc3339(),
4539 deadline_secs: deadline.secs(),
4540 source: deadline.source().as_str(),
4541 };
4542 self.append(&event).await;
4543 }
4544
4545 pub(crate) async fn write_spawner_ended(
4549 &self,
4550 spawned_by: Option<&str>,
4551 delegation_id: Option<&str>,
4552 ) {
4553 let event = SpawnerEndedEvent {
4554 event_type: "spawner_ended",
4555 event_id: new_event_id(),
4556 parent_id: Some(self.session_event_id.clone()),
4557 session_id: self.session_id.clone(),
4558 timestamp: timestamp_ms(),
4559 spawned_by: spawned_by.map(str::to_string),
4560 delegation_id: delegation_id.map(str::to_string),
4561 };
4562 self.append(&event).await;
4563 }
4564
4565 #[allow(clippy::too_many_arguments)]
4568 pub(crate) async fn write_peer_handle_mint(
4569 &self,
4570 handle_id: Option<String>,
4571 path: &str,
4572 audience: &str,
4573 expires_at_ms: Option<u64>,
4574 outcome: &str,
4575 reason: Option<String>,
4576 ) {
4577 let event = PeerHandleMintEvent {
4578 event_type: "peer_handle_mint",
4579 event_id: new_event_id(),
4580 parent_id: Some(self.session_event_id.clone()),
4581 session_id: self.session_id.clone(),
4582 timestamp: timestamp_ms(),
4583 handle_id,
4584 path: path.to_string(),
4585 audience: audience.to_string(),
4586 expires_at_ms,
4587 outcome: outcome.to_string(),
4588 reason,
4589 };
4590 self.append(&event).await;
4591 }
4592
4593 #[allow(clippy::too_many_arguments)]
4596 pub(crate) async fn write_peer_handle_redeem(
4597 &self,
4598 handle_id: &str,
4599 path: Option<String>,
4600 generation: u64,
4601 audience_asserted: Option<String>,
4602 bytes: Option<u64>,
4603 sha256: Option<String>,
4604 outcome: &str,
4605 reason: Option<String>,
4606 ) {
4607 let event = PeerHandleRedeemEvent {
4608 event_type: "peer_handle_redeem",
4609 event_id: new_event_id(),
4610 parent_id: Some(self.session_event_id.clone()),
4611 session_id: self.session_id.clone(),
4612 timestamp: timestamp_ms(),
4613 handle_id: handle_id.to_string(),
4614 path,
4615 generation,
4616 audience_asserted,
4617 bytes,
4618 sha256,
4619 outcome: outcome.to_string(),
4620 reason,
4621 };
4622 self.append(&event).await;
4623 }
4624
4625 #[allow(clippy::too_many_arguments)]
4627 pub(crate) async fn write_peer_file_fetch(
4628 &self,
4629 peer: &str,
4630 handle_id: &str,
4631 stored_path: Option<String>,
4632 bytes: Option<u64>,
4633 sha256: Option<String>,
4634 outcome: &str,
4635 reason: Option<String>,
4636 ) {
4637 let event = PeerFileFetchEvent {
4638 event_type: "peer_file_fetch",
4639 event_id: new_event_id(),
4640 parent_id: Some(self.session_event_id.clone()),
4641 session_id: self.session_id.clone(),
4642 timestamp: timestamp_ms(),
4643 peer: peer.to_string(),
4644 handle_id: handle_id.to_string(),
4645 stored_path,
4646 bytes,
4647 sha256,
4648 outcome: outcome.to_string(),
4649 reason,
4650 };
4651 self.append(&event).await;
4652 }
4653
4654 pub(crate) async fn write_delegation_start(
4659 &self,
4660 delegation_id: &str,
4661 capsule: &str,
4662 version: &str,
4663 child_session_id: &str,
4664 child_workdir: &str,
4665 child_records_dir: &str,
4666 ) {
4667 let event = DelegationStartEvent::new(
4668 &self.session_id,
4669 &self.session_event_id,
4670 delegation_id,
4671 capsule,
4672 version,
4673 child_session_id,
4674 child_workdir,
4675 child_records_dir,
4676 );
4677 self.append(&event).await;
4678 }
4679
4680 #[allow(clippy::too_many_arguments)]
4684 pub(crate) async fn write_delegation(
4685 &self,
4686 capsule: &str,
4687 version: &str,
4688 delegation_id: Option<String>,
4689 child_session_id: Option<String>,
4690 duration_ms: u64,
4691 outcome: &str,
4692 reason: Option<String>,
4693 ) {
4694 let event = DelegationEvent::new(
4695 &self.session_id,
4696 &self.session_event_id,
4697 capsule,
4698 version,
4699 delegation_id,
4700 child_session_id,
4701 duration_ms,
4702 outcome,
4703 reason,
4704 );
4705 self.append(&event).await;
4706 }
4707
4708 pub(crate) async fn write_member_call_start(
4710 &self,
4711 task_id: &str,
4712 call_id: &str,
4713 member: &str,
4714 member_task_id: &str,
4715 ) {
4716 let event = self.member_call_start(task_id, call_id, member, member_task_id);
4717 self.append(&event).await;
4718 }
4719
4720 pub(crate) async fn write_member_call_start_if(
4723 &self,
4724 task_id: &str,
4725 call_id: &str,
4726 member: &str,
4727 member_task_id: &str,
4728 admit: impl FnOnce() -> bool,
4729 ) -> bool {
4730 let event = self.member_call_start(task_id, call_id, member, member_task_id);
4731 self.append_if(&event, admit).await
4732 }
4733
4734 fn member_call_start(
4735 &self,
4736 task_id: &str,
4737 call_id: &str,
4738 member: &str,
4739 member_task_id: &str,
4740 ) -> MemberCallStartEvent {
4741 MemberCallStartEvent {
4742 event_type: "member_call_start",
4743 event_id: new_event_id(),
4744 parent_id: Some(self.session_event_id.clone()),
4745 session_id: self.session_id.clone(),
4746 timestamp: timestamp_ms(),
4747 task_id: task_id.to_string(),
4748 call_id: call_id.to_string(),
4749 member: member.to_string(),
4750 member_task_id: member_task_id.to_string(),
4751 }
4752 }
4753
4754 pub(crate) async fn write_member_call_busy(
4757 &self,
4758 task_id: &str,
4759 call_id: &str,
4760 member: &str,
4761 offer: u32,
4762 waited_ms: u64,
4763 message: &str,
4764 ) {
4765 let event = MemberCallBusyEvent {
4766 event_type: "member_call_busy",
4767 event_id: new_event_id(),
4768 parent_id: Some(self.session_event_id.clone()),
4769 session_id: self.session_id.clone(),
4770 timestamp: timestamp_ms(),
4771 task_id: task_id.to_string(),
4772 call_id: call_id.to_string(),
4773 member: member.to_string(),
4774 offer,
4775 waited_ms,
4776 message: message.to_string(),
4777 };
4778 self.append(&event).await;
4779 }
4780
4781 pub(crate) async fn write_member_call(
4783 &self,
4784 task_id: &str,
4785 outcome: &crate::member_call::MemberCallOutcome,
4786 delivered: bool,
4787 ) {
4788 let event = MemberCallEvent {
4789 event_type: "member_call",
4790 event_id: new_event_id(),
4791 parent_id: Some(self.session_event_id.clone()),
4792 session_id: self.session_id.clone(),
4793 timestamp: timestamp_ms(),
4794 task_id: task_id.to_string(),
4795 call_id: outcome.call_id.clone(),
4796 member: outcome.member.clone(),
4797 member_task_id: outcome.member_task_id.clone(),
4798 status: outcome.status.as_str().to_string(),
4799 duration_ms: outcome.duration_ms,
4800 output: outcome.output.clone(),
4801 truncated: outcome.truncated,
4802 delivered,
4803 no_answer_below: outcome
4804 .below
4805 .iter()
4806 .map(|(member, status)| NoAnswerBelow {
4807 member: member.clone(),
4808 status: status.as_str().to_string(),
4809 })
4810 .collect(),
4811 };
4812 self.append(&event).await;
4813 }
4814
4815 pub(crate) async fn write_inference_credential(
4818 &self,
4819 source: &'static str,
4820 credential: Option<&str>,
4821 change: CredentialChange,
4822 ) {
4823 let (change_name, trigger, status, retried, reason) = credential_change_fields(change);
4824 let event = InferenceCredentialEvent {
4825 event_type: "inference_credential",
4826 event_id: new_event_id(),
4827 parent_id: Some(self.session_event_id.clone()),
4828 session_id: self.session_id.clone(),
4829 timestamp: timestamp_ms(),
4830 source,
4831 credential,
4832 change: change_name,
4833 trigger,
4834 status,
4835 retried,
4836 reason,
4837 };
4838 self.append(&event).await;
4839 }
4840
4841 pub(crate) async fn write_gateway_credential(
4844 &self,
4845 artifact: &str,
4846 source: &'static str,
4847 credential: Option<&str>,
4848 change: CredentialChange,
4849 ) {
4850 let (change_name, trigger, status, retried, reason) = credential_change_fields(change);
4851 let event = GatewayCredentialEvent {
4852 event_type: "gateway_credential",
4853 event_id: new_event_id(),
4854 parent_id: Some(self.session_event_id.clone()),
4855 session_id: self.session_id.clone(),
4856 timestamp: timestamp_ms(),
4857 artifact,
4858 source,
4859 credential,
4860 change: change_name,
4861 trigger,
4862 status,
4863 retried,
4864 reason,
4865 };
4866 self.append(&event).await;
4867 }
4868
4869 pub(crate) async fn write_control_change(&self, change: ControlChange<'_>) {
4871 let event = ControlChangeEvent {
4872 event_type: "control_change",
4873 event_id: change.event_id,
4874 parent_id: Some(self.session_event_id.clone()),
4875 session_id: self.session_id.clone(),
4876 timestamp: timestamp_ms(),
4877 principal: change.principal.as_str(),
4878 token_id: change.token_id,
4879 kind: change.kind,
4880 name: change.name,
4881 action: change.action,
4882 previous: change.previous,
4883 value: change.value,
4884 applies_from: change.applies_from,
4885 replaced: change.replaced,
4886 };
4887 self.append(&event).await;
4888 }
4889
4890 pub(crate) async fn write_control_refused(
4893 &self,
4894 principal: crate::control_plane::Principal,
4895 status: u16,
4896 reason: &'static str,
4897 kind: Option<&'static str>,
4898 name: Option<&str>,
4899 token_id: Option<&str>,
4900 ) {
4901 let event = ControlRefusedEvent {
4902 event_type: "control_refused",
4903 event_id: new_event_id(),
4904 parent_id: Some(self.session_event_id.clone()),
4905 session_id: self.session_id.clone(),
4906 timestamp: timestamp_ms(),
4907 principal: principal.as_str(),
4908 status,
4909 reason,
4910 kind,
4911 name,
4912 token_id,
4913 };
4914 self.append(&event).await;
4915 }
4916
4917 async fn append(&self, event: &impl Serialize) {
4918 self.append_if(event, || true).await;
4919 }
4920
4921 async fn append_if(&self, event: &impl Serialize, admit: impl FnOnce() -> bool) -> bool {
4924 let Ok(mut line) = serde_json::to_string(event) else {
4925 return admit();
4926 };
4927 line.push('\n');
4928 let mut file = self.file.lock().await;
4929 if !admit() {
4930 return false;
4931 }
4932 if file.write_all(line.as_bytes()).await.is_ok() {
4933 let _ = file.flush().await;
4934 }
4935 true
4936 }
4937}
4938
4939pub(crate) struct PriorSessionTraceAppender {
4954 file: File,
4955 session_id: String,
4957 session_event_id: Option<String>,
4959 terminate_torn_tail: bool,
4963}
4964
4965impl PriorSessionTraceAppender {
4966 pub(crate) async fn open(
4971 records_group: &WorkdirRoot,
4972 trace: &str,
4973 session_id: String,
4974 session_event_id: Option<String>,
4975 ) -> std::io::Result<Self> {
4976 let contents = records_group
4977 .read_file(trace, u64::MAX)?
4978 .ok_or_else(|| std::io::Error::from(std::io::ErrorKind::NotFound))?;
4979 let terminate_torn_tail = contents.last().is_some_and(|last| *last != b'\n');
4981 let file = File::from_std(records_group.open_append(trace, DEFAULT_FILE_MODE)?);
4982 Ok(Self {
4983 file,
4984 session_id,
4985 session_event_id,
4986 terminate_torn_tail,
4987 })
4988 }
4989
4990 pub(crate) async fn write_shell_lost(
4996 &mut self,
4997 work: &crate::detached::LostWork,
4998 reconciled_by_session: &str,
4999 reconciled_task_id: &str,
5000 ) -> std::io::Result<()> {
5001 let event = ShellLostEvent {
5002 event_type: "shell_lost",
5003 event_id: new_event_id(),
5004 parent_id: self.session_event_id.clone(),
5005 session_id: self.session_id.clone(),
5006 timestamp: timestamp_ms(),
5007 work_id: work.work_id.clone(),
5008 binary: work.binary.clone(),
5009 command: work.command.clone(),
5010 detached_at_ms: work.detached_at_ms,
5011 reconciled_by_session: reconciled_by_session.to_string(),
5012 reconciled_task_id: reconciled_task_id.to_string(),
5013 };
5014 let mut line = if self.terminate_torn_tail {
5015 "\n".to_string()
5016 } else {
5017 String::new()
5018 };
5019 line.push_str(&serde_json::to_string(&event).map_err(std::io::Error::other)?);
5020 line.push('\n');
5021 self.file.write_all(line.as_bytes()).await?;
5022 self.file.flush().await?;
5023 self.terminate_torn_tail = false;
5024 Ok(())
5025 }
5026}
5027
5028pub(crate) fn timestamp_ms() -> u64 {
5029 SystemTime::now()
5030 .duration_since(UNIX_EPOCH)
5031 .unwrap_or_default()
5032 .as_millis() as u64
5033}
5034
5035#[derive(Serialize, Clone)]
5042pub(crate) struct PlanStepShape {
5043 pub(crate) step_id: String,
5044 pub(crate) kind: &'static str,
5047 pub(crate) depends_on: Vec<String>,
5048 pub(crate) has_condition: bool,
5050}
5051
5052#[derive(Serialize)]
5055struct PlanStartEvent {
5056 event_type: &'static str,
5057 event_id: String,
5058 parent_id: Option<String>,
5059 session_id: String,
5060 timestamp: u64,
5061 plan_id: String,
5062 step_count: usize,
5063 steps: Vec<PlanStepShape>,
5064}
5065
5066#[derive(Serialize)]
5072struct PlanStepStartEvent {
5073 event_type: &'static str,
5074 event_id: String,
5075 parent_id: Option<String>,
5076 session_id: String,
5077 timestamp: u64,
5078 plan_id: String,
5079 step_id: String,
5080 kind: &'static str,
5081 depends_on: Vec<String>,
5082}
5083
5084#[derive(Serialize)]
5088struct PlanStepEvent {
5089 event_type: &'static str,
5090 event_id: String,
5091 parent_id: Option<String>,
5092 session_id: String,
5093 timestamp: u64,
5094 plan_id: String,
5095 step_id: String,
5096 kind: &'static str,
5097 status: &'static str,
5098 attempts: u32,
5099 duration_ms: u64,
5100 #[serde(skip_serializing_if = "Option::is_none")]
5101 error: Option<String>,
5102 #[serde(skip_serializing_if = "Option::is_none")]
5105 input: Option<Value>,
5106 #[serde(skip_serializing_if = "Option::is_none")]
5110 state_effect: Option<String>,
5111 #[serde(skip_serializing_if = "Option::is_none")]
5113 resource_id: Option<String>,
5114}
5115
5116#[derive(Serialize)]
5118struct PlanEndEvent {
5119 event_type: &'static str,
5120 event_id: String,
5121 parent_id: Option<String>,
5122 session_id: String,
5123 timestamp: u64,
5124 plan_id: String,
5125 outcome: &'static str,
5126 #[serde(skip_serializing_if = "Option::is_none")]
5127 failed_step: Option<String>,
5128 steps_total: usize,
5129 steps_succeeded: usize,
5130 steps_failed: usize,
5131 steps_skipped: usize,
5132 duration_ms: u64,
5133 #[serde(skip_serializing_if = "Option::is_none")]
5136 reason: Option<String>,
5137}
5138
5139pub(crate) struct PlanStepRecord {
5141 pub(crate) plan_id: String,
5142 pub(crate) step_id: String,
5143 pub(crate) kind: &'static str,
5144 pub(crate) status: &'static str,
5147 pub(crate) attempts: u32,
5149 pub(crate) duration_ms: u64,
5150 pub(crate) error: Option<String>,
5151 pub(crate) input: Option<Value>,
5152 pub(crate) state_effect: Option<String>,
5153 pub(crate) resource_id: Option<String>,
5154}
5155
5156pub(crate) struct PlanEndRecord {
5158 pub(crate) plan_id: String,
5159 pub(crate) outcome: &'static str,
5160 pub(crate) failed_step: Option<String>,
5161 pub(crate) steps_total: usize,
5162 pub(crate) steps_succeeded: usize,
5163 pub(crate) steps_failed: usize,
5164 pub(crate) steps_skipped: usize,
5165 pub(crate) duration_ms: u64,
5166 pub(crate) reason: Option<String>,
5167}
5168
5169pub struct PlanTraceAppender {
5180 file: std::sync::Mutex<std::fs::File>,
5181 session_id: String,
5182 session_event_id: String,
5185}
5186
5187impl PlanTraceAppender {
5188 pub fn open(
5189 records_root: &WorkdirRoot,
5190 session_id: String,
5191 session_event_id: String,
5192 ) -> std::io::Result<Self> {
5193 let file = records_root.open_append(TRACE_FILE_NAME, DEFAULT_FILE_MODE)?;
5194 Ok(Self {
5195 file: std::sync::Mutex::new(file),
5196 session_id,
5197 session_event_id,
5198 })
5199 }
5200
5201 pub(crate) fn write_plan_start(&self, plan_id: &str, steps: Vec<PlanStepShape>) -> String {
5203 let event_id = new_event_id();
5204 let event = PlanStartEvent {
5205 event_type: "plan_start",
5206 event_id: event_id.clone(),
5207 parent_id: Some(self.session_event_id.clone()),
5208 session_id: self.session_id.clone(),
5209 timestamp: timestamp_ms(),
5210 plan_id: plan_id.to_string(),
5211 step_count: steps.len(),
5212 steps,
5213 };
5214 self.append(&event);
5215 event_id
5216 }
5217
5218 pub(crate) fn write_plan_step_start(
5219 &self,
5220 plan_node: &str,
5221 plan_id: &str,
5222 step_id: &str,
5223 kind: &'static str,
5224 depends_on: Vec<String>,
5225 ) {
5226 let event = PlanStepStartEvent {
5227 event_type: "plan_step_start",
5228 event_id: new_event_id(),
5229 parent_id: Some(plan_node.to_string()),
5230 session_id: self.session_id.clone(),
5231 timestamp: timestamp_ms(),
5232 plan_id: plan_id.to_string(),
5233 step_id: step_id.to_string(),
5234 kind,
5235 depends_on,
5236 };
5237 self.append(&event);
5238 }
5239
5240 pub(crate) fn write_plan_step(&self, plan_node: &str, record: PlanStepRecord) {
5241 let input = record.input.map(|mut value| {
5244 crate::peer_handoff::redact_handles_in_json(&mut value);
5245 value
5246 });
5247 let event = PlanStepEvent {
5248 event_type: "plan_step",
5249 event_id: new_event_id(),
5250 parent_id: Some(plan_node.to_string()),
5251 session_id: self.session_id.clone(),
5252 timestamp: timestamp_ms(),
5253 plan_id: record.plan_id,
5254 step_id: record.step_id,
5255 kind: record.kind,
5256 status: record.status,
5257 attempts: record.attempts,
5258 duration_ms: record.duration_ms,
5259 error: record.error,
5260 input,
5261 state_effect: record.state_effect,
5262 resource_id: record.resource_id,
5263 };
5264 self.append(&event);
5265 }
5266
5267 pub(crate) fn write_plan_end(&self, plan_node: Option<&str>, record: PlanEndRecord) {
5270 let event = PlanEndEvent {
5271 event_type: "plan_end",
5272 event_id: new_event_id(),
5273 parent_id: Some(
5274 plan_node
5275 .map(str::to_string)
5276 .unwrap_or_else(|| self.session_event_id.clone()),
5277 ),
5278 session_id: self.session_id.clone(),
5279 timestamp: timestamp_ms(),
5280 plan_id: record.plan_id,
5281 outcome: record.outcome,
5282 failed_step: record.failed_step,
5283 steps_total: record.steps_total,
5284 steps_succeeded: record.steps_succeeded,
5285 steps_failed: record.steps_failed,
5286 steps_skipped: record.steps_skipped,
5287 duration_ms: record.duration_ms,
5288 reason: record.reason,
5289 };
5290 self.append(&event);
5291 }
5292
5293 pub(crate) fn write_delegation_start(
5299 &self,
5300 delegation_id: &str,
5301 capsule: &str,
5302 version: &str,
5303 child_session_id: &str,
5304 child_workdir: &str,
5305 child_records_dir: &str,
5306 ) {
5307 let event = DelegationStartEvent::new(
5308 &self.session_id,
5309 &self.session_event_id,
5310 delegation_id,
5311 capsule,
5312 version,
5313 child_session_id,
5314 child_workdir,
5315 child_records_dir,
5316 );
5317 self.append(&event);
5318 }
5319
5320 #[allow(clippy::too_many_arguments)]
5328 pub(crate) fn write_delegation(
5329 &self,
5330 capsule: &str,
5331 version: &str,
5332 delegation_id: Option<String>,
5333 child_session_id: Option<String>,
5334 duration_ms: u64,
5335 outcome: &str,
5336 reason: Option<String>,
5337 ) {
5338 let event = DelegationEvent::new(
5339 &self.session_id,
5340 &self.session_event_id,
5341 capsule,
5342 version,
5343 delegation_id,
5344 child_session_id,
5345 duration_ms,
5346 outcome,
5347 reason,
5348 );
5349 self.append(&event);
5350 }
5351
5352 fn append(&self, event: &impl Serialize) {
5356 use std::io::Write as _;
5357
5358 let Ok(mut line) = serde_json::to_string(event) else {
5359 return;
5360 };
5361 line.push('\n');
5362 let Ok(mut file) = self.file.lock() else {
5363 return;
5364 };
5365 if file.write_all(line.as_bytes()).is_ok() {
5366 let _ = file.flush();
5367 }
5368 }
5369}
5370
5371#[cfg(test)]
5374mod tests {
5375 use super::*;
5376 use crate::{
5377 sandbox::EnforcementTier, sealed::UsernsGrant, trace_blobs::BLOB_DIR_NAME,
5378 types::CapabilityPolicy,
5379 };
5380 use murmur_artifact::{ArtifactRuntime, InterpreterRuntimeDir, InterpreterRuntimeGrant};
5381 use serde_json::Value;
5382 use std::path::Path;
5383
5384 use crate::origin::TaskOrigin;
5385
5386 fn event_provenance() -> TaskProvenance {
5388 TaskProvenance::derive(TaskOrigin::Event, None)
5389 }
5390
5391 fn report_for(
5395 declared: ContainmentClass,
5396 tier: EnforcementTier,
5397 workdir_exec: bool,
5398 ) -> ScopeReport {
5399 crate::containment::report_for(
5400 &CapabilityPolicy {
5401 workdir_exec_allowed: workdir_exec,
5402 ..CapabilityPolicy::default()
5403 },
5404 declared,
5405 tier,
5406 None,
5407 None,
5408 None,
5409 None,
5410 Vec::new(),
5411 Vec::new(),
5412 Vec::new(),
5413 &crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
5414 crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(tier, declared)),
5415 crate::cgroup::IoMaxReport::default(),
5416 )
5417 }
5418
5419 fn demotion_info() -> crate::detached::DetachedDispatchInfo {
5420 crate::detached::DetachedDispatchInfo {
5421 work_id: "wrk_0a1b2c3d4e5f6a7b".to_string(),
5422 binary: "/usr/bin/bash".to_string(),
5423 command: "make -j8".to_string(),
5424 grace_ms: 1_000,
5425 }
5426 }
5427
5428 #[tokio::test]
5431 async fn a_demotion_marker_that_cannot_be_written_does_not_fail_the_demotion() {
5432 let dir = tempfile::tempdir().unwrap();
5433 let mut writer = make_writer(dir.path()).await;
5437 let full = std::fs::OpenOptions::new()
5438 .write(true)
5439 .open("/dev/full")
5440 .unwrap();
5441 writer.writer = BufWriter::new(File::from_std(full));
5442 let info = demotion_info();
5443
5444 assert!(
5445 writer
5446 .write_shell_detached(1, &info.work_id, &info.binary, &info.command, info.grace_ms)
5447 .await
5448 .is_err(),
5449 "the sink under this writer must be unwritable for the claim below to mean anything"
5450 );
5451 crate::agent::record_demotion(&mut writer, 1, &info).await;
5452 }
5453
5454 #[tokio::test]
5456 async fn a_recorded_demotion_writes_only_its_own_marker() {
5457 let dir = tempfile::tempdir().unwrap();
5458 let mut writer = make_writer(dir.path()).await;
5459 crate::agent::record_demotion(&mut writer, 3, &demotion_info()).await;
5460
5461 let events = read_events(dir.path());
5462 assert_eq!(events.len(), 1);
5463 assert_eq!(events[0]["event_type"], "shell_detached");
5464 assert_eq!(events[0]["work_id"], "wrk_0a1b2c3d4e5f6a7b");
5465 assert_eq!(events[0]["turn"], 3);
5466 }
5467
5468 #[tokio::test]
5469 async fn an_unrecorded_demotion_names_the_work_id_and_the_write_error() {
5470 let dir = tempfile::tempdir().unwrap();
5471 let mut writer = make_writer(dir.path()).await;
5472 writer
5473 .write_shell_detach_unrecorded(2, "wrk_0a1b", "/usr/bin/bash", "No space left")
5474 .await
5475 .unwrap();
5476
5477 let events = read_events(dir.path());
5478 assert_eq!(events.len(), 1);
5479 assert_eq!(events[0]["event_type"], "shell_detach_unrecorded");
5480 assert_eq!(events[0]["turn"], 2);
5481 assert_eq!(events[0]["work_id"], "wrk_0a1b");
5482 assert_eq!(events[0]["binary"], "/usr/bin/bash");
5483 assert_eq!(events[0]["reason"], "No space left");
5484 }
5485
5486 #[tokio::test]
5490 async fn an_abandoned_command_still_running_omits_the_fields_it_cannot_know() {
5491 let dir = tempfile::tempdir().unwrap();
5492 let mut writer = make_writer(dir.path()).await;
5493 writer
5494 .write_shell_abandoned(
5495 "wrk_0a1b2c3d4e5f6a7b",
5496 "/usr/bin/bash",
5497 "sleep 45; echo done",
5498 45_012,
5499 None,
5500 )
5501 .await
5502 .unwrap();
5503
5504 let events = read_events(dir.path());
5505 assert_eq!(events.len(), 1);
5506 let event = &events[0];
5507 assert_eq!(event["event_type"], "shell_abandoned");
5508 assert_eq!(event["work_id"], "wrk_0a1b2c3d4e5f6a7b");
5509 assert_eq!(event["binary"], "/usr/bin/bash");
5510 assert_eq!(event["command"], "sleep 45; echo done");
5511 assert_eq!(event["running_ms"], 45_012);
5512 for absent in ["exit_code", "output_path", "output_bytes"] {
5513 assert!(
5514 event.get(absent).is_none(),
5515 "{absent} must be omitted, not written as null: {event}"
5516 );
5517 }
5518 }
5519
5520 #[tokio::test]
5524 async fn an_abandoned_command_that_finished_carries_its_result() {
5525 let dir = tempfile::tempdir().unwrap();
5526 let mut writer = make_writer(dir.path()).await;
5527 writer
5528 .write_shell_abandoned(
5529 "wrk_0a1b2c3d4e5f6a7b",
5530 "/usr/bin/bash",
5531 "sleep 2; echo done",
5532 2_014,
5533 Some((
5534 0,
5535 "/home/test/.murmur/records/work/ses_1/logs/wrk_0a1b2c3d4e5f6a7b.log",
5536 91,
5537 )),
5538 )
5539 .await
5540 .unwrap();
5541
5542 let event = &read_events(dir.path())[0];
5543 assert_eq!(event["event_type"], "shell_abandoned");
5544 assert_eq!(event["running_ms"], 2_014);
5545 assert_eq!(event["exit_code"], 0);
5546 assert_eq!(
5547 event["output_path"],
5548 "/home/test/.murmur/records/work/ses_1/logs/wrk_0a1b2c3d4e5f6a7b.log"
5549 );
5550 assert_eq!(event["output_bytes"], 91);
5551 }
5552
5553 async fn make_writer(dir: &std::path::Path) -> TraceWriter {
5554 make_writer_with_opts(dir, TraceCapture::Meta).await
5555 }
5556
5557 async fn make_writer_with_opts(dir: &std::path::Path, capture: TraceCapture) -> TraceWriter {
5558 make_writer_with_prompt(dir, capture, None, false).await
5559 }
5560
5561 async fn make_writer_with_prompt(
5562 dir: &std::path::Path,
5563 capture: TraceCapture,
5564 system_prompt: Option<&str>,
5565 system_prompt_overridden: bool,
5566 ) -> TraceWriter {
5567 TraceWriter::open(
5568 &crate::workdir_fs::WorkdirRoot::open(dir).unwrap(),
5569 "test-session-id".to_string(),
5570 "test-capsule".to_string(),
5571 "1.0.0".to_string(),
5572 "claude-test".to_string(),
5573 vec!["shell".to_string()],
5574 report_for(
5575 ContainmentClass::Advisory,
5576 EnforcementTier::EnvironmentOnly,
5577 false,
5578 ),
5579 capture,
5580 system_prompt.map(str::to_string),
5581 system_prompt_overridden,
5582 None,
5583 None,
5584 None,
5585 None,
5586 )
5587 .await
5588 .unwrap()
5589 }
5590
5591 fn read_events(dir: &std::path::Path) -> Vec<Value> {
5592 let content = std::fs::read_to_string(dir.join("trace.jsonl")).unwrap();
5593 content
5594 .lines()
5595 .filter(|l| !l.is_empty())
5596 .map(|l| serde_json::from_str(l).unwrap())
5597 .collect()
5598 }
5599
5600 #[tokio::test]
5601 async fn session_start_fields() {
5602 let dir = tempfile::tempdir().unwrap();
5603 let mut w = make_writer(dir.path()).await;
5604 w.write_session_start(Some(10), None, vec!["bash".to_string()])
5605 .await
5606 .unwrap();
5607 w.flush().await.unwrap();
5608
5609 let events = read_events(dir.path());
5610 assert_eq!(events.len(), 1);
5611 let e = &events[0];
5612 assert_eq!(e["event_type"], "session_start");
5613 assert_eq!(e["session_id"], "test-session-id");
5614 assert_eq!(e["capsule_name"], "test-capsule");
5615 assert_eq!(e["capsule_version"], "1.0.0");
5616 assert_eq!(e["model"], "claude-test");
5617 assert_eq!(e["max_turns"], 10);
5618 assert_eq!(e["capabilities"], serde_json::json!(["shell"]));
5619 assert_eq!(e["tools_declared"], serde_json::json!(["bash"]));
5620 assert_eq!(e["containment_declared"], "advisory");
5621 assert_eq!(e["containment_achieved"], "advisory");
5622 assert_eq!(e["workdir_exec"], false);
5623 assert!(e["timestamp"].as_u64().unwrap() > 0);
5624 assert_eq!(e["model_limits"], serde_json::Value::Null);
5625 }
5626
5627 #[tokio::test]
5629 async fn session_start_writes_no_turn_cap_and_the_model_limits() {
5630 let dir = tempfile::tempdir().unwrap();
5631 let mut w = make_writer(dir.path()).await;
5632 let limits = crate::model_limits::ResolvedModelLimits {
5633 max_output_tokens: 65536,
5634 max_output_tokens_source: crate::model_limits::LimitSource::Driver,
5635 context_window: None,
5636 context_window_source: crate::model_limits::LimitSource::None,
5637 max_input_tokens: None,
5638 max_input_tokens_source: crate::model_limits::LimitSource::None,
5639 input_budget: None,
5640 input_budget_basis: crate::model_limits::BudgetBasis::None,
5641 };
5642 w.write_session_start(None, Some(limits), Vec::new())
5643 .await
5644 .unwrap();
5645 w.flush().await.unwrap();
5646
5647 let e = &read_events(dir.path())[0];
5648 assert!(e.get("max_turns").is_some_and(serde_json::Value::is_null));
5649 assert_eq!(
5650 e["model_limits"],
5651 serde_json::json!({
5652 "max_output_tokens": 65536,
5653 "max_output_tokens_source": "driver",
5654 "context_window": null,
5655 "context_window_source": "none",
5656 "max_input_tokens": null,
5657 "max_input_tokens_source": "none",
5658 "input_budget": null,
5659 "input_budget_basis": "none",
5660 })
5661 );
5662 }
5663
5664 #[tokio::test]
5667 async fn session_start_is_the_first_line_and_carries_a_members_formation() {
5668 let formation = FormationId::mint();
5669 for member in [Some(&formation), None] {
5670 let dir = tempfile::tempdir().unwrap();
5671 let mut w = make_writer(dir.path()).await;
5672 w.set_formation_id(member);
5673 w.write_session_start(Some(10), None, Vec::new())
5674 .await
5675 .unwrap();
5676 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
5677 .await
5678 .unwrap();
5679 w.flush().await.unwrap();
5680
5681 let content = std::fs::read_to_string(dir.path().join("trace.jsonl")).unwrap();
5682 let first_line = content.lines().next().unwrap();
5683 let first: Value = serde_json::from_str(first_line).unwrap();
5684 assert_eq!(first["event_type"], "session_start");
5685 match member {
5686 Some(id) => {
5687 assert_eq!(first["formation_id"], id.as_str());
5688 assert!(
5689 first_line.contains(&format!("\"formation_id\":\"{id}\"")),
5690 "{first_line}"
5691 );
5692 }
5693 None => assert!(!first_line.contains("formation_id"), "{first_line}"),
5694 }
5695 assert_eq!(
5697 content.matches("formation_id").count(),
5698 usize::from(member.is_some())
5699 );
5700 }
5701 }
5702
5703 #[tokio::test]
5707 async fn a_members_trace_names_it_its_callees_and_its_callers() {
5708 let authority = crate::formation_credentials::FormationAuthority::for_test();
5709 let cases = [
5710 (Some(("coder", vec!["reviewer"])), Some("planner")),
5711 (Some(("reviewer", vec![])), None),
5712 (None, None),
5713 ];
5714 for (member, caller) in cases {
5715 let member = member.map(|(name, callees)| {
5716 crate::formation_credentials::FormationMember::from_bundle(
5717 authority.member_bundle(name, &callees),
5718 )
5719 });
5720 let dir = tempfile::tempdir().unwrap();
5721 let mut w = make_writer(dir.path()).await;
5722 w.set_formation_member(member.as_ref());
5723 w.write_session_start(Some(10), None, Vec::new())
5724 .await
5725 .unwrap();
5726 w.write_a2a_task_received("tsk_1", "ctx_1", "msg_1", None, caller)
5727 .await
5728 .unwrap();
5729 w.flush().await.unwrap();
5730 let content = std::fs::read_to_string(dir.path().join("trace.jsonl")).unwrap();
5731 let lines: Vec<Value> = content
5732 .lines()
5733 .map(|line| serde_json::from_str(line).unwrap())
5734 .collect();
5735 let start = &lines[0];
5736 let received = lines
5737 .iter()
5738 .find(|line| line["event_type"] == "a2a_task_received")
5739 .unwrap();
5740 match &member {
5741 Some(member) => {
5742 assert_eq!(start["formation_member"], member.name());
5743 assert_eq!(
5744 start["formation_callees"],
5745 serde_json::json!(member.callees().collect::<Vec<_>>())
5746 );
5747 }
5748 None => {
5749 assert!(start.get("formation_member").is_none(), "{start}");
5750 assert!(start.get("formation_callees").is_none(), "{start}");
5751 }
5752 }
5753 match caller {
5754 Some(caller) => assert_eq!(received["caller_member"], caller),
5755 None => assert!(received.get("caller_member").is_none(), "{received}"),
5756 }
5757 assert!(!content.contains("mft1."), "{content}");
5758 }
5759 }
5760
5761 #[tokio::test]
5766 async fn session_start_records_system_prompt_source_and_hash() {
5767 let expected_sha = murmur_artifact::sha256_hex(b"Be terse.");
5768
5769 for (prompt, overridden, source, sha) in [
5770 (Some("Be terse."), false, "manifest", Some(&expected_sha)),
5771 (Some("Be terse."), true, "cli", Some(&expected_sha)),
5772 (None, false, "none", None),
5773 (None, true, "cli", None),
5776 ] {
5777 let dir = tempfile::tempdir().unwrap();
5778 let mut w =
5779 make_writer_with_prompt(dir.path(), TraceCapture::Meta, prompt, overridden).await;
5780 w.write_session_start(Some(1), None, Vec::new())
5781 .await
5782 .unwrap();
5783 w.flush().await.unwrap();
5784
5785 let e = &read_events(dir.path())[0];
5786 assert_eq!(e["system_prompt_source"], source, "prompt={prompt:?}");
5787 match sha {
5788 Some(sha) => assert_eq!(e["system_prompt_sha256"], sha.as_str()),
5789 None => assert!(
5790 e["system_prompt_sha256"].is_null(),
5791 "no prompt in effect must hash to null, got {}",
5792 e["system_prompt_sha256"]
5793 ),
5794 }
5795 }
5796 }
5797
5798 #[tokio::test]
5802 async fn session_start_records_the_prompt_by_hash_and_stores_it_only_under_content() {
5803 let sha = murmur_artifact::sha256_hex(b"Be terse.");
5804
5805 let dir = tempfile::tempdir().unwrap();
5806 let mut w =
5807 make_writer_with_prompt(dir.path(), TraceCapture::Meta, Some("Be terse."), false).await;
5808 w.write_session_start(Some(1), None, Vec::new())
5809 .await
5810 .unwrap();
5811 w.flush().await.unwrap();
5812
5813 let e = &read_events(dir.path())[0];
5814 assert!(
5815 e.get("system_prompt").is_none(),
5816 "the verbatim prompt is not a session_start field, got {e}"
5817 );
5818 assert_eq!(e["system_prompt_sha256"], sha.as_str());
5819 assert!(
5820 !dir.path().join(BLOB_DIR_NAME).exists(),
5821 "meta stores no bodies"
5822 );
5823
5824 let opted_in = tempfile::tempdir().unwrap();
5825 let mut w = make_writer_with_prompt(
5826 opted_in.path(),
5827 TraceCapture::Content,
5828 Some("Be terse."),
5829 false,
5830 )
5831 .await;
5832 w.write_session_start(Some(1), None, Vec::new())
5833 .await
5834 .unwrap();
5835 w.flush().await.unwrap();
5836
5837 let event = &read_events(opted_in.path())[0];
5838 assert!(event.get("system_prompt").is_none());
5839 assert_eq!(event["system_prompt_sha256"], sha.as_str());
5840 let blob = opted_in.path().join(BLOB_DIR_NAME).join(&sha);
5841 assert_eq!(std::fs::read(&blob).unwrap(), b"Be terse.");
5842 }
5843
5844 #[tokio::test]
5847 async fn content_capture_stores_no_prompt_blob_when_there_is_no_prompt() {
5848 let dir = tempfile::tempdir().unwrap();
5849 let mut w = make_writer_with_prompt(dir.path(), TraceCapture::Content, None, false).await;
5850 w.write_session_start(Some(1), None, Vec::new())
5851 .await
5852 .unwrap();
5853 w.flush().await.unwrap();
5854
5855 let e = &read_events(dir.path())[0];
5856 assert!(e.get("system_prompt").is_none());
5857 assert!(e["system_prompt_sha256"].is_null());
5858 assert!(!dir.path().join(BLOB_DIR_NAME).exists());
5859 }
5860
5861 #[tokio::test]
5863 async fn every_task_session_start_repeats_the_same_system_prompt_record() {
5864 let dir = tempfile::tempdir().unwrap();
5865 let mut w =
5866 make_writer_with_prompt(dir.path(), TraceCapture::Meta, Some("Be terse."), true).await;
5867 w.write_session_start(Some(1), None, Vec::new())
5868 .await
5869 .unwrap();
5870 w.write_session_start(Some(1), None, Vec::new())
5871 .await
5872 .unwrap();
5873 w.flush().await.unwrap();
5874
5875 let events = read_events(dir.path());
5876 assert_eq!(events.len(), 2);
5877 assert_eq!(
5878 events[0]["system_prompt_source"],
5879 events[1]["system_prompt_source"]
5880 );
5881 assert_eq!(
5882 events[0]["system_prompt_sha256"],
5883 events[1]["system_prompt_sha256"]
5884 );
5885 assert_eq!(events[1]["system_prompt_source"], "cli");
5886 }
5887
5888 #[tokio::test]
5893 async fn session_start_records_workdir_exec_next_to_the_class_it_forced() {
5894 let dir = tempfile::tempdir().unwrap();
5895 let mut w = TraceWriter::open(
5896 &crate::workdir_fs::WorkdirRoot::open(dir.path()).unwrap(),
5897 "test-session-id".to_string(),
5898 "test-capsule".to_string(),
5899 "1.0.0".to_string(),
5900 "claude-test".to_string(),
5901 Vec::new(),
5902 report_for(
5904 ContainmentClass::Advisory,
5905 EnforcementTier::KernelFull,
5906 true,
5907 ),
5908 TraceCapture::Meta,
5909 None,
5910 false,
5911 None,
5912 None,
5913 None,
5914 None,
5915 )
5916 .await
5917 .unwrap();
5918 w.write_session_start(Some(1), None, Vec::new())
5919 .await
5920 .unwrap();
5921 w.flush().await.unwrap();
5922
5923 let events = read_events(dir.path());
5924 assert_eq!(events[0]["workdir_exec"], true);
5925 assert_eq!(events[0]["containment_achieved"], "advisory");
5926 }
5927
5928 #[tokio::test]
5931 async fn session_start_records_declared_and_achieved_containment_separately() {
5932 let dir = tempfile::tempdir().unwrap();
5933 let mut w = TraceWriter::open(
5934 &crate::workdir_fs::WorkdirRoot::open(dir.path()).unwrap(),
5935 "test-session-id".to_string(),
5936 "test-capsule".to_string(),
5937 "1.0.0".to_string(),
5938 "claude-test".to_string(),
5939 Vec::new(),
5940 report_for(ContainmentClass::Sealed, EnforcementTier::KernelFull, false),
5941 TraceCapture::Meta,
5942 None,
5943 false,
5944 None,
5945 None,
5946 None,
5947 None,
5948 )
5949 .await
5950 .unwrap();
5951 w.write_session_start(Some(1), None, Vec::new())
5952 .await
5953 .unwrap();
5954 w.flush().await.unwrap();
5955
5956 let events = read_events(dir.path());
5957 assert_eq!(events[0]["containment_declared"], "sealed");
5958 assert_eq!(events[0]["containment_achieved"], "scoped");
5959 }
5960
5961 #[tokio::test]
5969 async fn session_start_distinguishes_two_hosts_with_the_same_achieved_class() {
5970 async fn event_for(grant: UsernsGrant) -> Value {
5971 let report = crate::containment::report_for(
5972 &CapabilityPolicy::default(),
5973 ContainmentClass::Sealed,
5974 EnforcementTier::KernelSealed,
5975 None,
5976 Some(grant),
5977 None,
5978 None,
5979 Vec::new(),
5980 Vec::new(),
5981 Vec::new(),
5982 &crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
5983 crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
5984 EnforcementTier::KernelSealed,
5985 ContainmentClass::Sealed,
5986 )),
5987 crate::cgroup::IoMaxReport::default(),
5988 );
5989 let dir = tempfile::tempdir().unwrap();
5990 let mut w = TraceWriter::open(
5991 &crate::workdir_fs::WorkdirRoot::open(dir.path()).unwrap(),
5992 "test-session-id".to_string(),
5993 "test-capsule".to_string(),
5994 "1.0.0".to_string(),
5995 "claude-test".to_string(),
5996 Vec::new(),
5997 report,
5998 TraceCapture::Meta,
5999 None,
6000 false,
6001 None,
6002 None,
6003 None,
6004 None,
6005 )
6006 .await
6007 .unwrap();
6008 w.write_session_start(Some(1), None, Vec::new())
6009 .await
6010 .unwrap();
6011 w.flush().await.unwrap();
6012 let mut event = read_events(dir.path()).remove(0);
6013 event.as_object_mut().unwrap().remove("timestamp");
6016 event
6017 }
6018
6019 let through_profile = event_for(UsernsGrant::ProfileConfining).await;
6020 let host_wide = event_for(UsernsGrant::RestrictionDisabledHostWide).await;
6021
6022 assert_eq!(through_profile["containment_achieved"], "sealed");
6023 assert_eq!(host_wide["containment_achieved"], "sealed");
6024 assert_eq!(through_profile["userns_grant"], "profile_confining");
6025 assert_eq!(host_wide["userns_grant"], "restriction_disabled_host_wide");
6026 assert_ne!(
6027 through_profile, host_wide,
6028 "two hosts granting the user namespace by different mechanisms must not write the \
6029 same session_start record"
6030 );
6031
6032 let unprobed = crate::containment::report_for(
6035 &CapabilityPolicy::default(),
6036 ContainmentClass::Advisory,
6037 EnforcementTier::EnvironmentOnly,
6038 None,
6039 None,
6040 None,
6041 None,
6042 Vec::new(),
6043 Vec::new(),
6044 Vec::new(),
6045 &crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
6046 crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
6047 EnforcementTier::EnvironmentOnly,
6048 ContainmentClass::Advisory,
6049 )),
6050 crate::cgroup::IoMaxReport::default(),
6051 );
6052 assert!(
6053 serde_json::to_value(&unprobed).unwrap()["userns_grant"].is_null(),
6054 "off Linux the key is present and null, not absent"
6055 );
6056 }
6057
6058 #[tokio::test]
6066 async fn session_start_records_the_whole_scope_report_as_effective_grants() {
6067 let policy = CapabilityPolicy {
6068 network_allow: vec!["https://api.example.com".to_string()],
6069 unix_sockets_allowed: false,
6070 filesystem_scope: Some("workdir".to_string()),
6071 shell_allow: vec!["python3".to_string()],
6072 spawn_allow: vec!["helper".to_string()],
6073 env_allow: vec!["TZ".to_string()],
6074 shell_interpreter_runtime: vec![InterpreterRuntimeGrant {
6075 binary: "python3".to_string(),
6076 dirs: vec![InterpreterRuntimeDir {
6077 path: "/usr/lib/python3.11".to_string(),
6078 list_dir: true,
6079 }],
6080 }],
6081 ..CapabilityPolicy::default()
6082 };
6083 let expected = crate::containment::report_for(
6084 &policy,
6085 ContainmentClass::Scoped,
6086 EnforcementTier::KernelFull,
6087 None,
6088 None,
6089 Some(8),
6091 None,
6092 vec![crate::containment::StateStoreReport {
6095 artifact: "notes-tool".to_string(),
6096 store: "shey".to_string(),
6097 host_path: "/home/dev/.murmur/state/shey".to_string(),
6098 }],
6099 vec!["config-echo".to_string()],
6101 vec![crate::containment::PreopenReport {
6102 artifact: "notes-tool".to_string(),
6103 role: "tool".to_string(),
6104 scope: None,
6105 surface: crate::containment::PreopenSurface::WholeWorkdir,
6106 runtime_dir_reach: Some(crate::containment::RuntimeDirReach::MurmurRuntime),
6107 }],
6108 &crate::runtime_dir::RuntimeDirPlacement::outside_for_test(),
6109 crate::temp_dir::tier_temp_mapping(crate::sandbox::applied_tier(
6110 EnforcementTier::KernelFull,
6111 ContainmentClass::Scoped,
6112 )),
6113 crate::cgroup::IoMaxReport::default(),
6114 );
6115
6116 let dir = tempfile::tempdir().unwrap();
6117 let mut w = TraceWriter::open(
6118 &crate::workdir_fs::WorkdirRoot::open(dir.path()).unwrap(),
6119 "test-session-id".to_string(),
6120 "test-capsule".to_string(),
6121 "1.0.0".to_string(),
6122 "claude-test".to_string(),
6123 vec!["network".to_string(), "shell".to_string()],
6124 expected.clone(),
6125 TraceCapture::Meta,
6126 None,
6127 false,
6128 None,
6129 None,
6130 None,
6131 None,
6132 )
6133 .await
6134 .unwrap();
6135 w.write_session_start(Some(1), None, Vec::new())
6136 .await
6137 .unwrap();
6138 w.flush().await.unwrap();
6139
6140 let events = read_events(dir.path());
6141 assert_eq!(
6142 events[0]["effective_grants"],
6143 serde_json::to_value(&expected).unwrap()
6144 );
6145
6146 assert_eq!(
6149 events[0]["capabilities"],
6150 serde_json::json!(["network", "shell"])
6151 );
6152 assert_eq!(
6153 events[0]["effective_grants"]["state_stores"],
6154 serde_json::json!([{
6155 "artifact": "notes-tool",
6156 "store": "shey",
6157 "host_path": "/home/dev/.murmur/state/shey",
6158 }])
6159 );
6160 assert_eq!(
6163 events[0]["effective_grants"]["configured_artifacts"],
6164 serde_json::json!(["config-echo"])
6165 );
6166 assert_eq!(
6170 events[0]["effective_grants"]["preopens"],
6171 serde_json::json!([{
6172 "artifact": "notes-tool",
6173 "role": "tool",
6174 "scope": null,
6175 "surface": "whole-workdir",
6176 "runtime_dir_reach": "murmur-runtime",
6177 }])
6178 );
6179 assert_eq!(events[0]["containment_declared"], "scoped");
6180 assert_eq!(events[0]["containment_achieved"], "scoped");
6181 assert_eq!(events[0]["workdir_exec"], false);
6182 assert_eq!(
6183 events[0]["effective_grants"]["network_allow"],
6184 serde_json::json!(["https://api.example.com"])
6185 );
6186 assert_eq!(
6187 events[0]["effective_grants"]["interpreter_runtime_grants"],
6188 serde_json::json!(["python3: /usr/lib/python3.11 (list_dir)"])
6189 );
6190 }
6191
6192 #[tokio::test]
6193 async fn inference_fields_and_snake_case() {
6194 let dir = tempfile::tempdir().unwrap();
6195 let mut w = make_writer(dir.path()).await;
6196 w.write_inference(
6197 0,
6198 Some(100),
6199 Some(50),
6200 "tool_call".to_string(),
6201 Some("tool_call"),
6202 Some("bash".to_string()),
6203 None,
6204 None,
6205 vec!["msg_0000000000000000000000000000000a".to_string()],
6206 None,
6207 None,
6208 )
6209 .await
6210 .unwrap();
6211 w.flush().await.unwrap();
6212
6213 let events = read_events(dir.path());
6214 let e = &events[0];
6215 assert_eq!(e["event_type"], "inference");
6216 assert_eq!(
6217 e["message_ids"],
6218 serde_json::json!(["msg_0000000000000000000000000000000a"]),
6219 "an agent-loop turn names the messages its request embedded"
6220 );
6221 assert_eq!(e["turn"], 0);
6222 assert_eq!(e["input_tokens"], 100);
6223 assert_eq!(e["output_tokens"], 50);
6224 assert_eq!(e["decision"], "tool_call");
6225 assert_eq!(e["tool_name"], "bash");
6226 assert!(e.get("inputTokens").is_none(), "must use snake_case");
6228 assert!(e.get("toolName").is_none(), "must use snake_case");
6229 for key in [
6231 "input_tokens_actual",
6232 "output_tokens_actual",
6233 "cached_tokens",
6234 "cache_write_tokens",
6235 ] {
6236 assert!(e.get(key).is_none(), "{key} must be omitted, not null or 0");
6237 }
6238 }
6239
6240 #[tokio::test]
6243 async fn inference_writes_only_the_reported_usage_members() {
6244 let dir = tempfile::tempdir().unwrap();
6245 let mut w = make_writer(dir.path()).await;
6246 let usage = DriverUsage {
6247 input_tokens: Some(12043),
6248 output_tokens: None,
6249 cached_tokens: Some(0),
6250 cache_write_tokens: None,
6251 thinking_tokens: None,
6252 };
6253 w.write_inference(
6254 0,
6255 Some(100),
6256 Some(50),
6257 "end_turn".to_string(),
6258 Some("end_turn"),
6259 None,
6260 None,
6261 Some(&usage),
6262 Vec::new(),
6263 None,
6264 None,
6265 )
6266 .await
6267 .unwrap();
6268 w.flush().await.unwrap();
6269
6270 let events = read_events(dir.path());
6271 let e = &events[0];
6272 assert_eq!(e["input_tokens"], 100, "the estimate is untouched");
6273 assert_eq!(e["input_tokens_actual"], 12043);
6274 assert_eq!(e["cached_tokens"], 0, "a reported zero is written");
6275 assert!(e.get("output_tokens_actual").is_none());
6276 assert!(e.get("cache_write_tokens").is_none());
6277 }
6278
6279 #[tokio::test]
6280 async fn failed_inference_writes_its_keys_and_counts_only_what_it_spent() {
6281 let dir = tempfile::tempdir().unwrap();
6282 let mut w = make_writer_with_opts(dir.path(), TraceCapture::Content).await;
6283 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
6284 .await
6285 .unwrap();
6286 let usage = DriverUsage {
6287 input_tokens: Some(7),
6288 ..DriverUsage::default()
6289 };
6290 let long = "é".repeat(MAX_TASK_FAILURE_REASON_BYTES);
6291 w.write_failed_inference(FailedInference {
6292 turn: 0,
6293 error_code: INFERENCE_ERROR_CREDENTIAL_REJECTED,
6294 error: &long,
6295 provider_status: Some(401),
6296 usage: Some(&usage),
6297 origin: None,
6298 choice: None,
6299 message_ids: vec!["msg_1".to_string()],
6300 tool_name: None,
6301 spent: None,
6302 settings: None,
6303 retry: RetryNote::default(),
6304 })
6305 .await
6306 .unwrap();
6307 assert_eq!(w.task_turns(), 0, "an uncounted record moves no counter");
6308 assert_eq!((w.total_turns, w.total_input_tokens), (0, 0));
6309 assert_eq!((w.task_input_tokens, w.task_output_tokens), (0, 0));
6310
6311 w.write_failed_inference(FailedInference {
6312 turn: 1,
6313 error_code: INFERENCE_ERROR_HARNESS_AUTH,
6314 error: "bad login",
6315 provider_status: None,
6316 usage: None,
6317 origin: None,
6318 choice: None,
6319 message_ids: Vec::new(),
6320 tool_name: None,
6321 spent: Some((Some(10), Some(5))),
6322 settings: None,
6323 retry: RetryNote::default(),
6324 })
6325 .await
6326 .unwrap();
6327 assert_eq!(w.task_turns(), 1);
6328 assert_eq!(w.total_turns, 1);
6329 assert_eq!((w.total_input_tokens, w.total_output_tokens), (10, 5));
6330 assert_eq!((w.task_input_tokens, w.task_output_tokens), (10, 5));
6331 w.flush().await.unwrap();
6332
6333 let events = read_events(dir.path());
6334 let failed: Vec<&Value> = events
6335 .iter()
6336 .filter(|e| e["event_type"] == "inference")
6337 .collect();
6338 let keys = |e: &Value| {
6339 let mut keys: Vec<String> = e.as_object().unwrap().keys().cloned().collect();
6340 keys.sort();
6341 keys
6342 };
6343 assert_eq!(
6344 keys(failed[0]),
6345 [
6346 "decision",
6347 "error",
6348 "error_code",
6349 "event_id",
6350 "event_type",
6351 "input_tokens_actual",
6352 "message_ids",
6353 "parent_id",
6354 "provider_status",
6355 "session_id",
6356 "stop_reason",
6357 "task_id",
6358 "timestamp",
6359 "tool_name",
6360 "turn",
6361 ]
6362 );
6363 assert_eq!(failed[0]["decision"], "error");
6364 assert_eq!(failed[0]["stop_reason"], "error");
6365 assert_eq!(failed[0]["error_code"], "credential_rejected");
6366 assert_eq!(failed[0]["provider_status"], 401);
6367 assert_eq!(failed[0]["input_tokens_actual"], 7);
6368 let error = failed[0]["error"].as_str().unwrap();
6369 assert!(error.len() <= MAX_TASK_FAILURE_REASON_BYTES);
6370 assert!(error.len() > MAX_TASK_FAILURE_REASON_BYTES - 2);
6371
6372 assert_eq!(failed[1]["error_code"], "harness_auth");
6373 assert_eq!(failed[1]["error"], "bad login");
6374 assert_eq!(failed[1]["input_tokens"], 10);
6375 assert_eq!(failed[1]["output_tokens"], 5);
6376 assert!(failed[1].get("provider_status").is_none());
6377 assert!(
6378 !dir.path().join("blobs").exists(),
6379 "a failed record stores no blob, even under trace.capture: content"
6380 );
6381 let task_start = events
6383 .iter()
6384 .find(|e| e["event_type"] == "task_start")
6385 .unwrap();
6386 assert_eq!(failed[0]["parent_id"], task_start["event_id"]);
6387 }
6388
6389 #[tokio::test]
6390 async fn a_failed_hook_record_hangs_off_the_turn_and_keeps_hook_accounting() {
6391 let dir = tempfile::tempdir().unwrap();
6392 let mut w = make_writer(dir.path()).await;
6393 w.write_inference(
6394 0,
6395 Some(10),
6396 Some(5),
6397 "end_turn".to_string(),
6398 Some("end_turn"),
6399 None,
6400 None,
6401 None,
6402 Vec::new(),
6403 None,
6404 None,
6405 )
6406 .await
6407 .unwrap();
6408 let origin = InferenceOrigin {
6409 source: "hook:compactor".to_string(),
6410 model: "m".to_string(),
6411 };
6412 w.write_failed_inference(FailedInference {
6413 turn: 0,
6414 error_code: INFERENCE_ERROR_DRIVER_ERROR,
6415 error: "HTTP 500: boom",
6416 provider_status: Some(500),
6417 usage: None,
6418 origin: Some(&origin),
6419 choice: None,
6420 message_ids: Vec::new(),
6421 tool_name: None,
6422 spent: Some((Some(40), None)),
6423 settings: None,
6424 retry: RetryNote::default(),
6425 })
6426 .await
6427 .unwrap();
6428 assert_eq!(w.total_turns, 2);
6429 assert_eq!((w.total_input_tokens, w.total_output_tokens), (50, 5));
6430 w.flush().await.unwrap();
6431
6432 let events = read_events(dir.path());
6433 let turn = events.iter().find(|e| e["decision"] == "end_turn").unwrap();
6434 let hook = events.iter().find(|e| e["decision"] == "error").unwrap();
6435 assert_eq!(hook["parent_id"], turn["event_id"]);
6436 assert_eq!(hook["origin"], "hook:compactor");
6437 assert_eq!(hook["model"], "m");
6438 assert_eq!(hook["input_tokens"], 40);
6439 assert!(hook.get("output_tokens").is_none());
6440 }
6441
6442 #[tokio::test]
6446 async fn inference_call_settings_are_written_on_every_kind_of_http_line() {
6447 let dir = tempfile::tempdir().unwrap();
6448 let mut w = make_writer(dir.path()).await;
6449 let turn = InferenceCallSettings {
6450 max_output_tokens: 4096,
6451 input_budget: Some(195_904),
6452 driver_config: Some(serde_json::json!({"reasoning_effort": "high"})),
6453 };
6454 let unconfigured = InferenceCallSettings {
6455 max_output_tokens: 1024,
6456 input_budget: None,
6457 driver_config: None,
6458 };
6459 let hook = InferenceCallSettings {
6460 max_output_tokens: 2000,
6461 input_budget: Some(198_000),
6462 driver_config: Some(serde_json::json!({"reasoning_effort": "low"})),
6463 };
6464 let origin = InferenceOrigin {
6465 source: "hook:compactor".to_string(),
6466 model: "m".to_string(),
6467 };
6468 w.write_turn_inference(
6469 0,
6470 Some(10),
6471 Some(5),
6472 "end_turn".to_string(),
6473 Some("end_turn"),
6474 None,
6475 None,
6476 Vec::new(),
6477 None,
6478 None,
6479 Some(&turn),
6480 RetryNote::default(),
6481 )
6482 .await
6483 .unwrap();
6484 w.write_turn_inference(
6485 1,
6486 Some(10),
6487 Some(5),
6488 "end_turn".to_string(),
6489 Some("end_turn"),
6490 None,
6491 None,
6492 Vec::new(),
6493 None,
6494 None,
6495 Some(&unconfigured),
6496 RetryNote::default(),
6497 )
6498 .await
6499 .unwrap();
6500 w.write_inference(
6501 1,
6502 Some(30),
6503 Some(8),
6504 "end_turn".to_string(),
6505 None,
6506 None,
6507 Some(&origin),
6508 None,
6509 Vec::new(),
6510 None,
6511 Some(&hook),
6512 )
6513 .await
6514 .unwrap();
6515 w.write_failed_inference(FailedInference {
6516 turn: 2,
6517 error_code: INFERENCE_ERROR_DRIVER_ERROR,
6518 error: "HTTP 500: boom",
6519 provider_status: Some(500),
6520 usage: None,
6521 origin: None,
6522 choice: None,
6523 message_ids: Vec::new(),
6524 tool_name: None,
6525 spent: None,
6526 settings: Some(&turn),
6527 retry: RetryNote::default(),
6528 })
6529 .await
6530 .unwrap();
6531 w.flush().await.unwrap();
6532
6533 let events = read_events(dir.path());
6534 let settings: Vec<(Value, Value, Value, Value)> = events
6535 .iter()
6536 .filter(|e| e["event_type"] == "inference")
6537 .map(|e| {
6538 (
6539 e["decision"].clone(),
6540 e["max_output_tokens"].clone(),
6541 e.get("input_budget")
6542 .cloned()
6543 .expect("input_budget is written, null or not"),
6544 e.get("driver_config")
6545 .cloned()
6546 .expect("driver_config is written, null or not"),
6547 )
6548 })
6549 .collect();
6550 assert_eq!(
6551 settings,
6552 [
6553 (
6554 serde_json::json!("end_turn"),
6555 serde_json::json!(4096),
6556 serde_json::json!(195_904),
6557 serde_json::json!({"reasoning_effort": "high"})
6558 ),
6559 (
6560 serde_json::json!("end_turn"),
6561 serde_json::json!(1024),
6562 Value::Null,
6563 Value::Null
6564 ),
6565 (
6566 serde_json::json!("end_turn"),
6567 serde_json::json!(2000),
6568 serde_json::json!(198_000),
6569 serde_json::json!({"reasoning_effort": "low"})
6570 ),
6571 (
6572 serde_json::json!("error"),
6573 serde_json::json!(4096),
6574 serde_json::json!(195_904),
6575 serde_json::json!({"reasoning_effort": "high"})
6576 ),
6577 ]
6578 );
6579 }
6580
6581 #[tokio::test]
6583 async fn an_inference_line_without_call_settings_writes_neither_key() {
6584 let dir = tempfile::tempdir().unwrap();
6585 let mut w = make_writer(dir.path()).await;
6586 w.write_inference(
6587 0,
6588 Some(10),
6589 Some(5),
6590 "end_turn".to_string(),
6591 None,
6592 None,
6593 None,
6594 None,
6595 Vec::new(),
6596 None,
6597 None,
6598 )
6599 .await
6600 .unwrap();
6601 w.write_failed_inference(FailedInference {
6602 turn: 1,
6603 error_code: INFERENCE_ERROR_HARNESS_AUTH,
6604 error: "bad login",
6605 provider_status: None,
6606 usage: None,
6607 origin: None,
6608 choice: None,
6609 message_ids: Vec::new(),
6610 tool_name: None,
6611 spent: None,
6612 settings: None,
6613 retry: RetryNote::default(),
6614 })
6615 .await
6616 .unwrap();
6617 w.flush().await.unwrap();
6618
6619 let events = read_events(dir.path());
6620 let lines: Vec<&Value> = events
6621 .iter()
6622 .filter(|e| e["event_type"] == "inference")
6623 .collect();
6624 assert_eq!(lines.len(), 2);
6625 for line in lines {
6626 assert!(line.get("max_output_tokens").is_none(), "{line}");
6627 assert!(line.get("input_budget").is_none(), "{line}");
6628 assert!(line.get("driver_config").is_none(), "{line}");
6629 }
6630 }
6631
6632 fn driver_config_env(value: &str) -> Vec<(String, String)> {
6633 vec![
6634 ("MURMUR_INFERENCE_MODEL".to_string(), "gpt-5".to_string()),
6635 (
6636 "MURMUR_INFERENCE_DRIVER_CONFIG".to_string(),
6637 value.to_string(),
6638 ),
6639 ]
6640 }
6641
6642 fn redact_secret(text: &str) -> String {
6643 text.replace("sk-secret", crate::gateway_credential::REDACTED)
6644 }
6645
6646 #[test]
6647 fn recorded_driver_config_is_none_without_the_variable() {
6648 let env = vec![("MURMUR_INFERENCE_MODEL".to_string(), "gpt-5".to_string())];
6649 assert_eq!(recorded_driver_config(&env, redact_secret), None);
6650 }
6651
6652 #[test]
6653 fn recorded_driver_config_is_the_parsed_object() {
6654 assert_eq!(
6655 recorded_driver_config(
6656 &driver_config_env(r#"{"reasoning_effort":"low"}"#),
6657 redact_secret
6658 ),
6659 Some(serde_json::json!({"reasoning_effort": "low"}))
6660 );
6661 }
6662
6663 #[test]
6664 fn recorded_driver_config_redacts_keys_and_string_leaves_at_any_depth() {
6665 let env = driver_config_env(
6666 r#"{"note":"x-sk-secret-y","nested":{"list":["sk-secret"]},"sk-secret":1}"#,
6667 );
6668 assert_eq!(
6669 recorded_driver_config(&env, redact_secret),
6670 Some(serde_json::json!({
6671 "note": "x-[redacted]-y",
6672 "nested": {"list": ["[redacted]"]},
6673 "[redacted]": 1,
6674 }))
6675 );
6676 }
6677
6678 #[test]
6679 fn recorded_driver_config_replaces_a_number_the_redactor_changes() {
6680 let redact_pin = |text: &str| text.replace("424242", crate::gateway_credential::REDACTED);
6681 let env = driver_config_env(r#"{"pin":424242,"retries":3,"store":false,"seed":null}"#);
6682 assert_eq!(
6683 recorded_driver_config(&env, redact_pin),
6684 Some(serde_json::json!({
6685 "pin": "[redacted]",
6686 "retries": 3,
6687 "store": false,
6688 "seed": null,
6689 }))
6690 );
6691 }
6692
6693 #[test]
6694 fn recorded_driver_config_records_text_that_is_not_json_as_a_redacted_string() {
6695 assert_eq!(
6696 recorded_driver_config(&driver_config_env("effort=sk-secret"), redact_secret),
6697 Some(Value::String("effort=[redacted]".to_string()))
6698 );
6699 }
6700
6701 #[tokio::test]
6702 async fn tool_call_fields() {
6703 let dir = tempfile::tempdir().unwrap();
6704 let mut w = make_writer(dir.path()).await;
6705 w.write_tool_call(
6706 1,
6707 "bash".to_string(),
6708 None,
6709 serde_json::json!({"command": "echo hi"}),
6710 42,
6711 "hi\n",
6712 100,
6713 55,
6714 "ok".to_string(),
6715 None,
6716 None,
6717 None,
6718 )
6719 .await
6720 .unwrap();
6721 w.flush().await.unwrap();
6722
6723 let events = read_events(dir.path());
6724 let e = &events[0];
6725 assert_eq!(e["event_type"], "tool_call");
6726 assert_eq!(e["turn"], 1);
6727 assert_eq!(e["tool_name"], "bash");
6728 assert_eq!(e["input_bytes"], 42);
6729 assert_eq!(e["output_bytes"], 100);
6730 assert_eq!(e["duration_ms"], 55);
6731 assert_eq!(e["status"], "ok");
6732 assert!(
6733 e.get("state_effect").is_none(),
6734 "state_effect must be omitted when the tool declared nothing"
6735 );
6736 }
6737
6738 #[tokio::test]
6739 async fn tool_call_records_declared_state_effect() {
6740 let dir = tempfile::tempdir().unwrap();
6741 let mut w = make_writer(dir.path()).await;
6742 w.write_tool_call(
6743 0,
6744 "murmur-tool-editor".to_string(),
6745 None,
6746 serde_json::json!({"operation": "read_file", "path": "a.txt"}),
6747 10,
6748 "hi\n",
6749 3,
6750 5,
6751 "ok".to_string(),
6752 Some("read".to_string()),
6753 None,
6754 None,
6755 )
6756 .await
6757 .unwrap();
6758 w.flush().await.unwrap();
6759
6760 let events = read_events(dir.path());
6761 assert_eq!(
6762 events[0]["state_effect"], "read",
6763 "declared state_effect must be recorded verbatim on the tool_call event"
6764 );
6765 }
6766
6767 #[tokio::test]
6768 async fn tool_call_records_declared_resource_id() {
6769 let dir = tempfile::tempdir().unwrap();
6770 let mut w = make_writer(dir.path()).await;
6771 w.write_tool_call(
6772 0,
6773 "murmur-tool-code-graph".to_string(),
6774 None,
6775 serde_json::json!({"symbol": "Foo::bar"}),
6776 10,
6777 "hi\n",
6778 3,
6779 5,
6780 "ok".to_string(),
6781 Some("read".to_string()),
6782 Some("sym:Foo".to_string()),
6783 None,
6784 )
6785 .await
6786 .unwrap();
6787 w.flush().await.unwrap();
6788
6789 let events = read_events(dir.path());
6790 assert_eq!(
6791 events[0]["resource_id"], "sym:Foo",
6792 "declared resource_id must reach trace.jsonl verbatim, unparsed"
6793 );
6794 }
6795
6796 #[tokio::test]
6797 async fn tool_call_omits_undeclared_resource_id() {
6798 let dir = tempfile::tempdir().unwrap();
6799 let mut w = make_writer(dir.path()).await;
6800 w.write_tool_call(
6801 0,
6802 "bash".to_string(),
6803 None,
6804 serde_json::json!({"command": "echo hi"}),
6805 10,
6806 "hi\n",
6807 3,
6808 5,
6809 "ok".to_string(),
6810 None,
6811 None,
6812 None,
6813 )
6814 .await
6815 .unwrap();
6816 w.flush().await.unwrap();
6817
6818 let events = read_events(dir.path());
6819 assert!(
6820 events[0].get("resource_id").is_none(),
6821 "resource_id must be omitted entirely (not null) when the tool declared nothing"
6822 );
6823 }
6824
6825 #[tokio::test]
6826 async fn shell_fields() {
6827 let dir = tempfile::tempdir().unwrap();
6828 let mut w = make_writer(dir.path()).await;
6829 w.write_shell(
6830 0,
6831 "/usr/bin/echo".to_string(),
6832 "echo hi".to_string(),
6833 0,
6834 7,
6835 0,
6836 10,
6837 None,
6838 )
6839 .await
6840 .unwrap();
6841 w.flush().await.unwrap();
6842
6843 let events = read_events(dir.path());
6844 let e = &events[0];
6845 assert_eq!(e["event_type"], "shell");
6846 assert_eq!(e["binary"], "/usr/bin/echo");
6849 assert_eq!(e["command"], "echo hi");
6850 assert_eq!(e["exit_code"], 0);
6851 assert_eq!(e["stdout_bytes"], 7);
6852 assert_eq!(e["stderr_bytes"], 0);
6853 assert_eq!(e["duration_ms"], 10);
6854 }
6855
6856 #[tokio::test]
6857 async fn compaction_fields() {
6858 let dir = tempfile::tempdir().unwrap();
6859 let mut w = make_writer(dir.path()).await;
6860 w.write_compaction(3, 80000, OccupancyBasis::Actual, 20000)
6861 .await
6862 .unwrap();
6863 w.write_compaction(4, 30000, OccupancyBasis::Estimate, 10000)
6864 .await
6865 .unwrap();
6866 w.flush().await.unwrap();
6867
6868 let events = read_events(dir.path());
6869 let e = &events[0];
6870 assert_eq!(e["event_type"], "compaction");
6871 assert_eq!(e["turn"], 3);
6872 assert_eq!(e["tokens_before"], 80000);
6873 assert_eq!(e["tokens_before_source"], "actual");
6874 assert_eq!(e["tokens_after"], 20000);
6875 assert_eq!(events[1]["tokens_before_source"], "estimate");
6876 }
6877
6878 #[tokio::test]
6879 async fn hook_dispatch_error_fields() {
6880 let dir = tempfile::tempdir().unwrap();
6881 let mut w = make_writer(dir.path()).await;
6882 w.write_hook_dispatch_error("my-hook", "on-tool-call", "write-manifests")
6883 .await
6884 .unwrap();
6885 w.flush().await.unwrap();
6886
6887 let events = read_events(dir.path());
6888 let e = &events[0];
6889 assert_eq!(e["event_type"], "hook_dispatch_error");
6890 assert_eq!(e["session_id"], "test-session-id");
6891 assert_eq!(e["hook_name"], "my-hook");
6892 assert_eq!(e["event"], "on-tool-call");
6893 assert_eq!(e["arm"], "write-manifests");
6894 assert!(e["timestamp"].as_u64().unwrap() > 0);
6895 assert!(e.get("hookName").is_none(), "must use snake_case");
6897 }
6898
6899 #[tokio::test]
6900 async fn session_end_fields_and_snake_case() {
6901 let dir = tempfile::tempdir().unwrap();
6902 let mut w = make_writer(dir.path()).await;
6903 w.write_inference(
6904 0,
6905 Some(100),
6906 Some(50),
6907 "tool_call".to_string(),
6908 Some("tool_call"),
6909 None,
6910 None,
6911 None,
6912 Vec::new(),
6913 None,
6914 None,
6915 )
6916 .await
6917 .unwrap();
6918 w.write_tool_call(
6919 0,
6920 "bash".to_string(),
6921 None,
6922 serde_json::json!({"command": "ls"}),
6923 10,
6924 "file.txt\n",
6925 20,
6926 5,
6927 "ok".to_string(),
6928 None,
6929 None,
6930 None,
6931 )
6932 .await
6933 .unwrap();
6934 w.write_shell(0, "/bin/ls".to_string(), "ls".to_string(), 0, 3, 0, 2, None)
6935 .await
6936 .unwrap();
6937 w.write_session_end("ok").await.unwrap();
6938 w.flush().await.unwrap();
6939
6940 let events = read_events(dir.path());
6941 let se = events.last().unwrap();
6942 assert_eq!(se["event_type"], "session_end");
6943 assert_eq!(se["total_turns"], 1);
6944 assert_eq!(se["total_input_tokens"], 100);
6945 assert_eq!(se["total_output_tokens"], 50);
6946 assert_eq!(se["total_tool_calls"], 1);
6947 assert_eq!(se["total_shell_calls"], 1);
6948 assert_eq!(se["exit_status"], "ok");
6949 assert!(se["duration_ms"].as_u64().is_some());
6950 assert!(se.get("exitStatus").is_none(), "must use snake_case");
6952 assert!(se.get("totalTurns").is_none(), "must use snake_case");
6953 }
6954
6955 #[tokio::test]
6956 async fn write_session_end_if_not_ended_is_noop_when_already_ended() {
6957 let dir = tempfile::tempdir().unwrap();
6958 let mut w = make_writer(dir.path()).await;
6959 w.write_session_start(Some(10), None, vec![]).await.unwrap();
6960 w.write_session_end("ok").await.unwrap();
6961 w.write_session_end_if_not_ended("failed").await.unwrap();
6962 w.flush().await.unwrap();
6963
6964 let events = read_events(dir.path());
6965 assert_eq!(
6966 events
6967 .iter()
6968 .filter(|e| e["event_type"] == "session_end")
6969 .count(),
6970 1,
6971 "session_end should appear exactly once"
6972 );
6973 }
6974
6975 #[tokio::test]
6976 async fn write_session_end_if_not_ended_writes_when_started_not_ended() {
6977 let dir = tempfile::tempdir().unwrap();
6978 let mut w = make_writer(dir.path()).await;
6979 w.write_session_start(Some(10), None, vec![]).await.unwrap();
6980 w.write_session_end_if_not_ended("failed").await.unwrap();
6982 w.flush().await.unwrap();
6983
6984 let events = read_events(dir.path());
6985 let se = events
6986 .iter()
6987 .find(|e| e["event_type"] == "session_end")
6988 .unwrap();
6989 assert_eq!(se["exit_status"], "failed");
6990 }
6991
6992 #[tokio::test]
6993 async fn no_session_end_written_when_session_never_started() {
6994 let dir = tempfile::tempdir().unwrap();
6995 let mut w = make_writer(dir.path()).await;
6996 w.write_session_end_if_not_ended("failed").await.unwrap();
6998 w.flush().await.unwrap();
6999
7000 let events = read_events(dir.path());
7001 assert!(
7002 events.iter().all(|e| e["event_type"] != "session_end"),
7003 "session_end must not appear when session_start was never written"
7004 );
7005 }
7006
7007 #[tokio::test]
7008 async fn session_id_on_every_event() {
7009 let dir = tempfile::tempdir().unwrap();
7010 let mut w = make_writer(dir.path()).await;
7011 w.write_session_start(Some(10), None, vec!["bash".to_string()])
7012 .await
7013 .unwrap();
7014 w.write_inference(
7015 0,
7016 Some(10),
7017 Some(5),
7018 "end_turn".to_string(),
7019 Some("end_turn"),
7020 None,
7021 None,
7022 None,
7023 Vec::new(),
7024 None,
7025 None,
7026 )
7027 .await
7028 .unwrap();
7029 w.write_session_end("ok").await.unwrap();
7030 w.flush().await.unwrap();
7031
7032 let events = read_events(dir.path());
7033 for e in &events {
7034 assert_eq!(e["session_id"], "test-session-id");
7035 }
7036 }
7037
7038 #[tokio::test]
7039 async fn counts_match_per_event_sum() {
7040 let dir = tempfile::tempdir().unwrap();
7041 let mut w = make_writer(dir.path()).await;
7042 w.write_session_start(Some(10), None, vec![]).await.unwrap();
7043 w.write_inference(
7044 0,
7045 Some(50),
7046 Some(25),
7047 "tool_call".to_string(),
7048 Some("tool_call"),
7049 Some("bash".to_string()),
7050 None,
7051 None,
7052 vec!["msg_0000000000000000000000000000000a".to_string()],
7053 None,
7054 None,
7055 )
7056 .await
7057 .unwrap();
7058 w.write_tool_call(
7059 0,
7060 "bash".to_string(),
7061 None,
7062 serde_json::json!({"command": "echo"}),
7063 5,
7064 "ok",
7065 10,
7066 3,
7067 "ok".to_string(),
7068 None,
7069 None,
7070 None,
7071 )
7072 .await
7073 .unwrap();
7074 w.write_shell(
7075 0,
7076 "/bin/echo".to_string(),
7077 "echo".to_string(),
7078 0,
7079 4,
7080 0,
7081 1,
7082 None,
7083 )
7084 .await
7085 .unwrap();
7086 w.write_inference(
7087 1,
7088 Some(60),
7089 Some(30),
7090 "end_turn".to_string(),
7091 Some("end_turn"),
7092 None,
7093 None,
7094 None,
7095 Vec::new(),
7096 None,
7097 None,
7098 )
7099 .await
7100 .unwrap();
7101 w.write_session_end("ok").await.unwrap();
7102 w.flush().await.unwrap();
7103
7104 let events = read_events(dir.path());
7105 let inference_count = events
7106 .iter()
7107 .filter(|e| e["event_type"] == "inference")
7108 .count();
7109 let tool_count = events
7110 .iter()
7111 .filter(|e| e["event_type"] == "tool_call")
7112 .count();
7113 let shell_count = events.iter().filter(|e| e["event_type"] == "shell").count();
7114
7115 let se = events
7116 .iter()
7117 .find(|e| e["event_type"] == "session_end")
7118 .unwrap();
7119 assert_eq!(
7120 se["total_turns"].as_u64().unwrap() as usize,
7121 inference_count
7122 );
7123 assert_eq!(
7124 se["total_tool_calls"].as_u64().unwrap() as usize,
7125 tool_count
7126 );
7127 assert_eq!(
7128 se["total_shell_calls"].as_u64().unwrap() as usize,
7129 shell_count
7130 );
7131 }
7132
7133 #[tokio::test]
7134 async fn tool_call_input_always_present() {
7135 let dir = tempfile::tempdir().unwrap();
7136 let mut w = make_writer(dir.path()).await; w.write_tool_call(
7138 0,
7139 "bash".to_string(),
7140 None,
7141 serde_json::json!({"command": "ls -la"}),
7142 20,
7143 "total 8\n",
7144 80,
7145 12,
7146 "ok".to_string(),
7147 None,
7148 None,
7149 None,
7150 )
7151 .await
7152 .unwrap();
7153 w.flush().await.unwrap();
7154
7155 let events = read_events(dir.path());
7156 let e = &events[0];
7157 assert_eq!(e["event_type"], "tool_call");
7158 assert_eq!(e["input"]["command"], "ls -la");
7159 assert!(e.get("input").is_some(), "input must always be present");
7160 }
7161
7162 async fn tool_call_with_error(
7164 dir: &std::path::Path,
7165 capture: TraceCapture,
7166 error: Option<&str>,
7167 ) -> Value {
7168 let mut w = make_writer_with_opts(dir, capture).await;
7169 w.write_tool_call(
7170 0,
7171 "tool-fixture".to_string(),
7172 Some("toolu_1".to_string()),
7173 serde_json::json!({}),
7174 2,
7175 "native tool execution failed",
7176 28,
7177 3,
7178 "error".to_string(),
7179 None,
7180 None,
7181 error,
7182 )
7183 .await
7184 .unwrap();
7185 w.flush().await.unwrap();
7186 read_events(dir).remove(0)
7187 }
7188
7189 #[tokio::test]
7190 async fn tool_call_error_is_written_without_output_under_meta() {
7191 let dir = tempfile::tempdir().unwrap();
7192 let text = "sandbox: native tool enforcement setup failed before exec: Landlock: \
7193 Too many open files (os error 24)";
7194 let e = tool_call_with_error(dir.path(), TraceCapture::Meta, Some(text)).await;
7195 assert_eq!(e["error"], text);
7196 assert!(e.get("output").is_none(), "meta stores no output: {e}");
7197 }
7198
7199 #[tokio::test]
7200 async fn tool_call_error_is_written_under_every_capture() {
7201 for capture in [
7202 TraceCapture::None,
7203 TraceCapture::Meta,
7204 TraceCapture::Content,
7205 ] {
7206 let dir = tempfile::tempdir().unwrap();
7207 let e = tool_call_with_error(dir.path(), capture, Some("spawn failed")).await;
7208 assert_eq!(e["error"], "spawn failed", "{capture:?}: {e}");
7209 }
7210 }
7211
7212 #[tokio::test]
7213 async fn tool_call_error_is_capped_on_a_char_boundary() {
7214 let dir = tempfile::tempdir().unwrap();
7215 let long = format!("a{}", "é".repeat(2_999));
7218 assert_eq!(long.chars().count(), 3_000);
7219 let e = tool_call_with_error(dir.path(), TraceCapture::Meta, Some(&long)).await;
7220 let error = e["error"].as_str().unwrap();
7221 assert!(error.len() <= MAX_TASK_FAILURE_REASON_BYTES);
7222 assert!(error.len() > MAX_TASK_FAILURE_REASON_BYTES - 2);
7223 assert!(long.starts_with(error));
7224 }
7225
7226 #[tokio::test]
7227 async fn tool_call_error_redacts_a_peer_handle() {
7228 let dir = tempfile::tempdir().unwrap();
7229 let token = crate::peer_handoff::mint(
7230 &crate::peer_handoff::PeerMintKey::generate().unwrap(),
7231 &crate::peer_handoff::HandlePayload {
7232 v: 1,
7233 iss: "ses_1".to_string(),
7234 p: "report.md".to_string(),
7235 exp: timestamp_ms() + 3_600_000,
7236 n: "0123456789abcdef0123456789abcdef".to_string(),
7237 },
7238 "a@b",
7239 )
7240 .unwrap();
7241 let text = format!("fetch-peer-file {token} failed");
7242 let e = tool_call_with_error(dir.path(), TraceCapture::Meta, Some(&text)).await;
7243 let id = crate::peer_handoff::handle_id(&token);
7244 assert_eq!(e["error"], format!("fetch-peer-file <handle:{id}> failed"));
7245 }
7246
7247 #[tokio::test]
7248 async fn tool_call_without_an_error_has_no_error_key() {
7249 let dir = tempfile::tempdir().unwrap();
7250 let e = tool_call_with_error(dir.path(), TraceCapture::Meta, None).await;
7251 assert!(e.get("error").is_none(), "{e}");
7252 }
7253
7254 #[tokio::test]
7255 async fn tool_call_output_absent_by_default() {
7256 let dir = tempfile::tempdir().unwrap();
7257 let mut w = make_writer(dir.path()).await; w.write_tool_call(
7259 0,
7260 "bash".to_string(),
7261 None,
7262 serde_json::json!({"command": "ls"}),
7263 10,
7264 "file.txt\n",
7265 90,
7266 8,
7267 "ok".to_string(),
7268 None,
7269 None,
7270 None,
7271 )
7272 .await
7273 .unwrap();
7274 w.flush().await.unwrap();
7275
7276 let events = read_events(dir.path());
7277 let e = &events[0];
7278 assert!(
7279 e.get("output").is_none(),
7280 "output must be absent under a capture mode that stores no bodies"
7281 );
7282 assert_eq!(e["output_bytes"], 90, "output_bytes must still be recorded");
7283 }
7284
7285 #[tokio::test]
7286 async fn skill_call_writes_skill_call_event() {
7287 let dir = tempfile::tempdir().unwrap();
7288 let mut w = make_writer(dir.path()).await;
7289 w.write_skill_call(
7290 2,
7291 "my-skill".to_string(),
7292 &LockOrigin::Operator,
7293 1024,
7294 8,
7295 "ok".to_string(),
7296 )
7297 .await
7298 .unwrap();
7299 w.flush().await.unwrap();
7300
7301 let events = read_events(dir.path());
7302 let e = &events[0];
7303 assert_eq!(e["event_type"], "skill_call");
7304 assert_eq!(e["turn"], 2);
7305 assert_eq!(e["skill_name"], "my-skill");
7306 assert_eq!(e["output_bytes"], 1024);
7307 assert_eq!(e["duration_ms"], 8);
7308 assert_eq!(e["status"], "ok");
7309 }
7310
7311 #[tokio::test]
7312 async fn skill_call_does_not_increment_total_tool_calls() {
7313 let dir = tempfile::tempdir().unwrap();
7314 let mut w = make_writer(dir.path()).await;
7315 w.write_session_start(Some(10), None, vec![]).await.unwrap();
7316 w.write_tool_call(
7317 0,
7318 "real-tool".to_string(),
7319 None,
7320 serde_json::json!({}),
7321 2,
7322 "data",
7323 4,
7324 5,
7325 "ok".to_string(),
7326 None,
7327 None,
7328 None,
7329 )
7330 .await
7331 .unwrap();
7332 w.write_skill_call(
7333 1,
7334 "my-skill".to_string(),
7335 &LockOrigin::Operator,
7336 512,
7337 3,
7338 "ok".to_string(),
7339 )
7340 .await
7341 .unwrap();
7342 w.write_session_end("ok").await.unwrap();
7343 w.flush().await.unwrap();
7344
7345 let events = read_events(dir.path());
7346 let se = events
7347 .iter()
7348 .find(|e| e["event_type"] == "session_end")
7349 .unwrap();
7350 assert_eq!(
7351 se["total_tool_calls"], 1,
7352 "skill_call must not inflate total_tool_calls"
7353 );
7354 }
7355
7356 #[tokio::test]
7357 async fn tool_call_output_present_when_opted_in() {
7358 let dir = tempfile::tempdir().unwrap();
7359 let mut w = make_writer_with_opts(dir.path(), TraceCapture::Content).await;
7360 w.write_tool_call(
7361 0,
7362 "bash".to_string(),
7363 None,
7364 serde_json::json!({"command": "echo hello"}),
7365 22,
7366 "hello\n",
7367 60,
7368 5,
7369 "ok".to_string(),
7370 None,
7371 None,
7372 None,
7373 )
7374 .await
7375 .unwrap();
7376 w.flush().await.unwrap();
7377
7378 let events = read_events(dir.path());
7379 let e = &events[0];
7380 assert_eq!(
7381 e["output"].as_str().unwrap(),
7382 "hello\n",
7383 "output must be present under capture: content"
7384 );
7385 assert_eq!(e["output_bytes"], 60);
7386 }
7387
7388 #[tokio::test]
7389 async fn task_end_carries_reopen_count() {
7390 let dir = tempfile::tempdir().unwrap();
7391 let mut w = make_writer(dir.path()).await;
7392 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
7393 .await
7394 .unwrap();
7395 w.write_task_end("tsk_1", "ok", 2, Vec::new(), None, None)
7396 .await
7397 .unwrap();
7398 w.flush().await.unwrap();
7399
7400 let events = read_events(dir.path());
7401 let end = events
7402 .iter()
7403 .find(|e| e["event_type"] == "task_end")
7404 .unwrap();
7405 assert_eq!(end["task_id"], "tsk_1");
7406 assert_eq!(end["exit_status"], "ok");
7407 assert_eq!(end["reopen_count"], 2);
7408 }
7409
7410 #[tokio::test]
7413 async fn task_end_carries_keep_working_only_when_set() {
7414 let dir = tempfile::tempdir().unwrap();
7415 let mut w = make_writer(dir.path()).await;
7416 for (task_id, keep_working) in [
7417 (
7418 "tsk_1",
7419 Some(KeepWorkingSummary {
7420 used: 2,
7421 limit: 2,
7422 stopped_by: Some("limit"),
7423 }),
7424 ),
7425 (
7426 "tsk_2",
7427 Some(KeepWorkingSummary {
7428 used: 1,
7429 limit: 5,
7430 stopped_by: None,
7431 }),
7432 ),
7433 ("tsk_3", None),
7434 ] {
7435 w.write_task_start(task_id, "ctx_1", "a2a", event_provenance(), 3, &[], &[])
7436 .await
7437 .unwrap();
7438 w.write_task_end(task_id, "ok", 0, Vec::new(), keep_working, None)
7439 .await
7440 .unwrap();
7441 }
7442 w.flush().await.unwrap();
7443
7444 let events = read_events(dir.path());
7445 let ends: Vec<_> = events
7446 .iter()
7447 .filter(|e| e["event_type"] == "task_end")
7448 .collect();
7449 assert_eq!(
7450 ends[0]["keep_working"],
7451 serde_json::json!({"used": 2, "limit": 2, "stopped_by": "limit"})
7452 );
7453 assert_eq!(
7454 ends[1]["keep_working"],
7455 serde_json::json!({"used": 1, "limit": 5, "stopped_by": null})
7456 );
7457 assert!(ends[2].get("keep_working").is_none(), "{}", ends[2]);
7458 }
7459
7460 #[tokio::test]
7463 async fn keep_working_record_carries_its_fields_and_feeds_session_status() {
7464 let dir = tempfile::tempdir().unwrap();
7465 let mut w = make_writer(dir.path()).await;
7466 let status = std::sync::Arc::new(crate::session_status::SessionStatus::new(
7467 std::time::Instant::now(),
7468 false,
7469 None,
7470 None,
7471 0,
7472 Some(3),
7473 None,
7474 ));
7475 w.set_session_status(std::sync::Arc::clone(&status));
7476 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
7477 .await
7478 .unwrap();
7479 w.write_keep_working("tsk_1", 1, 3, Some(4), 66)
7480 .await
7481 .unwrap();
7482 w.write_keep_working("tsk_1", 2, 3, None, 66).await.unwrap();
7483 w.flush().await.unwrap();
7484
7485 let events = read_events(dir.path());
7486 let start = events
7487 .iter()
7488 .find(|e| e["event_type"] == "task_start")
7489 .unwrap();
7490 let records: Vec<_> = events
7491 .iter()
7492 .filter(|e| e["event_type"] == "keep_working")
7493 .collect();
7494 assert_eq!(records.len(), 2);
7495 for (record, (number, turns)) in records
7496 .iter()
7497 .zip([(1, serde_json::json!(4)), (2, serde_json::Value::Null)])
7498 {
7499 assert_eq!(record["task_id"], "tsk_1");
7500 assert_eq!(record["number"], number);
7501 assert_eq!(record["limit"], 3);
7502 assert_eq!(record["turns_remaining"], turns);
7503 assert_eq!(record["message_bytes"], 66);
7504 assert_eq!(record["parent_id"], start["event_id"]);
7505 assert!(record["event_id"].is_string());
7506 }
7507 let spend = crate::session_status::SpendReading {
7508 used: 0,
7509 ceiling: None,
7510 machine: None,
7511 };
7512 assert_eq!(
7513 serde_json::to_value(status.snapshot(spend)).unwrap()["keep_working"],
7514 serde_json::json!({"used": 2, "left": 1})
7515 );
7516 }
7517
7518 #[tokio::test]
7520 async fn task_end_carries_done_check_only_when_set() {
7521 let dir = tempfile::tempdir().unwrap();
7522 let mut w = make_writer(dir.path()).await;
7523 for (task_id, done_check) in [
7524 (
7525 "tsk_1",
7526 Some(DoneCheckSummary {
7527 runs: 3,
7528 reopens: 2,
7529 limit: 2,
7530 met: Some(false),
7531 stopped_by: Some("limit"),
7532 }),
7533 ),
7534 (
7535 "tsk_2",
7536 Some(DoneCheckSummary {
7537 runs: 0,
7538 reopens: 0,
7539 limit: 1,
7540 met: None,
7541 stopped_by: None,
7542 }),
7543 ),
7544 ("tsk_3", None),
7545 ] {
7546 w.write_task_start(task_id, "ctx_1", "a2a", event_provenance(), 3, &[], &[])
7547 .await
7548 .unwrap();
7549 w.write_task_end(task_id, "ok", 0, Vec::new(), None, done_check)
7550 .await
7551 .unwrap();
7552 }
7553 w.flush().await.unwrap();
7554
7555 let events = read_events(dir.path());
7556 let ends: Vec<_> = events
7557 .iter()
7558 .filter(|e| e["event_type"] == "task_end")
7559 .collect();
7560 assert_eq!(
7561 ends[0]["done_check"],
7562 serde_json::json!({"runs": 3, "reopens": 2, "limit": 2, "met": false,
7563 "stopped_by": "limit"})
7564 );
7565 assert_eq!(
7566 ends[1]["done_check"],
7567 serde_json::json!({"runs": 0, "reopens": 0, "limit": 1, "met": null,
7568 "stopped_by": null})
7569 );
7570 assert!(ends[2].get("done_check").is_none(), "{}", ends[2]);
7571 }
7572
7573 #[tokio::test]
7576 async fn done_check_record_carries_its_fields_and_feeds_session_status() {
7577 let dir = tempfile::tempdir().unwrap();
7578 let mut w = make_writer(dir.path()).await;
7579 let status = std::sync::Arc::new(crate::session_status::SessionStatus::new(
7580 std::time::Instant::now(),
7581 false,
7582 None,
7583 None,
7584 0,
7585 None,
7586 Some(3),
7587 ));
7588 w.set_session_status(std::sync::Arc::clone(&status));
7589 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
7590 .await
7591 .unwrap();
7592 let mut run = DoneCheckRun {
7593 task_id: "tsk_1",
7594 number: 1,
7595 command: "bash -c 'exit 1'",
7596 outcome: "failed",
7597 exit_code: Some(1),
7598 duration_ms: 12,
7599 stdout_bytes: 5,
7600 stderr_bytes: 0,
7601 truncated: false,
7602 resource_limit: None,
7603 reopened: true,
7604 reopens_left: 2,
7605 turns_remaining: Some(4),
7606 };
7607 w.write_done_check(&run, 1).await.unwrap();
7608 run.number = 2;
7609 run.outcome = "timed_out";
7610 run.exit_code = None;
7611 run.reopened = false;
7612 run.turns_remaining = None;
7613 w.write_done_check(&run, 1).await.unwrap();
7614 w.flush().await.unwrap();
7615
7616 let text = std::fs::read_to_string(dir.path().join("trace.jsonl")).unwrap();
7617 let line = text
7618 .lines()
7619 .find(|line| line.contains("\"done_check\""))
7620 .unwrap();
7621 let order: Vec<usize> = [
7622 "event_type",
7623 "event_id",
7624 "parent_id",
7625 "session_id",
7626 "timestamp",
7627 "task_id",
7628 "number",
7629 "command",
7630 "outcome",
7631 "exit_code",
7632 "duration_ms",
7633 "stdout_bytes",
7634 "stderr_bytes",
7635 "truncated",
7636 "resource_limit",
7637 "reopened",
7638 "reopens_left",
7639 "turns_remaining",
7640 ]
7641 .iter()
7642 .map(|key| line.find(&format!("\"{key}\":")).unwrap())
7643 .collect();
7644 assert!(order.windows(2).all(|pair| pair[0] < pair[1]), "{line}");
7645
7646 let events = read_events(dir.path());
7647 let start = events
7648 .iter()
7649 .find(|e| e["event_type"] == "task_start")
7650 .unwrap();
7651 let records: Vec<_> = events
7652 .iter()
7653 .filter(|e| e["event_type"] == "done_check")
7654 .collect();
7655 assert_eq!(records.len(), 2);
7656 assert_eq!(records[0]["parent_id"], start["event_id"]);
7657 assert_eq!(records[0]["command"], "bash -c 'exit 1'");
7658 assert_eq!(records[0]["exit_code"], 1);
7659 assert_eq!(records[0]["reopened"], true);
7660 assert_eq!(records[0]["turns_remaining"], 4);
7661 assert_eq!(records[1]["outcome"], "timed_out");
7662 assert_eq!(records[1]["exit_code"], serde_json::Value::Null);
7663 assert_eq!(records[1]["resource_limit"], serde_json::Value::Null);
7664 assert_eq!(records[1]["turns_remaining"], serde_json::Value::Null);
7665 let spend = crate::session_status::SpendReading {
7666 used: 0,
7667 ceiling: None,
7668 machine: None,
7669 };
7670 assert_eq!(
7671 serde_json::to_value(status.snapshot(spend)).unwrap()["done_check"],
7672 serde_json::json!({"used": 1, "left": 2})
7673 );
7674 }
7675
7676 #[tokio::test]
7679 async fn task_end_names_the_commands_still_running() {
7680 let dir = tempfile::tempdir().unwrap();
7681 let mut w = make_writer(dir.path()).await;
7682 for (task_id, running) in [
7683 ("tsk_1", vec!["wrk_0a".to_string(), "wrk_0b".to_string()]),
7684 ("tsk_2", Vec::new()),
7685 ] {
7686 w.write_task_start(task_id, "ctx_1", "a2a", event_provenance(), 3, &[], &[])
7687 .await
7688 .unwrap();
7689 w.write_task_end(task_id, "ok", 0, running, None, None)
7690 .await
7691 .unwrap();
7692 }
7693 w.flush().await.unwrap();
7694
7695 let events = read_events(dir.path());
7696 let ends: Vec<_> = events
7697 .iter()
7698 .filter(|e| e["event_type"] == "task_end")
7699 .collect();
7700 assert_eq!(
7701 ends[0]["running_commands"],
7702 serde_json::json!(["wrk_0a", "wrk_0b"])
7703 );
7704 assert!(ends[1].get("running_commands").is_none(), "{}", ends[1]);
7705 }
7706
7707 #[tokio::test]
7710 async fn shell_completed_without_a_completion_task_omits_the_field() {
7711 let dir = tempfile::tempdir().unwrap();
7712 let mut w = make_writer(dir.path()).await;
7713 for task in [None, Some("tsk_2")] {
7714 w.write_shell_completed(
7715 "wrk_0a",
7716 "/usr/bin/bash",
7717 "sleep 3",
7718 0,
7719 3_000,
7720 "/records/logs/wrk_0a.log",
7721 10,
7722 None,
7723 "ok",
7724 task,
7725 )
7726 .await
7727 .unwrap();
7728 }
7729 w.flush().await.unwrap();
7730
7731 let events = read_events(dir.path());
7732 let completed: Vec<_> = events
7733 .iter()
7734 .filter(|e| e["event_type"] == "shell_completed")
7735 .collect();
7736 assert!(
7737 completed[0].get("completion_task_id").is_none(),
7738 "{}",
7739 completed[0]
7740 );
7741 assert_eq!(completed[1]["completion_task_id"], "tsk_2");
7742 }
7743
7744 #[tokio::test]
7745 async fn task_reopened_names_hook_reason_and_ordinal() {
7746 let dir = tempfile::tempdir().unwrap();
7747 let mut w = make_writer(dir.path()).await;
7748 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
7749 .await
7750 .unwrap();
7751 w.write_task_reopened(
7752 "tsk_1",
7753 "gatekeeper",
7754 "tests still fail",
7755 1,
7756 REOPEN_CONTEXT_CONTINUED,
7757 Some(7),
7758 )
7759 .await
7760 .unwrap();
7761 w.flush().await.unwrap();
7762
7763 let events = read_events(dir.path());
7764 let re = events
7765 .iter()
7766 .find(|e| e["event_type"] == "task_reopened")
7767 .unwrap();
7768 assert_eq!(re["task_id"], "tsk_1");
7769 assert_eq!(re["hook_name"], "gatekeeper");
7770 assert_eq!(re["reason"], "tests still fail");
7771 assert_eq!(re["reopen_number"], 1);
7772 assert_eq!(re["attempt_context"], "continued");
7773 assert_eq!(re["turns_remaining"], 7);
7774
7775 w.write_task_reopened(
7776 "tsk_1",
7777 "gatekeeper",
7778 "again",
7779 2,
7780 REOPEN_CONTEXT_CONTINUED,
7781 None,
7782 )
7783 .await
7784 .unwrap();
7785 w.flush().await.unwrap();
7786 let uncapped = read_events(dir.path())
7787 .into_iter()
7788 .filter(|e| e["event_type"] == "task_reopened")
7789 .nth(1)
7790 .unwrap();
7791 assert!(uncapped
7792 .get("turns_remaining")
7793 .is_some_and(serde_json::Value::is_null));
7794 assert!(re["timestamp"].as_u64().unwrap() > 0);
7795 }
7796
7797 #[tokio::test]
7798 async fn task_continued_names_the_round_under_the_task_node() {
7799 let dir = tempfile::tempdir().unwrap();
7800 let mut w = make_writer(dir.path()).await;
7801 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
7802 .await
7803 .unwrap();
7804 w.write_task_continued(
7805 "tsk_1",
7806 1,
7807 &["mcl_b".to_string(), "mcl_a".to_string()],
7808 &["dlg_1".to_string()],
7809 1234,
7810 Some(8),
7811 )
7812 .await
7813 .unwrap();
7814 w.flush().await.unwrap();
7815
7816 let events = read_events(dir.path());
7817 let start = events
7818 .iter()
7819 .find(|e| e["event_type"] == "task_start")
7820 .unwrap();
7821 let continued = events
7822 .iter()
7823 .find(|e| e["event_type"] == "task_continued")
7824 .unwrap();
7825 assert_eq!(continued["parent_id"], start["event_id"]);
7826 assert_eq!(continued["task_id"], "tsk_1");
7827 assert_eq!(continued["continuation_number"], 1);
7828 assert_eq!(
7829 continued["member_calls"],
7830 serde_json::json!(["mcl_b", "mcl_a"])
7831 );
7832 assert_eq!(continued["delegations"], serde_json::json!(["dlg_1"]));
7833 assert_eq!(continued["waited_ms"], 1234);
7834 assert_eq!(continued["turns_remaining"], 8);
7835 }
7836
7837 #[tokio::test]
7841 async fn task_turns_accessor_tracks_and_resets() {
7842 let dir = tempfile::tempdir().unwrap();
7843 let mut w = make_writer(dir.path()).await;
7844 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
7845 .await
7846 .unwrap();
7847 assert_eq!(w.task_turns(), 0);
7848 w.write_inference(
7849 0,
7850 Some(10),
7851 Some(5),
7852 "end_turn".to_string(),
7853 Some("end_turn"),
7854 None,
7855 None,
7856 None,
7857 Vec::new(),
7858 None,
7859 None,
7860 )
7861 .await
7862 .unwrap();
7863 w.write_inference(
7864 1,
7865 Some(10),
7866 Some(5),
7867 "end_turn".to_string(),
7868 Some("end_turn"),
7869 None,
7870 None,
7871 None,
7872 Vec::new(),
7873 None,
7874 None,
7875 )
7876 .await
7877 .unwrap();
7878 assert_eq!(w.task_turns(), 2);
7879 w.write_task_start("tsk_2", "ctx_2", "a2a", event_provenance(), 3, &[], &[])
7881 .await
7882 .unwrap();
7883 assert_eq!(w.task_turns(), 0);
7884 }
7885
7886 #[tokio::test]
7889 async fn task_agent_turns_counts_agent_loop_inference_only() {
7890 let dir = tempfile::tempdir().unwrap();
7891 let mut w = make_writer(dir.path()).await;
7892 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
7893 .await
7894 .unwrap();
7895 assert_eq!(w.task_agent_turns(), 0);
7896 let hook = InferenceOrigin {
7897 source: "hook:x".into(),
7898 model: "m".into(),
7899 };
7900 for (turn, origin) in [(0, None), (0, Some(&hook)), (1, None)] {
7901 w.write_inference(
7902 turn,
7903 Some(10),
7904 Some(5),
7905 "end_turn".to_string(),
7906 Some("end_turn"),
7907 None,
7908 origin,
7909 None,
7910 Vec::new(),
7911 None,
7912 None,
7913 )
7914 .await
7915 .unwrap();
7916 }
7917 assert_eq!(w.task_agent_turns(), 2);
7918 assert_eq!(w.task_turns(), 3);
7919 assert_eq!(w.task_hook_inferences(), 1);
7920 w.write_task_end("tsk_1", "completed", 0, Vec::new(), None, None)
7921 .await
7922 .unwrap();
7923 assert_eq!(
7924 w.task_agent_turns(),
7925 0,
7926 "a run outside a task frame starts at 0"
7927 );
7928 w.write_inference(
7929 0,
7930 Some(10),
7931 Some(5),
7932 "end_turn".to_string(),
7933 Some("end_turn"),
7934 None,
7935 None,
7936 None,
7937 Vec::new(),
7938 None,
7939 None,
7940 )
7941 .await
7942 .unwrap();
7943 w.write_task_start("tsk_2", "ctx_2", "a2a", event_provenance(), 3, &[], &[])
7944 .await
7945 .unwrap();
7946 assert_eq!(w.task_agent_turns(), 0);
7947 }
7948
7949 #[tokio::test]
7955 async fn every_event_carries_a_unique_id_and_parents_to_an_earlier_event() {
7956 let dir = tempfile::tempdir().unwrap();
7957 let mut w = make_writer(dir.path()).await;
7958 w.write_session_start(Some(10), None, vec!["bash".to_string()])
7959 .await
7960 .unwrap();
7961 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
7962 .await
7963 .unwrap();
7964 w.write_inference(
7965 0,
7966 Some(10),
7967 Some(5),
7968 "tool_call".to_string(),
7969 Some("tool_call"),
7970 None,
7971 None,
7972 None,
7973 Vec::new(),
7974 None,
7975 None,
7976 )
7977 .await
7978 .unwrap();
7979 w.write_tool_call(
7980 0,
7981 "bash".to_string(),
7982 Some("toolu_1".to_string()),
7983 serde_json::json!({"command": "echo hi"}),
7984 10,
7985 "hi\n",
7986 3,
7987 5,
7988 "ok".to_string(),
7989 None,
7990 None,
7991 None,
7992 )
7993 .await
7994 .unwrap();
7995 w.write_shell(
7996 0,
7997 "/bin/bash".to_string(),
7998 "echo hi".to_string(),
7999 0,
8000 3,
8001 0,
8002 2,
8003 None,
8004 )
8005 .await
8006 .unwrap();
8007 w.write_artifact_pulled(&pulled(
8008 runtime_summary("pulled-style", ArtifactRuntime::Skill),
8009 "global store",
8010 None,
8011 ))
8012 .await
8013 .unwrap();
8014 w.write_task_end("tsk_1", "ok", 0, Vec::new(), None, None)
8015 .await
8016 .unwrap();
8017 w.write_session_end("ok").await.unwrap();
8018 w.flush().await.unwrap();
8019
8020 let events = read_events(dir.path());
8021 let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
8022 for (i, event) in events.iter().enumerate() {
8023 let id = event["event_id"].as_str().unwrap();
8024 assert!(id.starts_with("evt_") && id.len() == 36, "event {i}: {id}");
8025 assert!(id[4..].chars().all(|c| c.is_ascii_hexdigit()), "event {i}");
8026 assert!(seen.insert(id.to_string()), "event {i} reuses {id}");
8027 assert!(
8028 event.get("parent_id").is_some(),
8029 "event {i} has no parent_id"
8030 );
8031 if let Some(parent) = event["parent_id"].as_str() {
8032 assert!(seen.contains(parent), "event {i} names dangling {parent}");
8033 }
8034 }
8035
8036 let by_type = |ty: &str| {
8037 events
8038 .iter()
8039 .find(|e| e["event_type"] == ty)
8040 .unwrap_or_else(|| panic!("no {ty}"))
8041 .clone()
8042 };
8043 let session = by_type("session_start");
8044 assert!(session["parent_id"].is_null());
8045 assert_eq!(session["event_id"], w.session_event_id());
8046
8047 let task = by_type("task_start");
8048 assert_eq!(task["parent_id"], session["event_id"]);
8049
8050 let turn = by_type("inference");
8051 assert_eq!(turn["parent_id"], task["event_id"]);
8052
8053 assert_eq!(by_type("tool_call")["parent_id"], turn["event_id"]);
8054 assert_eq!(by_type("shell")["parent_id"], turn["event_id"]);
8055 assert_eq!(by_type("task_end")["parent_id"], task["event_id"]);
8056 assert_eq!(by_type("session_end")["parent_id"], session["event_id"]);
8057 assert_eq!(by_type("artifact_pulled")["parent_id"], session["event_id"]);
8058 }
8059
8060 fn runtime_summary(name: &str, runtime: ArtifactRuntime) -> InstalledArtifactSummary {
8061 InstalledArtifactSummary {
8062 name: name.to_string(),
8063 version: "1.0.0".to_string(),
8064 runtime,
8065 implementation: None,
8066 origin: LockOrigin::Runtime {
8067 session: "ses_puller".to_string(),
8068 },
8069 }
8070 }
8071
8072 fn operator_summary(name: &str, runtime: ArtifactRuntime) -> InstalledArtifactSummary {
8073 InstalledArtifactSummary {
8074 origin: LockOrigin::Operator,
8075 ..runtime_summary(name, runtime)
8076 }
8077 }
8078
8079 fn pulled(
8081 summary: InstalledArtifactSummary,
8082 fetched_from: &str,
8083 owner_id: Option<u64>,
8084 ) -> crate::runtime::PulledArtifact {
8085 crate::runtime::PulledArtifact {
8086 summary,
8087 fetched_from: fetched_from.to_string(),
8088 owner_id,
8089 notes: Vec::new(),
8090 }
8091 }
8092
8093 #[tokio::test]
8096 async fn runtime_origin_session_start_lists_nothing_for_an_operator_session() {
8097 for set in [false, true] {
8098 let dir = tempfile::tempdir().unwrap();
8099 let mut w = make_writer(dir.path()).await;
8100 if set {
8101 w.set_runtime_artifacts(&[
8102 operator_summary("house-style", ArtifactRuntime::Skill),
8103 operator_summary("fetcher", ArtifactRuntime::Tool),
8104 ]);
8105 }
8106 w.write_session_start(Some(10), None, Vec::new())
8107 .await
8108 .unwrap();
8109 w.flush().await.unwrap();
8110 let events = read_events(dir.path());
8111 assert_eq!(events[0]["runtime_artifacts"], serde_json::json!([]));
8112 }
8113 }
8114
8115 #[tokio::test]
8118 async fn runtime_origin_session_start_lists_runtime_artifacts_in_order() {
8119 let dir = tempfile::tempdir().unwrap();
8120 let mut w = make_writer(dir.path()).await;
8121 w.set_runtime_artifacts(&[
8122 runtime_summary("zeta-tool", ArtifactRuntime::Tool),
8123 operator_summary("house-style", ArtifactRuntime::Skill),
8124 InstalledArtifactSummary {
8125 origin: LockOrigin::Runtime {
8126 session: "ses_other".to_string(),
8127 },
8128 ..runtime_summary("alpha-style", ArtifactRuntime::Skill)
8129 },
8130 ]);
8131 w.write_session_start(Some(10), None, Vec::new())
8132 .await
8133 .unwrap();
8134 w.flush().await.unwrap();
8135
8136 let events = read_events(dir.path());
8137 assert_eq!(
8138 events[0]["runtime_artifacts"],
8139 serde_json::json!([
8140 {"name": "zeta-tool", "version": "1.0.0", "origin": "runtime",
8141 "session": "ses_puller", "trust": "untrusted"},
8142 {"name": "alpha-style", "version": "1.0.0", "origin": "runtime",
8143 "session": "ses_other", "trust": "untrusted"},
8144 ])
8145 );
8146 }
8147
8148 #[tokio::test]
8149 async fn runtime_origin_skill_call_carries_origin_and_trust() {
8150 let dir = tempfile::tempdir().unwrap();
8151 let mut w = make_writer(dir.path()).await;
8152 for origin in [
8153 LockOrigin::Operator,
8154 LockOrigin::Runtime {
8155 session: "ses_puller".to_string(),
8156 },
8157 ] {
8158 w.write_skill_call(0, "a-skill".to_string(), &origin, 1, 1, "ok".to_string())
8159 .await
8160 .unwrap();
8161 }
8162 w.flush().await.unwrap();
8163
8164 let events = read_events(dir.path());
8165 assert_eq!(events[0]["origin"], "operator");
8166 assert_eq!(events[0]["trust"], "trusted");
8167 assert_eq!(events[1]["origin"], "runtime");
8168 assert_eq!(events[1]["trust"], "untrusted");
8169 }
8170
8171 #[tokio::test]
8174 async fn runtime_origin_artifact_pulled_writes_every_key() {
8175 let dir = tempfile::tempdir().unwrap();
8176 let mut w = make_writer(dir.path()).await;
8177 w.write_artifact_pulled(&pulled(
8178 runtime_summary("pulled-style", ArtifactRuntime::Skill),
8179 "github:ana/artifacts",
8180 Some(4242),
8181 ))
8182 .await
8183 .unwrap();
8184 w.write_artifact_pulled(&pulled(
8185 operator_summary("fetcher", ArtifactRuntime::Tool),
8186 "project store",
8187 None,
8188 ))
8189 .await
8190 .unwrap();
8191 w.flush().await.unwrap();
8192
8193 let events = read_events(dir.path());
8194 let runtime = &events[0];
8195 assert_eq!(runtime["event_type"], "artifact_pulled");
8196 assert_eq!(runtime["name"], "pulled-style");
8197 assert_eq!(runtime["version"], "1.0.0");
8198 assert_eq!(runtime["runtime"], "skill");
8199 assert_eq!(runtime["origin"], "runtime");
8200 assert_eq!(runtime["session"], "ses_puller");
8201 assert_eq!(runtime["trust"], "untrusted");
8202 assert_eq!(runtime["fetched_from"], "github:ana/artifacts");
8203 assert_eq!(runtime["owner_id"], 4242);
8204 assert!(
8205 runtime["parent_id"].is_null(),
8206 "no session_start was written"
8207 );
8208
8209 let operator = &events[1];
8210 assert_eq!(operator["runtime"], "tool");
8211 assert_eq!(operator["origin"], "operator");
8212 assert!(operator.get("session").is_some_and(Value::is_null));
8213 assert_eq!(operator["trust"], "trusted");
8214 assert_eq!(operator["fetched_from"], "project store");
8215 assert!(operator.get("owner_id").is_none());
8216 }
8217
8218 #[tokio::test]
8222 async fn hook_origin_inference_hangs_off_the_turn_rather_than_becoming_one() {
8223 let dir = tempfile::tempdir().unwrap();
8224 let mut w = make_writer(dir.path()).await;
8225 w.write_session_start(Some(10), None, Vec::new())
8226 .await
8227 .unwrap();
8228 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
8229 .await
8230 .unwrap();
8231 w.write_inference(
8232 0,
8233 Some(10),
8234 Some(5),
8235 "tool_call".to_string(),
8236 Some("tool_call"),
8237 None,
8238 None,
8239 None,
8240 Vec::new(),
8241 None,
8242 None,
8243 )
8244 .await
8245 .unwrap();
8246 let origin = InferenceOrigin {
8247 source: "hook:gatekeeper".to_string(),
8248 model: "claude-test".to_string(),
8249 };
8250 w.write_inference(
8251 0,
8252 Some(1),
8253 Some(1),
8254 "end_turn".to_string(),
8255 None,
8257 None,
8258 Some(&origin),
8259 None,
8260 Vec::new(),
8261 None,
8262 None,
8263 )
8264 .await
8265 .unwrap();
8266 w.flush().await.unwrap();
8267
8268 let events = read_events(dir.path());
8269 let turn = &events[2];
8270 let hook = &events[3];
8271 assert!(turn["origin"].is_null());
8272 assert_eq!(hook["origin"], "hook:gatekeeper");
8273 assert_eq!(hook["parent_id"], turn["event_id"]);
8274 assert!(hook.get("message_ids").is_none(), "{hook}");
8277 }
8278
8279 #[tokio::test]
8283 async fn a2a_send_parents_to_null_without_a_session_frame_and_to_it_with_one() {
8284 let bare = tempfile::tempdir().unwrap();
8285 let mut w = make_writer(bare.path()).await;
8286 w.write_a2a_send(
8287 "http://peer",
8288 "msg_1",
8289 "tsk_1",
8290 "ctx_1",
8291 None,
8292 TrustClass::Untrusted,
8293 )
8294 .await
8295 .unwrap();
8296 w.flush().await.unwrap();
8297 assert!(read_events(bare.path())[0]["parent_id"].is_null());
8298
8299 let framed = tempfile::tempdir().unwrap();
8300 let mut w = make_writer(framed.path()).await;
8301 w.write_session_start(Some(1), None, Vec::new())
8302 .await
8303 .unwrap();
8304 w.write_a2a_send(
8305 "http://peer",
8306 "msg_1",
8307 "tsk_1",
8308 "ctx_1",
8309 None,
8310 TrustClass::Untrusted,
8311 )
8312 .await
8313 .unwrap();
8314 w.flush().await.unwrap();
8315 let events = read_events(framed.path());
8316 assert_eq!(events[1]["parent_id"], events[0]["event_id"]);
8317 }
8318
8319 #[tokio::test]
8321 async fn task_rejected_names_the_task_source_cause_and_reason_under_the_session() {
8322 let dir = tempfile::tempdir().unwrap();
8323 let mut w = make_writer(dir.path()).await;
8324 w.write_session_start(Some(10), None, Vec::new())
8325 .await
8326 .unwrap();
8327 w.write_task_start("tsk_ran", "ctx_ran", "a2a", event_provenance(), 3, &[], &[])
8328 .await
8329 .unwrap();
8330 w.write_task_end("tsk_ran", "ok", 0, Vec::new(), None, None)
8331 .await
8332 .unwrap();
8333 w.write_task_rejected(
8334 "tsk_1",
8335 "ctx_1",
8336 "detached_shell",
8337 TASK_REJECTED_SESSION_ENDED,
8338 "task rejected: the session ended before this task started",
8339 )
8340 .await
8341 .unwrap();
8342 w.flush().await.unwrap();
8343
8344 let events = read_events(dir.path());
8345 let session = events
8346 .iter()
8347 .find(|e| e["event_type"] == "session_start")
8348 .unwrap();
8349 let rejected: Vec<_> = events
8350 .iter()
8351 .filter(|e| e["event_type"] == "task_rejected")
8352 .collect();
8353 assert_eq!(rejected.len(), 1);
8354 let rejected = rejected[0];
8355 assert_eq!(rejected["parent_id"], session["event_id"], "{rejected}");
8356 assert_eq!(rejected["session_id"], session["session_id"]);
8357 assert_eq!(rejected["task_id"], "tsk_1");
8358 assert_eq!(rejected["context_id"], "ctx_1");
8359 assert_eq!(rejected["source"], "detached_shell");
8360 assert_eq!(rejected["cause"], "session_ended");
8361 assert_eq!(
8362 rejected["reason"],
8363 "task rejected: the session ended before this task started"
8364 );
8365 assert!(rejected["event_id"].is_string());
8366 assert!(rejected["timestamp"].is_u64());
8367 assert_eq!(rejected.as_object().unwrap().len(), 10, "{rejected}");
8368 }
8369
8370 #[tokio::test]
8373 async fn task_failed_names_the_task_turn_cause_and_capped_reason() {
8374 let dir = tempfile::tempdir().unwrap();
8375 let mut w = make_writer(dir.path()).await;
8376 w.write_session_start(Some(10), None, Vec::new())
8377 .await
8378 .unwrap();
8379 assert_eq!(w.last_task_failure(), None);
8380 w.write_task_failed(None, TASK_FAILED_RUNTIME_ERROR, "before any task")
8381 .await
8382 .unwrap();
8383 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
8384 .await
8385 .unwrap();
8386 let long = "é".repeat(MAX_TASK_FAILURE_REASON_BYTES);
8387 w.write_task_failed(Some(2), TASK_FAILED_DRIVER_ERROR, &long)
8388 .await
8389 .unwrap();
8390 w.write_task_end("tsk_1", "failed", 0, Vec::new(), None, None)
8391 .await
8392 .unwrap();
8393 w.flush().await.unwrap();
8394
8395 assert_eq!(w.task_failures_written(), 2);
8396 let events = read_events(dir.path());
8397 let failed: Vec<_> = events
8398 .iter()
8399 .filter(|e| e["event_type"] == "task_failed")
8400 .collect();
8401 assert_eq!(failed.len(), 2);
8402
8403 assert!(failed[0].get("task_id").is_none(), "{}", failed[0]);
8404 assert!(failed[0].get("turn").is_none(), "{}", failed[0]);
8405 assert_eq!(failed[0]["cause"], "runtime_error");
8406 assert_eq!(failed[0]["reason"], "before any task");
8407
8408 let start = events
8409 .iter()
8410 .find(|e| e["event_type"] == "task_start")
8411 .unwrap();
8412 assert_eq!(failed[1]["task_id"], "tsk_1");
8413 assert_eq!(failed[1]["turn"], 2);
8414 assert_eq!(failed[1]["cause"], "driver_error");
8415 assert_eq!(failed[1]["parent_id"], start["event_id"]);
8416 assert_eq!(failed[1]["session_id"], start["session_id"]);
8417 assert!(failed[1]["event_id"].is_string());
8418 assert!(failed[1]["timestamp"].is_u64());
8419 let reason = failed[1]["reason"].as_str().unwrap();
8420 assert!(reason.len() <= MAX_TASK_FAILURE_REASON_BYTES);
8421 assert!(reason.len() > MAX_TASK_FAILURE_REASON_BYTES - 2);
8422 assert_eq!(w.last_task_failure(), Some(reason));
8423 }
8424
8425 #[tokio::test]
8428 async fn turn_level_events_carry_the_active_task_id() {
8429 let dir = tempfile::tempdir().unwrap();
8430 let mut w = make_writer(dir.path()).await;
8431 w.write_session_start(Some(10), None, Vec::new())
8432 .await
8433 .unwrap();
8434 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
8435 .await
8436 .unwrap();
8437 w.write_inference(
8438 0,
8439 Some(10),
8440 Some(5),
8441 "tool_call".to_string(),
8442 Some("tool_call"),
8443 None,
8444 None,
8445 None,
8446 Vec::new(),
8447 None,
8448 None,
8449 )
8450 .await
8451 .unwrap();
8452 w.write_skill_call(
8453 0,
8454 "house-style".to_string(),
8455 &LockOrigin::Operator,
8456 12,
8457 1,
8458 "ok".to_string(),
8459 )
8460 .await
8461 .unwrap();
8462 w.write_compaction(0, 100, OccupancyBasis::Estimate, 40)
8463 .await
8464 .unwrap();
8465 w.write_task_end("tsk_1", "ok", 0, Vec::new(), None, None)
8466 .await
8467 .unwrap();
8468 w.write_inference(
8469 1,
8470 Some(1),
8471 Some(1),
8472 "end_turn".to_string(),
8473 Some("end_turn"),
8474 None,
8475 None,
8476 None,
8477 Vec::new(),
8478 None,
8479 None,
8480 )
8481 .await
8482 .unwrap();
8483 w.flush().await.unwrap();
8484
8485 let events = read_events(dir.path());
8486 for event in events
8487 .iter()
8488 .filter(|e| {
8489 matches!(
8490 e["event_type"].as_str(),
8491 Some("inference" | "skill_call" | "compaction")
8492 )
8493 })
8494 .take(3)
8495 {
8496 assert_eq!(event["task_id"], "tsk_1", "{event}");
8497 }
8498 let after = events.last().unwrap();
8499 assert_eq!(after["event_type"], "inference");
8500 assert!(
8501 after["task_id"].is_null(),
8502 "an inference outside any task carries a null task_id, not the last task's"
8503 );
8504 assert_eq!(after["parent_id"], events[0]["event_id"]);
8506 }
8507
8508 #[tokio::test]
8511 async fn tool_call_records_the_provider_id_verbatim_and_empty_as_null() {
8512 for (given, expected) in [
8513 (Some("toolu_1".to_string()), Some("toolu_1")),
8514 (Some(String::new()), None),
8515 (None, None),
8516 ] {
8517 let dir = tempfile::tempdir().unwrap();
8518 let mut w = make_writer(dir.path()).await;
8519 w.write_tool_call(
8520 0,
8521 "bash".to_string(),
8522 given.clone(),
8523 serde_json::json!({}),
8524 2,
8525 "",
8526 0,
8527 1,
8528 "ok".to_string(),
8529 None,
8530 None,
8531 None,
8532 )
8533 .await
8534 .unwrap();
8535 w.flush().await.unwrap();
8536 let e = &read_events(dir.path())[0];
8537 match expected {
8538 Some(id) => assert_eq!(e["tool_call_id"], id, "given={given:?}"),
8539 None => assert!(e["tool_call_id"].is_null(), "given={given:?}"),
8540 }
8541 }
8542 }
8543
8544 #[tokio::test]
8547 async fn tool_input_refused_records_its_fields_in_every_capture_mode() {
8548 for capture in [
8549 TraceCapture::None,
8550 TraceCapture::Meta,
8551 TraceCapture::Content,
8552 ] {
8553 for (given, expected) in [
8554 (Some("toolu_1"), Some("toolu_1")),
8555 (Some(""), None),
8556 (None, None),
8557 ] {
8558 let dir = tempfile::tempdir().unwrap();
8559 let mut w = make_writer_with_opts(dir.path(), capture).await;
8560 w.write_tool_input_refused(
8561 2,
8562 "murmur-tool-editor",
8563 given,
8564 &["operation".to_string(), "dest_path".to_string()],
8565 "murmur-tool-editor: missing required fields",
8566 )
8567 .await
8568 .unwrap();
8569 w.flush().await.unwrap();
8570 let e = &read_events(dir.path())[0];
8571 assert_eq!(e["event_type"], "tool_input_refused", "{capture:?}");
8572 assert!(e["event_id"].is_string());
8573 assert!(e["session_id"].is_string());
8574 assert!(e["timestamp"].is_u64());
8575 assert_eq!(e["turn"], 2);
8576 assert_eq!(e["tool_name"], "murmur-tool-editor");
8577 match expected {
8578 Some(id) => assert_eq!(e["tool_call_id"], id, "given={given:?}"),
8579 None => assert!(e["tool_call_id"].is_null(), "given={given:?}"),
8580 }
8581 assert_eq!(e["missing"], serde_json::json!(["operation", "dest_path"]));
8582 assert_eq!(e["reason"], "murmur-tool-editor: missing required fields");
8583 }
8584 }
8585 }
8586
8587 #[tokio::test]
8591 async fn tools_refreshed_event_records_trigger_names_and_task_parent() {
8592 let dir = tempfile::tempdir().unwrap();
8593 let mut w = make_writer(dir.path()).await;
8594 w.set_tool_refresh(Some("immediate"));
8595 w.write_session_start(Some(10), None, Vec::new())
8596 .await
8597 .unwrap();
8598 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
8599 .await
8600 .unwrap();
8601 w.write_inference(
8602 0,
8603 Some(10),
8604 Some(5),
8605 "tool_call".to_string(),
8606 Some("tool_call"),
8607 None,
8608 None,
8609 None,
8610 Vec::new(),
8611 None,
8612 None,
8613 )
8614 .await
8615 .unwrap();
8616 w.write_tools_refreshed(
8617 1,
8618 "immediate",
8619 &["aaa-late-skill".to_string()],
8620 &[],
8621 &[
8622 "aaa-late-skill".to_string(),
8623 "zzz-existing-tool".to_string(),
8624 ],
8625 )
8626 .await
8627 .unwrap();
8628 w.flush().await.unwrap();
8629
8630 let events = read_events(dir.path());
8631 assert_eq!(events[0]["tool_refresh"], "immediate");
8632 let task = events
8633 .iter()
8634 .find(|e| e["event_type"] == "task_start")
8635 .unwrap();
8636 let refreshed: Vec<&Value> = events
8637 .iter()
8638 .filter(|e| e["event_type"] == "tools_refreshed")
8639 .collect();
8640 assert_eq!(refreshed.len(), 1);
8641 let r = refreshed[0];
8642 assert!(r["event_id"].as_str().unwrap().starts_with("evt_"));
8643 assert_eq!(r["parent_id"], task["event_id"]);
8644 assert_eq!(r["session_id"], "test-session-id");
8645 assert!(r["timestamp"].as_u64().is_some());
8646 assert_eq!(r["turn"], 1);
8647 assert_eq!(r["task_id"], "tsk_1");
8648 assert_eq!(r["trigger"], "immediate");
8649 assert_eq!(r["added"], serde_json::json!(["aaa-late-skill"]));
8650 assert_eq!(r["removed"], serde_json::json!([]));
8651 assert_eq!(
8652 r["tools"],
8653 serde_json::json!(["aaa-late-skill", "zzz-existing-tool"])
8654 );
8655
8656 let process_dir = tempfile::tempdir().unwrap();
8657 let mut process = make_writer(process_dir.path()).await;
8658 process.set_tool_refresh(None);
8659 process
8660 .write_session_start(Some(10), None, Vec::new())
8661 .await
8662 .unwrap();
8663 process.flush().await.unwrap();
8664 let start = &read_events(process_dir.path())[0];
8665 assert!(start.as_object().unwrap().contains_key("tool_refresh"));
8666 assert!(start["tool_refresh"].is_null());
8667 }
8668
8669 #[tokio::test]
8672 async fn compaction_declined_records_turn_task_tokens_and_reason() {
8673 let dir = tempfile::tempdir().unwrap();
8674 let mut w = make_writer(dir.path()).await;
8675 w.write_session_start(Some(10), None, Vec::new())
8676 .await
8677 .unwrap();
8678 w.write_task_start("tsk_1", "ctx_1", "a2a", event_provenance(), 3, &[], &[])
8679 .await
8680 .unwrap();
8681 w.write_inference(
8682 2,
8683 Some(10),
8684 Some(5),
8685 "tool_call".to_string(),
8686 Some("tool_call"),
8687 None,
8688 None,
8689 None,
8690 Vec::new(),
8691 None,
8692 None,
8693 )
8694 .await
8695 .unwrap();
8696 w.write_compaction_declined(
8697 2,
8698 4321,
8699 OccupancyBasis::Actual,
8700 COMPACTION_DECLINED_UNRESOLVED_TOOL_CALL,
8701 )
8702 .await
8703 .unwrap();
8704 w.flush().await.unwrap();
8705
8706 let events = read_events(dir.path());
8707 let declined: Vec<&Value> = events
8708 .iter()
8709 .filter(|e| e["event_type"] == "compaction_declined")
8710 .collect();
8711 assert_eq!(declined.len(), 1);
8712 let d = declined[0];
8713 assert_eq!(d["reason"], "unresolved_tool_call");
8714 assert_eq!(d["turn"], 2);
8715 assert_eq!(d["task_id"], "tsk_1");
8716 assert_eq!(d["tokens"], 4321);
8717 assert_eq!(d["tokens_source"], "actual");
8718 assert!(d["event_id"].as_str().unwrap().starts_with("evt_"));
8719 assert_eq!(
8720 d["parent_id"],
8721 events
8722 .iter()
8723 .find(|e| e["event_type"] == "inference")
8724 .unwrap()["event_id"]
8725 );
8726 }
8727
8728 #[tokio::test]
8731 async fn event_ids_sort_by_mint_time() {
8732 let dir = tempfile::tempdir().unwrap();
8733 let mut w = make_writer(dir.path()).await;
8734 for turn in 0..5 {
8735 w.write_inference(
8736 turn,
8737 Some(1),
8738 Some(1),
8739 "end_turn".to_string(),
8740 Some("end_turn"),
8741 None,
8742 None,
8743 None,
8744 Vec::new(),
8745 None,
8746 None,
8747 )
8748 .await
8749 .unwrap();
8750 }
8751 w.flush().await.unwrap();
8752
8753 let ids: Vec<String> = read_events(dir.path())
8754 .iter()
8755 .map(|e| e["event_id"].as_str().unwrap().to_string())
8756 .collect();
8757 let mut sorted = ids.clone();
8758 sorted.sort();
8759 assert_eq!(ids, sorted, "evt_ ids must sort into mint order");
8760 }
8761
8762 fn wire_payload(system: &str, messages: usize) -> Value {
8766 let messages: Vec<Value> = (0..messages)
8767 .map(|i| serde_json::json!({"role": "user", "content": format!("m{i}")}))
8768 .collect();
8769 serde_json::json!({
8770 "model": "claude-test",
8771 "max_tokens": 1024,
8772 "messages": messages,
8773 "tools": [{"name": "bash"}],
8774 "params": {},
8775 "system": system,
8776 })
8777 }
8778
8779 async fn write_turn(w: &mut TraceWriter, turn: u32, payload: &Value, response: &str) {
8780 let wire = WireCapture::from_driver_payload(payload, response);
8781 w.write_inference(
8782 turn,
8783 Some(10),
8784 Some(5),
8785 "end_turn".to_string(),
8786 Some("end_turn"),
8787 None,
8788 None,
8789 None,
8790 vec!["msg_0000000000000000000000000000000a".to_string()],
8791 Some(&wire),
8792 None,
8793 )
8794 .await
8795 .unwrap();
8796 }
8797
8798 fn blob_path(dir: &std::path::Path, sha: &str) -> std::path::PathBuf {
8799 dir.join(BLOB_DIR_NAME).join(sha)
8800 }
8801
8802 #[tokio::test]
8805 async fn content_capture_names_blobs_that_exist_and_rehash_to_their_names() {
8806 let dir = tempfile::tempdir().unwrap();
8807 let mut w = make_writer_with_opts(dir.path(), TraceCapture::Content).await;
8808 let payload = wire_payload("You are a capsule.", 3);
8809 write_turn(&mut w, 0, &payload, r#"{"stop_reason":"end_turn"}"#).await;
8810 w.flush().await.unwrap();
8811
8812 let e = &read_events(dir.path())[0];
8813 let mut named: Vec<String> = ["system_sha", "tools_sha", "response_sha"]
8814 .iter()
8815 .map(|key| {
8816 e[*key]
8817 .as_str()
8818 .unwrap_or_else(|| panic!("{key} missing from {e}"))
8819 .to_string()
8820 })
8821 .collect();
8822 let shas = e["message_shas"].as_array().expect("message_shas");
8823 assert_eq!(shas.len(), 3);
8824 named.extend(shas.iter().map(|s| s.as_str().unwrap().to_string()));
8825
8826 for sha in &named {
8827 let path = blob_path(dir.path(), sha);
8828 let bytes =
8829 std::fs::read(&path).unwrap_or_else(|err| panic!("blob {sha} must exist: {err}"));
8830 assert_eq!(murmur_artifact::sha256_hex(&bytes), *sha);
8831 assert_eq!(sha.len(), 64, "{sha}");
8834 assert!(
8835 sha.chars()
8836 .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()),
8837 "{sha}"
8838 );
8839 }
8840 }
8841
8842 #[tokio::test]
8845 async fn the_system_blob_holds_the_prompt_as_readable_text() {
8846 let dir = tempfile::tempdir().unwrap();
8847 let mut w = make_writer_with_opts(dir.path(), TraceCapture::Content).await;
8848 let payload = wire_payload("[Capsule] you are cap.\nBe terse.", 1);
8849 write_turn(&mut w, 0, &payload, "{}").await;
8850 w.flush().await.unwrap();
8851
8852 let sha = read_events(dir.path())[0]["system_sha"]
8853 .as_str()
8854 .unwrap()
8855 .to_string();
8856 assert_eq!(
8857 std::fs::read_to_string(blob_path(dir.path(), &sha)).unwrap(),
8858 "[Capsule] you are cap.\nBe terse."
8859 );
8860 }
8861
8862 #[tokio::test]
8865 async fn meta_writes_the_same_hashes_and_no_bodies() {
8866 let payload = wire_payload("You are a capsule.", 2);
8867 let response = r#"{"stop_reason":"end_turn"}"#;
8868
8869 let meta = tempfile::tempdir().unwrap();
8870 let mut w = make_writer_with_opts(meta.path(), TraceCapture::Meta).await;
8871 write_turn(&mut w, 0, &payload, response).await;
8872 w.flush().await.unwrap();
8873
8874 let content = tempfile::tempdir().unwrap();
8875 let mut w = make_writer_with_opts(content.path(), TraceCapture::Content).await;
8876 write_turn(&mut w, 0, &payload, response).await;
8877 w.flush().await.unwrap();
8878
8879 let meta_event = read_events(meta.path()).remove(0);
8880 let content_event = read_events(content.path()).remove(0);
8881 for key in ["system_sha", "tools_sha", "response_sha", "message_shas"] {
8882 assert_eq!(meta_event[key], content_event[key], "{key}");
8883 }
8884 assert!(
8885 !meta.path().join(BLOB_DIR_NAME).exists(),
8886 "meta must not create the blob directory"
8887 );
8888 assert!(content.path().join(BLOB_DIR_NAME).is_dir());
8889 }
8890
8891 #[tokio::test]
8893 async fn the_default_capture_mode_is_meta() {
8894 assert_eq!(TraceCapture::default(), TraceCapture::Meta);
8895 let dir = tempfile::tempdir().unwrap();
8896 let mut w = make_writer_with_prompt(dir.path(), TraceCapture::default(), None, false).await;
8897 write_turn(&mut w, 0, &wire_payload("s", 1), "{}").await;
8898 w.flush().await.unwrap();
8899
8900 let e = &read_events(dir.path())[0];
8901 assert!(e.get("system_sha").is_some());
8902 assert!(!dir.path().join(BLOB_DIR_NAME).exists());
8903 }
8904
8905 #[tokio::test]
8907 async fn none_omits_every_hash_and_leaves_the_rest_of_the_event_intact() {
8908 let payload = wire_payload("You are a capsule.", 2);
8909
8910 let off = tempfile::tempdir().unwrap();
8911 let mut w = make_writer_with_opts(off.path(), TraceCapture::None).await;
8912 write_turn(&mut w, 3, &payload, "{}").await;
8913 w.flush().await.unwrap();
8914
8915 let e = read_events(off.path()).remove(0);
8916 for key in ["system_sha", "tools_sha", "response_sha", "message_shas"] {
8917 assert!(
8918 e.get(key).is_none(),
8919 "{key} must be absent under none, got {e}"
8920 );
8921 }
8922 assert!(!off.path().join(BLOB_DIR_NAME).exists());
8923 for key in [
8924 "event_type",
8925 "event_id",
8926 "parent_id",
8927 "session_id",
8928 "timestamp",
8929 "turn",
8930 "task_id",
8931 "input_tokens",
8932 "output_tokens",
8933 "decision",
8934 "tool_name",
8935 "message_ids",
8936 ] {
8937 assert!(
8938 e.get(key).is_some(),
8939 "{key} must survive under none, got {e}"
8940 );
8941 }
8942 assert_eq!(e["turn"], 3);
8943 assert_eq!(e["message_ids"].as_array().unwrap().len(), 1);
8944 }
8945
8946 #[tokio::test]
8949 async fn an_unchanged_system_prompt_is_stored_once() {
8950 let dir = tempfile::tempdir().unwrap();
8951 let mut w = make_writer_with_opts(dir.path(), TraceCapture::Content).await;
8952 let payload = wire_payload("You are a capsule.", 1);
8953 write_turn(&mut w, 0, &payload, "{}").await;
8954
8955 let sha = {
8956 w.flush().await.unwrap();
8957 read_events(dir.path())[0]["system_sha"]
8958 .as_str()
8959 .unwrap()
8960 .to_string()
8961 };
8962 let path = blob_path(dir.path(), &sha);
8963 std::fs::write(&path, b"sentinel").unwrap();
8964
8965 for turn in 1..10 {
8966 write_turn(&mut w, turn, &payload, "{}").await;
8967 }
8968 w.flush().await.unwrap();
8969
8970 let events = read_events(dir.path());
8971 assert_eq!(events.len(), 10);
8972 for e in &events {
8973 assert_eq!(e["system_sha"], sha.as_str());
8974 }
8975 assert_eq!(
8976 std::fs::read(&path).unwrap(),
8977 b"sentinel",
8978 "an existing blob is never rewritten"
8979 );
8980 let matching = std::fs::read_dir(dir.path().join(BLOB_DIR_NAME))
8981 .unwrap()
8982 .filter(|entry| entry.as_ref().unwrap().file_name() == sha.as_str())
8983 .count();
8984 assert_eq!(matching, 1, "exactly one file bears that name");
8985 }
8986
8987 #[tokio::test]
8992 async fn stop_reason_is_absent_when_no_driver_response_was_parsed() {
8993 let dir = tempfile::tempdir().unwrap();
8994 let mut w = make_writer(dir.path()).await;
8995 w.write_inference(
8996 0,
8997 Some(10),
8998 Some(5),
8999 "end_turn".to_string(),
9000 None,
9001 None,
9002 None,
9003 None,
9004 Vec::new(),
9005 None,
9006 None,
9007 )
9008 .await
9009 .unwrap();
9010 w.flush().await.unwrap();
9011
9012 let event = read_events(dir.path()).remove(0);
9013 assert!(
9014 event.get("stop_reason").is_none(),
9015 "stop_reason must be absent, got {event}"
9016 );
9017 }
9018
9019 #[tokio::test]
9022 async fn a_hook_origin_record_carries_no_hashes() {
9023 let dir = tempfile::tempdir().unwrap();
9024 let mut w = make_writer_with_opts(dir.path(), TraceCapture::Content).await;
9025 let origin = InferenceOrigin {
9026 source: "hook:gatekeeper".to_string(),
9027 model: "claude-test".to_string(),
9028 };
9029 w.write_inference(
9030 0,
9031 Some(1),
9032 Some(1),
9033 "end_turn".to_string(),
9034 None,
9036 None,
9037 Some(&origin),
9038 None,
9039 Vec::new(),
9040 None,
9041 None,
9042 )
9043 .await
9044 .unwrap();
9045 w.flush().await.unwrap();
9046
9047 let e = read_events(dir.path()).remove(0);
9048 assert_eq!(e["origin"], "hook:gatekeeper");
9049 for key in ["system_sha", "tools_sha", "response_sha", "message_shas"] {
9050 assert!(e.get(key).is_none(), "{key} must be absent, got {e}");
9051 }
9052 assert!(!dir.path().join(BLOB_DIR_NAME).exists());
9053 }
9054
9055 #[tokio::test]
9058 async fn message_shas_locate_the_index_two_runs_diverge_at() {
9059 async fn shas_for(payload: &Value) -> Vec<String> {
9060 let dir = tempfile::tempdir().unwrap();
9061 let mut w = make_writer_with_opts(dir.path(), TraceCapture::Meta).await;
9062 write_turn(&mut w, 0, payload, "{}").await;
9063 w.flush().await.unwrap();
9064 read_events(dir.path())[0]["message_shas"]
9065 .as_array()
9066 .unwrap()
9067 .iter()
9068 .map(|s| s.as_str().unwrap().to_string())
9069 .collect()
9070 }
9071
9072 let first = wire_payload("You are a capsule.", 4);
9073 let mut second = first.clone();
9074 second["messages"][2]["content"] = serde_json::json!("diverged");
9075
9076 let a = shas_for(&first).await;
9077 let b = shas_for(&second).await;
9078 assert_eq!(a.len(), 4);
9079 let divergence = a.iter().zip(&b).position(|(x, y)| x != y);
9080 assert_eq!(divergence, Some(2));
9081 assert_eq!(a[3], b[3], "only the changed message changes its own sha");
9082 }
9083
9084 #[tokio::test]
9087 async fn identical_messages_share_one_sha_and_one_blob() {
9088 let dir = tempfile::tempdir().unwrap();
9089 let mut w = make_writer_with_opts(dir.path(), TraceCapture::Content).await;
9090 let payload = serde_json::json!({
9091 "messages": [{"role": "user", "content": "same"}, {"role": "user", "content": "same"}],
9092 "tools": [],
9093 "system": "s",
9094 });
9095 write_turn(&mut w, 0, &payload, "{}").await;
9096 w.flush().await.unwrap();
9097
9098 let shas = read_events(dir.path())[0]["message_shas"]
9099 .as_array()
9100 .unwrap()
9101 .iter()
9102 .map(|s| s.as_str().unwrap().to_string())
9103 .collect::<Vec<_>>();
9104 assert_eq!(shas[0], shas[1]);
9105 assert!(blob_path(dir.path(), &shas[0]).is_file());
9106 }
9107
9108 #[tokio::test]
9111 async fn a_message_blob_is_the_body_verbatim() {
9112 let dir = tempfile::tempdir().unwrap();
9113 let mut w = make_writer_with_opts(dir.path(), TraceCapture::Content).await;
9114 let message = serde_json::json!({"role": "user", "content": "hello"});
9115 let payload = serde_json::json!({
9116 "messages": [message.clone()],
9117 "tools": [],
9118 "system": "s",
9119 });
9120 write_turn(&mut w, 0, &payload, "{}").await;
9121 w.flush().await.unwrap();
9122
9123 let sha = read_events(dir.path())[0]["message_shas"][0]
9124 .as_str()
9125 .unwrap()
9126 .to_string();
9127 assert_eq!(
9128 std::fs::read(blob_path(dir.path(), &sha)).unwrap(),
9129 serde_json::to_vec(&message).unwrap()
9130 );
9131 }
9132
9133 #[tokio::test]
9141 async fn a_plan_written_delegation_record_matches_a_delegate_task_one() {
9142 let plan_dir = tempfile::tempdir().unwrap();
9143 let tool_dir = tempfile::tempdir().unwrap();
9144 let session = "ses_0000000000004000800000000000plan".to_string();
9145 let node = "evt_0000000000004000800000000000node".to_string();
9146
9147 let plan = PlanTraceAppender::open(
9148 &crate::workdir_fs::WorkdirRoot::open(plan_dir.path()).unwrap(),
9149 session.clone(),
9150 node.clone(),
9151 )
9152 .unwrap();
9153 let tool = ResourceTraceAppender::open(
9154 &crate::workdir_fs::WorkdirRoot::open(tool_dir.path()).unwrap(),
9155 session.clone(),
9156 node.clone(),
9157 )
9158 .await
9159 .unwrap();
9160
9161 plan.write_delegation_start(
9162 "dlg_abc123",
9163 "worker",
9164 "0.1.0",
9165 "ses_child",
9166 ".murmur/c",
9167 "/r/ses_child",
9168 );
9169 tool.write_delegation_start(
9170 "dlg_abc123",
9171 "worker",
9172 "0.1.0",
9173 "ses_child",
9174 ".murmur/c",
9175 "/r/ses_child",
9176 )
9177 .await;
9178 plan.write_delegation(
9179 "worker",
9180 "0.1.0",
9181 Some("dlg_abc123".to_string()),
9182 Some("ses_child".to_string()),
9183 42,
9184 "completed",
9185 None,
9186 );
9187 tool.write_delegation(
9188 "worker",
9189 "0.1.0",
9190 Some("dlg_abc123".to_string()),
9191 Some("ses_child".to_string()),
9192 42,
9193 "completed",
9194 None,
9195 )
9196 .await;
9197
9198 let read = |dir: &Path| -> Vec<serde_json::Map<String, Value>> {
9199 std::fs::read_to_string(dir.join("trace.jsonl"))
9200 .unwrap()
9201 .lines()
9202 .map(|line| serde_json::from_str::<serde_json::Map<_, _>>(line).unwrap())
9203 .collect()
9204 };
9205 let from_plan = read(plan_dir.path());
9206 let from_tool = read(tool_dir.path());
9207 assert_eq!(from_plan.len(), 2);
9208 assert_eq!(from_tool.len(), 2);
9209
9210 for (planned, called, event_type) in [
9211 (&from_plan[0], &from_tool[0], "delegation_start"),
9212 (&from_plan[1], &from_tool[1], "delegation"),
9213 ] {
9214 assert_eq!(planned["event_type"], event_type);
9215 assert_eq!(called["event_type"], event_type);
9216 assert_eq!(
9217 planned.keys().collect::<Vec<_>>(),
9218 called.keys().collect::<Vec<_>>(),
9219 "{planned:?} vs {called:?}"
9220 );
9221 assert_eq!(planned["parent_id"], node.as_str());
9222 assert_eq!(called["parent_id"], node.as_str());
9223 for (key, value) in planned {
9224 if key == "event_id" || key == "timestamp" {
9225 continue;
9226 }
9227 assert_eq!(value, &called[key], "{key}: {planned:?} vs {called:?}");
9228 }
9229 for key in ["plan_id", "step_id", "surface", "kind"] {
9230 assert!(!planned.contains_key(key), "{key} is on {planned:?}");
9231 assert!(!called.contains_key(key), "{key} is on {called:?}");
9232 }
9233 }
9234 }
9235
9236 #[tokio::test]
9240 async fn spawner_ended_names_the_lineage_or_omits_it() {
9241 let dir = tempfile::tempdir().unwrap();
9242 let session = "ses_0000000000004000800000000child".to_string();
9243 let node = "evt_0000000000004000800000000000node".to_string();
9244 let appender = ResourceTraceAppender::open(
9245 &crate::workdir_fs::WorkdirRoot::open(dir.path()).unwrap(),
9246 session.clone(),
9247 node.clone(),
9248 )
9249 .await
9250 .unwrap();
9251
9252 appender
9253 .write_spawner_ended(Some("ses_parent"), Some("dlg_abc123"))
9254 .await;
9255 appender.write_spawner_ended(None, None).await;
9256
9257 let lines: Vec<serde_json::Map<String, Value>> =
9258 std::fs::read_to_string(dir.path().join("trace.jsonl"))
9259 .unwrap()
9260 .lines()
9261 .map(|line| serde_json::from_str(line).unwrap())
9262 .collect();
9263 assert_eq!(lines.len(), 2);
9264 let named = &lines[0];
9265 assert_eq!(named["event_type"], "spawner_ended");
9266 assert_eq!(named["parent_id"], node.as_str());
9267 assert_eq!(named["session_id"], session.as_str());
9268 assert_eq!(named["spawned_by"], "ses_parent");
9269 assert_eq!(named["delegation_id"], "dlg_abc123");
9270 assert_eq!(
9271 named.keys().collect::<Vec<_>>(),
9272 [
9273 "delegation_id",
9274 "event_id",
9275 "event_type",
9276 "parent_id",
9277 "session_id",
9278 "spawned_by",
9279 "timestamp"
9280 ]
9281 );
9282 let bare = &lines[1];
9283 assert_eq!(bare["event_type"], "spawner_ended");
9284 assert!(!bare.contains_key("spawned_by"), "{bare:?}");
9285 assert!(!bare.contains_key("delegation_id"), "{bare:?}");
9286 }
9287}