1use crate::session::{Session, SessionRegistry, Workflow};
4use av_bridge::EventBus;
5use av_core::metrics::Registry;
6use av_core::time::elapsed_us;
7use av_events::StopReason;
8use av_receipts::{Receipt, ReceiptSubject, Signer};
9use serde::Serialize;
10use std::path::PathBuf;
11use std::sync::Arc;
12use std::time::Instant;
13
14const MAX_ATIF_RECOVERY_BYTES: u64 = 256 * 1024 * 1024;
23
24pub(crate) const LIFECYCLE_OUTBOX_PENDING_HELP: &str =
31 "Count of unacked lifecycle outbox files (RECEIPT + SESSION_CLOSE) \
32 sitting in the spool at the last reconciler tick. Steady-state 0 on \
33 a healthy node (each outbox is emitted, published, acked, and \
34 removed within one close). A rising baseline indicates the bridge \
35 is unreachable or slow — outboxes persist across restarts, so \
36 growth is disk-bounded rather than memory-bounded, but a sustained \
37 rise is the earliest signal of a broker outage before its \
38 downstream effects (disk fill, `av_reconcile_errors_total`) fire.";
39
40pub(crate) const MAX_RECOVERY_ENTRIES_PER_TICK: usize = 10_000;
69
70pub(crate) const MAX_RECOVERY_DIRENTS_PER_TICK: usize = 100_000;
99
100pub(crate) fn bump_recovery_scan_cap(
110 metrics: &Registry,
111 pass_label: &'static str,
112 examined: usize,
113 dirents_seen: usize,
114) {
115 metrics
116 .counter(
117 &format!("av_recovery_scan_capped_total{{pass=\"{pass_label}\"}}"),
118 "Recovery pass hit the per-tick entry-scan cap and returned early; \
119 the next tick will re-open the directory and resume. A steady rate \
120 > 0 indicates a poisoned spool (millions of stale files) that will \
121 starve legitimate recovery work until an operator quarantines it.",
122 )
123 .inc();
124 tracing::warn!(
125 pass = %pass_label,
126 examined,
127 dirents_seen,
128 work_cap = MAX_RECOVERY_ENTRIES_PER_TICK,
129 dirent_cap = MAX_RECOVERY_DIRENTS_PER_TICK,
130 "recovery pass hit per-tick scan cap; resuming next tick"
131 );
132}
133
134#[derive(Debug, Clone, Serialize)]
136#[serde(tag = "kind", rename_all = "snake_case")]
137pub enum FinalizeOutcome {
138 Receipt {
140 receipt: Box<Receipt>,
142 },
143 Atif {
145 path: PathBuf,
147 },
148 AlreadyClosed,
150}
151
152#[derive(Debug, thiserror::Error)]
162#[non_exhaustive]
163pub enum FinalizeError {
164 #[error("finalization task failed: {0}")]
166 Task(#[from] tokio::task::JoinError),
167 #[error("finalization task failed: {0}")]
173 Invariant(String),
174 #[error("receipt issuance failed: {context}")]
176 Receipt {
177 context: String,
180 #[source]
182 source: Option<crate::pipeline::ErrorSource>,
183 },
184 #[error("ATIF finalization failed: {context}")]
186 Atif {
187 context: String,
190 #[source]
192 source: Option<crate::pipeline::ErrorSource>,
193 },
194 #[error("promotion refused: {0}")]
197 Promotion(String),
198 #[error("session capture is incomplete; refusing final artifact")]
200 CaptureIncomplete,
201 #[error("tool execution evidence unavailable: {0}")]
204 ToolEvidence(String),
205 #[error("lifecycle event publication failed: {context}")]
207 Bridge {
208 context: String,
211 #[source]
213 source: Option<crate::pipeline::ErrorSource>,
214 },
215 #[error("lifecycle event publication failed (permanent): {context}")]
225 BridgeConfig {
226 context: String,
229 #[source]
231 source: Option<crate::pipeline::ErrorSource>,
232 },
233}
234
235impl FinalizeError {
236 pub fn receipt(context: impl Into<String>) -> Self {
238 Self::Receipt {
239 context: context.into(),
240 source: None,
241 }
242 }
243
244 pub fn receipt_source<E>(error: E) -> Self
248 where
249 E: std::error::Error + Send + Sync + 'static,
250 {
251 Self::Receipt {
252 context: error.to_string(),
253 source: Some(Box::new(error)),
254 }
255 }
256
257 pub fn atif(context: impl Into<String>) -> Self {
259 Self::Atif {
260 context: context.into(),
261 source: None,
262 }
263 }
264
265 pub fn atif_source<E>(error: E) -> Self
267 where
268 E: std::error::Error + Send + Sync + 'static,
269 {
270 Self::Atif {
271 context: error.to_string(),
272 source: Some(Box::new(error)),
273 }
274 }
275
276 pub fn bridge(context: impl Into<String>) -> Self {
278 Self::Bridge {
279 context: context.into(),
280 source: None,
281 }
282 }
283
284 pub fn bridge_source<E>(error: E) -> Self
286 where
287 E: std::error::Error + Send + Sync + 'static,
288 {
289 Self::Bridge {
290 context: error.to_string(),
291 source: Some(Box::new(error)),
292 }
293 }
294
295 pub fn is_transient_io(&self) -> bool {
309 let root = match self {
310 Self::Atif { source: Some(s), .. } | Self::Receipt { source: Some(s), .. } => &**s,
311 _ => return false,
312 };
313 let mut current: &dyn std::error::Error = root;
314 loop {
315 if let Some(io) = current.downcast_ref::<std::io::Error>() {
316 return matches!(
317 io.kind(),
318 std::io::ErrorKind::StorageFull
319 | std::io::ErrorKind::ReadOnlyFilesystem
320 | std::io::ErrorKind::QuotaExceeded
321 | std::io::ErrorKind::ResourceBusy
322 | std::io::ErrorKind::WouldBlock
323 | std::io::ErrorKind::Interrupted
324 | std::io::ErrorKind::TimedOut
325 );
326 }
327 match current.source() {
328 Some(next) => current = next,
329 None => return false,
330 }
331 }
332 }
333}
334
335impl From<av_bridge::BusError> for FinalizeError {
336 fn from(error: av_bridge::BusError) -> Self {
344 let context = error.to_string();
345 if error.is_permanent() {
346 Self::BridgeConfig {
347 context,
348 source: Some(Box::new(error)),
349 }
350 } else {
351 Self::Bridge {
352 context,
353 source: Some(Box::new(error)),
354 }
355 }
356 }
357}
358
359#[derive(Clone)]
361pub struct Finalizer {
362 signer: Arc<dyn Signer>,
363 spool_dir: PathBuf,
364 metrics: Arc<Registry>,
365 bridge: Option<Arc<dyn EventBus>>,
366 state_store: Option<Arc<dyn av_state::StateStore>>,
370 recovery_lock: Arc<tokio::sync::Mutex<()>>,
371 lifecycle_locks: Arc<SessionLockTable>,
378 quarantined_sessions: Arc<parking_lot::Mutex<QuarantinedSessions>>,
379 warned_artifacts: Arc<parking_lot::Mutex<WarnedArtifacts>>,
386 journal_key: [u8; 32],
387 vector_sink: Option<Arc<dyn av_loopdetect::VectorSink>>,
390}
391
392pub(crate) struct WarnedArtifacts {
404 order: std::collections::VecDeque<PathBuf>,
405 set: std::collections::HashSet<PathBuf>,
406 cap: usize,
407}
408
409impl WarnedArtifacts {
410 fn new(cap: usize) -> Self {
411 let cap = cap.max(1);
420 Self {
421 order: std::collections::VecDeque::new(),
422 set: std::collections::HashSet::new(),
423 cap,
424 }
425 }
426
427 fn insert(&mut self, path: PathBuf) -> bool {
428 if self.set.contains(&path) {
429 return false;
430 }
431 if self.order.len() >= self.cap {
432 if let Some(evicted) = self.order.pop_front() {
433 self.set.remove(&evicted);
434 }
435 }
436 self.order.push_back(path.clone());
437 self.set.insert(path);
438 true
439 }
440
441 #[cfg(test)]
442 fn len(&self) -> usize {
443 self.set.len()
444 }
445}
446
447const WARNED_ARTIFACTS_CAP: usize = 4096;
453
454pub(crate) const MAX_QUARANTINED_SESSIONS: usize = 4096;
472
473pub(crate) struct QuarantinedSessions {
481 order: std::collections::VecDeque<String>,
482 set: std::collections::HashSet<String>,
483 cap: usize,
484}
485
486impl QuarantinedSessions {
487 pub(crate) fn new(cap: usize) -> Self {
488 let cap = cap.max(1);
493 Self {
494 order: std::collections::VecDeque::new(),
495 set: std::collections::HashSet::new(),
496 cap,
497 }
498 }
499
500 pub(crate) fn insert(&mut self, id: String) -> bool {
501 if self.set.contains(&id) {
502 return false;
503 }
504 if self.order.len() >= self.cap {
505 if let Some(evicted) = self.order.pop_front() {
506 self.set.remove(&evicted);
507 }
508 }
509 self.order.push_back(id.clone());
510 self.set.insert(id);
511 true
512 }
513
514 pub(crate) fn contains(&self, id: &str) -> bool {
515 self.set.contains(id)
516 }
517
518 #[cfg(test)]
519 fn len(&self) -> usize {
520 self.set.len()
521 }
522}
523
524#[derive(Default)]
536pub struct SessionLockTable {
537 inner: dashmap::DashMap<String, Arc<tokio::sync::Mutex<()>>>,
538}
539
540impl SessionLockTable {
541 fn arc_for(&self, session_id: &str) -> Arc<tokio::sync::Mutex<()>> {
542 self.inner
544 .entry(session_id.to_owned())
545 .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
546 .value()
547 .clone()
548 }
549
550 fn try_gc(&self, session_id: &str) {
554 self.inner
555 .remove_if(session_id, |_, arc| Arc::strong_count(arc) == 1);
556 }
557
558 #[cfg(test)]
560 pub(crate) fn len(&self) -> usize {
561 self.inner.len()
562 }
563}
564
565pub struct SessionLifecycleGuard {
568 permit: Option<tokio::sync::OwnedMutexGuard<()>>,
569 table: Arc<SessionLockTable>,
570 session_id: String,
571}
572
573impl Drop for SessionLifecycleGuard {
574 fn drop(&mut self) {
575 drop(self.permit.take());
580 self.table.try_gc(&self.session_id);
581 }
582}
583
584struct CloseClaim<'a> {
585 session: &'a Session,
586 committed: bool,
587}
588
589const LIFECYCLE_OUTBOX_SCHEMA_V1: u16 = 1;
597
598fn lifecycle_outbox_schema_v1() -> u16 {
599 LIFECYCLE_OUTBOX_SCHEMA_V1
600}
601
602#[derive(serde::Serialize, serde::Deserialize)]
603struct LifecycleOutbox {
604 #[serde(default = "lifecycle_outbox_schema_v1")]
605 schema_version: u16,
606 session_id: String,
607 kind: String,
608 topic: String,
609 key: String,
610 value: serde_json::Value,
611 ack: Option<av_bridge::PublishAck>,
612 #[serde(default)]
621 session_scope: Option<String>,
622}
623
624#[derive(serde::Serialize, serde::Deserialize)]
625struct AtifProvenance {
626 session_id: String,
627 digest: String,
628}
629
630#[derive(serde::Serialize, serde::Deserialize)]
631struct PromotionMarker {
632 session_id: String,
633 trajectory_digest: String,
634}
635
636#[derive(serde::Serialize, serde::Deserialize)]
657struct CloseCompleteMarker {
658 session_id: String,
659 digest: String,
660}
661
662const CLOSE_COMPLETE_DOMAIN: &str = "unsigned-close-complete";
663
664fn close_complete_marker_path(atif_path: &std::path::Path) -> std::path::PathBuf {
665 atif_path.with_extension("close-complete")
666}
667
668impl Drop for CloseClaim<'_> {
669 fn drop(&mut self) {
670 if !self.committed {
671 self.session.reset_close();
672 }
673 }
674}
675
676struct PromotionClaim<'a> {
687 session: &'a Session,
688 committed: bool,
689}
690
691impl Drop for PromotionClaim<'_> {
692 fn drop(&mut self) {
693 if !self.committed {
694 self.session.reset_promotion();
695 }
696 }
697}
698
699impl Finalizer {
700 pub fn metrics(&self) -> &Arc<Registry> {
704 &self.metrics
705 }
706
707 pub fn new(signer: Arc<dyn Signer>, spool_dir: PathBuf, metrics: Arc<Registry>) -> Self {
709 let journal_key = crate::journal::key_from_signer(signer.as_ref());
710 Self {
711 signer,
712 spool_dir,
713 metrics,
714 bridge: None,
715 state_store: None,
716 recovery_lock: Arc::new(tokio::sync::Mutex::new(())),
717 lifecycle_locks: Arc::new(SessionLockTable::default()),
718 quarantined_sessions: Arc::new(parking_lot::Mutex::new(QuarantinedSessions::new(
719 MAX_QUARANTINED_SESSIONS,
720 ))),
721 warned_artifacts: Arc::new(parking_lot::Mutex::new(WarnedArtifacts::new(
722 WARNED_ARTIFACTS_CAP,
723 ))),
724 journal_key,
725 vector_sink: None,
726 }
727 }
728
729 pub fn with_bridge(
731 signer: Arc<dyn Signer>,
732 spool_dir: PathBuf,
733 metrics: Arc<Registry>,
734 bridge: Arc<dyn EventBus>,
735 ) -> Self {
736 let journal_key = crate::journal::key_from_signer(signer.as_ref());
737 Self {
738 signer,
739 spool_dir,
740 metrics,
741 bridge: Some(bridge),
742 state_store: None,
743 recovery_lock: Arc::new(tokio::sync::Mutex::new(())),
744 lifecycle_locks: Arc::new(SessionLockTable::default()),
745 quarantined_sessions: Arc::new(parking_lot::Mutex::new(QuarantinedSessions::new(
746 MAX_QUARANTINED_SESSIONS,
747 ))),
748 warned_artifacts: Arc::new(parking_lot::Mutex::new(WarnedArtifacts::new(
749 WARNED_ARTIFACTS_CAP,
750 ))),
751 journal_key,
752 vector_sink: None,
753 }
754 }
755
756 async fn acquire_lifecycle(&self, session_id: &str) -> SessionLifecycleGuard {
761 let arc = self.lifecycle_locks.arc_for(session_id);
762 let permit = arc.lock_owned().await;
763 SessionLifecycleGuard {
764 permit: Some(permit),
765 table: Arc::clone(&self.lifecycle_locks),
766 session_id: session_id.to_owned(),
767 }
768 }
769
770 pub async fn prune_sealed_atif(&self, max_age: std::time::Duration) -> Result<usize, FinalizeError> {
776 let spool_dir = self.spool_dir.clone();
777 tokio::task::spawn_blocking(move || {
778 prune_sealed_atif_blocking(&spool_dir, max_age).map_err(FinalizeError::atif_source)
779 })
780 .await
781 .map_err(FinalizeError::Task)?
782 }
783}
784
785pub fn prune_sealed_atif_blocking(
810 spool_dir: &std::path::Path,
811 max_age: std::time::Duration,
812) -> std::io::Result<usize> {
813 let entries = match std::fs::read_dir(spool_dir) {
814 Ok(entries) => entries,
815 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(0),
816 Err(error) => return Err(error),
817 };
818 let mut pruned = 0usize;
819 let now = std::time::SystemTime::now();
820 let mut dir_changed = false;
821 for entry in entries {
822 let entry = entry?;
823 let path = entry.path();
824 let extension = path.extension().and_then(std::ffi::OsStr::to_str);
831 if extension == Some("close-complete") {
832 let is_plain_stem_marker = path
842 .file_stem()
843 .and_then(std::ffi::OsStr::to_str)
844 .is_some_and(|stem| stem.len() == 32 && stem.chars().all(|c| c.is_ascii_hexdigit()));
845 if is_plain_stem_marker && !path.with_extension("json").exists() {
846 let old_enough = entry
847 .metadata()
848 .and_then(|m| m.modified())
849 .ok()
850 .map(|modified| {
851 now.duration_since(modified).unwrap_or(std::time::Duration::ZERO) >= max_age
852 })
853 .unwrap_or(false);
854 if old_enough {
855 match std::fs::remove_file(&path) {
856 Ok(()) => dir_changed = true,
857 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
858 Err(error) => return Err(error),
859 }
860 }
861 }
862 continue;
863 }
864 if extension != Some("json") {
865 continue;
866 }
867 if path
868 .file_name()
869 .and_then(std::ffi::OsStr::to_str)
870 .is_some_and(|name| name.ends_with(".session.json"))
871 {
872 continue;
873 }
874 let sidecar = path.with_extension("atif-auth");
875 if !sidecar.exists() {
876 continue;
877 }
878 if path.with_extension("promote").exists() {
891 continue;
892 }
893 let metadata = match entry.metadata() {
894 Ok(m) => m,
895 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
896 Err(error) => return Err(error),
897 };
898 let modified = match metadata.modified() {
899 Ok(t) => t,
900 Err(_) => continue,
901 };
902 let age = now.duration_since(modified).unwrap_or(std::time::Duration::ZERO);
903 if age < max_age {
904 continue;
905 }
906 let mut removed_any = false;
911 match std::fs::remove_file(&sidecar) {
912 Ok(()) => removed_any = true,
913 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
914 Err(error) => return Err(error),
915 }
916 match std::fs::remove_file(&path) {
917 Ok(()) => removed_any = true,
918 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
919 Err(error) => return Err(error),
920 }
921 match std::fs::remove_file(path.with_extension("close-complete")) {
927 Ok(()) => {}
928 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
929 Err(error) => return Err(error),
930 }
931 if removed_any {
932 pruned += 1;
933 dir_changed = true;
934 }
935 }
936 if dir_changed {
937 av_core::fsutil::sync_directory(spool_dir)?;
938 }
939 Ok(pruned)
940}
941
942impl Finalizer {
943 #[cfg(test)]
944 pub(crate) fn lifecycle_locks(&self) -> Arc<SessionLockTable> {
945 Arc::clone(&self.lifecycle_locks)
946 }
947
948 fn warn_once(&self, path: PathBuf) -> bool {
957 self.warned_artifacts.lock().insert(path)
958 }
959
960 #[must_use]
963 pub fn with_state_store(mut self, store: Arc<dyn av_state::StateStore>) -> Self {
964 self.state_store = Some(store);
965 self
966 }
967
968 pub fn with_vector_sink(mut self, sink: Arc<dyn av_loopdetect::VectorSink>) -> Self {
972 self.vector_sink = Some(sink);
973 self
974 }
975
976 fn clear_budget_state(&self, session_id: &str) {
985 if let Some(store) = self.state_store.as_deref() {
986 store.remove_prefix(&av_state::ActionBudget::session_prefix(session_id));
987 }
988 }
989
990 #[tracing::instrument(
993 name = "agentvisor.session.close",
994 skip_all,
995 fields(session.id = %session.id, workflow = ?session.workflow)
996 )]
997 pub async fn close_session(
998 &self,
999 session: Arc<Session>,
1000 stop_reason: StopReason,
1001 ) -> Result<FinalizeOutcome, FinalizeError> {
1002 let _lifecycle = self.acquire_lifecycle(&session.id).await;
1003 self.close_session_locked(session, stop_reason).await
1004 }
1005
1006 async fn close_session_locked(
1007 &self,
1008 session: Arc<Session>,
1009 stop_reason: StopReason,
1010 ) -> Result<FinalizeOutcome, FinalizeError> {
1011 let close_guard = session.close_guard();
1012 if !session.try_close() {
1013 return Ok(FinalizeOutcome::AlreadyClosed);
1014 }
1015 drop(close_guard);
1016 let mut claim = CloseClaim {
1017 session: &session,
1018 committed: false,
1019 };
1020 session.wait_for_streams().await;
1021 session.wait_for_worker_jobs().await;
1022 if session.capture_failed() {
1023 self.emit_quarantine_event(&session).await?;
1024 self.metrics
1025 .counter(
1026 "av_incomplete_sessions_total",
1027 "Sessions refused due to incomplete capture",
1028 )
1029 .inc();
1030 session.mark_artifact_committed();
1034 claim.committed = true;
1035 self.clear_budget_state(&session.id);
1036 self.quarantine_tool_executions(&session.id).await;
1039 return Err(FinalizeError::CaptureIncomplete);
1040 }
1041 crate::routes::ensure_tool_executions_resolved(&self.spool_dir, &session.id, &self.journal_key)
1047 .await
1048 .map_err(FinalizeError::ToolEvidence)?;
1049 let started = Instant::now();
1050 struct FinalizeObserver<'a> {
1059 metrics: &'a Registry,
1060 started: Instant,
1061 }
1062 impl Drop for FinalizeObserver<'_> {
1063 fn drop(&mut self) {
1064 self.metrics
1065 .histogram(
1066 "av_session_finalize_duration_seconds",
1067 "Session finalization latency",
1068 )
1069 .observe_us(elapsed_us(self.started));
1070 }
1071 }
1072 let _finalize_observer = FinalizeObserver {
1073 metrics: self.metrics.as_ref(),
1074 started,
1075 };
1076 let outcome = match session.workflow {
1077 Workflow::Signed => {
1078 let subject = {
1079 let chain = session.chain.lock();
1080 ReceiptSubject::EventChain {
1081 chain_head: chain.head_hex(),
1082 event_count: chain.count(),
1083 }
1084 };
1085 let persisted_receipt = { session.receipt.lock().clone() };
1086 let receipt = if let Some(receipt) = persisted_receipt {
1087 let verified = self.verify_configured_receipt(&receipt).and_then(|()| {
1098 if receipt.body.subject == subject {
1099 Ok(())
1100 } else {
1101 Err(FinalizeError::receipt(
1102 "persisted receipt subject does not match reconstructed chain".to_owned(),
1103 ))
1104 }
1105 });
1106 if let Err(error) = verified {
1107 tracing::warn!(
1108 session = %session.id,
1109 error = &error as &dyn std::error::Error,
1110 "persisted receipt does not verify; sealing session terminally \
1111 (evidence quarantined, no per-tick retry)"
1112 );
1113 session.mark_capture_failed();
1114 self.emit_quarantine_event(&session).await?;
1115 session.mark_artifact_committed();
1116 claim.committed = true;
1117 self.clear_budget_state(&session.id);
1118 self.quarantine_tool_executions(&session.id).await;
1119 self.quarantined_sessions.lock().insert(session.id.clone());
1120 return Err(error);
1121 }
1122 receipt
1123 } else {
1124 let body = session.receipt_body(subject, stop_reason);
1125 let sign_started = Instant::now();
1126 let receipt_result = Receipt::issue(body, self.signer.as_ref());
1127 self.metrics
1136 .histogram("av_receipt_sign_duration_seconds", "Receipt signing latency")
1137 .observe_us(elapsed_us(sign_started));
1138 let receipt = receipt_result.map_err(FinalizeError::receipt_source)?;
1139 self.persist_receipt(&session.id, &receipt).await?;
1140 *session.receipt.lock() = Some(receipt.clone());
1141 receipt
1142 };
1143 session.mark_artifact_committed();
1144 self.emit_receipt_event(&session, &receipt).await?;
1145 FinalizeOutcome::Receipt {
1146 receipt: Box::new(receipt),
1147 }
1148 }
1149 Workflow::Unsigned => {
1150 let existing_path = { session.atif_path.lock().clone() };
1151 let path = if let Some(path) = existing_path {
1152 path
1153 } else {
1154 let mut trajectory = match self.rebuild_unsigned_trajectory(&session).await? {
1171 Some(trajectory) => trajectory,
1172 None => session.snapshot_trajectory(),
1173 };
1174 if trajectory.steps.is_empty() {
1177 session.mark_artifact_committed();
1178 claim.committed = true;
1179 self.clear_budget_state(&session.id);
1180 return Err(FinalizeError::atif(
1181 "cannot finalize an unsigned session with no captured steps".to_owned(),
1182 ));
1183 }
1184 let identity = session.current_identity();
1185 trajectory.agent.extra = Some(serde_json::json!({
1186 "charter": identity.charter,
1187 "instance_uid": identity.instance_uid,
1188 "ttl_remaining_s": identity.ttl_remaining_s,
1189 "principal_binding": session.principal_binding(),
1190 }));
1191 if let Some(metrics) = trajectory.final_metrics.as_mut() {
1192 metrics.total_prompt_tokens = Some(
1193 session
1194 .totals
1195 .prompt_tokens
1196 .load(std::sync::atomic::Ordering::Acquire),
1197 );
1198 metrics.total_completion_tokens = Some(
1199 session
1200 .totals
1201 .completion_tokens
1202 .load(std::sync::atomic::Ordering::Acquire),
1203 );
1204 metrics.total_cached_tokens = Some(
1205 session
1206 .totals
1207 .cached_tokens
1208 .load(std::sync::atomic::Ordering::Acquire),
1209 );
1210 metrics.total_cost_usd = Some(
1211 session
1212 .totals
1213 .cost_usd_micros
1214 .load(std::sync::atomic::Ordering::Acquire)
1215 as f64
1216 / av_core::units::USD_MICROS_PER_DOLLAR as f64,
1217 );
1218 metrics.extra = Some(serde_json::json!({
1219 "tool_calls": session.totals.tool_calls.load(std::sync::atomic::Ordering::Acquire),
1220 "tool_allowed": session.totals.tool_allowed.load(std::sync::atomic::Ordering::Acquire),
1221 "tool_blocked": session.totals.tool_blocked.load(std::sync::atomic::Ordering::Acquire),
1222 "cost_usd_micros": session.totals.cost_usd_micros.load(std::sync::atomic::Ordering::Acquire),
1223 "stop_reason_id": session.recorded_stop_reason_id(),
1224 "enforcement_latched": session.enforcement_latched_id(),
1231 }));
1232 }
1233 let name = format!(
1234 "{}.json",
1235 &av_core::digest::sha256_hex(session.id.as_bytes())[..32]
1236 );
1237 let path = self.spool_dir.join(name);
1238 let write_path = path.clone();
1239 let archive_session = session.id.clone();
1240 let new_trajectory_id = trajectory.trajectory_id.clone();
1241 tokio::task::spawn_blocking(move || {
1242 archive_conflicting_atif(&write_path, new_trajectory_id.as_deref(), &archive_session)
1246 .map_err(av_atif::writer::WriterError::Io)?;
1247 av_atif::write_atomic(&trajectory, &write_path)
1248 })
1249 .await
1250 .map_err(FinalizeError::Task)?
1251 .map_err(FinalizeError::atif_source)?;
1252 path
1253 };
1254 self.ensure_atif_provenance(&path, &session.id).await?;
1255 *session.atif_path.lock() = Some(path.clone());
1274 session.mark_artifact_committed();
1275 FinalizeOutcome::Atif { path }
1276 }
1277 };
1278 let workflow = session.workflow.as_str();
1279 self.emit_bridge_event(
1280 &session,
1281 av_events::EventClass::Session,
1282 serde_json::json!({"action": "closed", "workflow": workflow}),
1283 crate::journal::SESSION_CLOSE_OUTBOX_KIND,
1284 )
1285 .await?;
1286 self.remove_step_journal(&session.id).await?;
1287 self.remove_tool_executions(&session.id).await?;
1288 self.remove_lifecycle_outbox(&session.id, crate::journal::RECEIPT_OUTBOX_KIND)
1289 .await?;
1290 self.remove_lifecycle_outbox(&session.id, crate::journal::SESSION_CLOSE_OUTBOX_KIND)
1291 .await?;
1292 self.metrics
1300 .counter("av_sessions_finalized_total", "Sessions finalized")
1301 .inc();
1302 claim.committed = true;
1303 if session.workflow == Workflow::Unsigned {
1308 self.persist_close_complete_marker(&session).await;
1309 }
1310 self.clear_budget_state(&session.id);
1317 session.mark_close_complete();
1320 if let Some(sink) = &self.vector_sink {
1326 if let Err(error) = sink.delete_scope(&session.session_scope).await {
1327 tracing::warn!(
1328 session = %session.id,
1329 %error,
1330 "vector-store scope cleanup failed at close (points remain as dead weight)"
1331 );
1332 }
1333 }
1334 if session.workflow == Workflow::Unsigned {
1343 session.drain_trajectory_builder();
1344 }
1345 session.loop_state.release_embedding();
1349 Ok(outcome)
1350 }
1351
1352 #[tracing::instrument(
1354 name = "agentvisor.session.promote",
1355 skip_all,
1356 fields(session.id = %session.id)
1357 )]
1358 pub async fn promote(&self, session: Arc<Session>) -> Result<Receipt, FinalizeError> {
1359 let _lifecycle = self.acquire_lifecycle(&session.id).await;
1360 if session.workflow != Workflow::Unsigned {
1361 return session
1362 .receipt
1363 .lock()
1364 .clone()
1365 .ok_or_else(|| FinalizeError::Promotion("signed session has no issued receipt".to_owned()));
1366 }
1367 if !session.is_closed() {
1368 self.close_session_locked(Arc::clone(&session), StopReason::SessionClosed)
1369 .await?;
1370 }
1371 if session.is_promoted() {
1392 let persisted_receipt = { session.receipt.lock().clone() };
1393 let receipt = persisted_receipt.ok_or_else(|| {
1394 FinalizeError::Promotion("promoted session has no persisted receipt".to_owned())
1395 })?;
1396 let atif_path_opt: Option<std::path::PathBuf> = { session.atif_path.lock().clone() };
1413 if let Some(atif_path) = atif_path_opt {
1414 let marker = atif_path.with_extension("promote");
1415 if marker.exists() {
1416 if let Err(error) = remove_outbox(&marker).await {
1417 tracing::warn!(
1418 %error,
1419 path = %av_core::fsutil::basename(&marker),
1420 "failed to clean up orphan promotion marker (promotion still succeeded)"
1421 );
1422 }
1423 }
1424 }
1425 return Ok(receipt);
1426 }
1427 if session.capture_failed() {
1428 return Err(FinalizeError::CaptureIncomplete);
1429 }
1430 crate::routes::ensure_tool_executions_resolved(&self.spool_dir, &session.id, &self.journal_key)
1434 .await
1435 .map_err(FinalizeError::ToolEvidence)?;
1436 let persisted_receipt = { session.receipt.lock().clone() };
1437 let path =
1438 session.atif_path.lock().clone().ok_or_else(|| {
1439 FinalizeError::Promotion("session has no persisted ATIF artifact".to_owned())
1440 })?;
1441 let marker = path.with_extension("promote");
1442 if !path.with_extension("atif-auth").exists() {
1443 return Err(FinalizeError::atif(
1444 "ATIF artifact has no authenticated provenance".to_owned(),
1445 ));
1446 }
1447 self.ensure_atif_provenance(&path, &session.id).await?;
1448 let bytes = read_capped_async(path.clone(), av_core::fsutil::MAX_ATIF_BYTES)
1449 .await
1450 .map_err(FinalizeError::atif_source)?;
1451 let trajectory: av_atif::Trajectory =
1452 serde_json::from_slice(&bytes).map_err(FinalizeError::atif_source)?;
1453 let issues = match av_atif::validate_bytes(&bytes, av_atif::Mode::Strict) {
1460 Ok(issues) => issues,
1461 Err(reason) => {
1462 return Err(FinalizeError::atif(format!(
1463 "strict validation failed (bytes-level): {reason}"
1464 )));
1465 }
1466 };
1467 if !issues.is_empty() {
1468 const RENDER_ISSUE_HEAD: usize = 16;
1475 let total = issues.len();
1476 let head: Vec<_> = issues.iter().take(RENDER_ISSUE_HEAD).collect();
1477 return Err(FinalizeError::atif(format!(
1478 "strict validation failed ({total} issues, showing first {}): {head:?}",
1479 head.len()
1480 )));
1481 }
1482 let trajectory_digest = av_core::digest::sha256_hex(&bytes);
1483 let subject = ReceiptSubject::AtifTrajectory {
1484 trajectory_digest: trajectory_digest.clone(),
1485 step_count: trajectory.steps.len() as u64,
1486 retroactive: true,
1487 };
1488 let marker_payload = PromotionMarker {
1489 session_id: session.id.clone(),
1490 trajectory_digest,
1491 };
1492 if marker.exists() {
1493 let sealed = read_capped_async(marker.clone(), av_core::fsutil::MAX_CONTROL_BYTES)
1494 .await
1495 .map_err(FinalizeError::atif_source)?;
1496 let actual: PromotionMarker =
1497 crate::journal::open(&self.journal_key, "promotion-marker", 0, &sealed)
1498 .map_err(FinalizeError::atif)?;
1499 if actual.session_id != marker_payload.session_id
1500 || actual.trajectory_digest != marker_payload.trajectory_digest
1501 {
1502 return Err(FinalizeError::atif(
1503 "promotion marker does not match session and trajectory".to_owned(),
1504 ));
1505 }
1506 } else {
1507 let sealed = crate::journal::seal(&self.journal_key, "promotion-marker", 0, &marker_payload)
1508 .map_err(FinalizeError::atif)?;
1509 persist_marker(&marker, &sealed).await?;
1510 }
1511 if !session.try_promote() {
1512 return Err(FinalizeError::Promotion(
1513 "promotion is already in progress".to_owned(),
1514 ));
1515 }
1516 let mut claim = PromotionClaim {
1520 session: &session,
1521 committed: false,
1522 };
1523 let receipt = if let Some(receipt) = persisted_receipt {
1524 self.verify_configured_receipt(&receipt)?;
1525 if receipt.body.subject != subject {
1526 return Err(FinalizeError::receipt(
1527 "persisted promotion receipt does not match ATIF artifact".to_owned(),
1528 ));
1529 }
1530 receipt
1531 } else {
1532 let body = session.receipt_body(subject, StopReason::SessionClosed);
1533 let sign_started = Instant::now();
1544 let receipt_result = Receipt::issue(body, self.signer.as_ref());
1545 self.metrics
1546 .histogram("av_receipt_sign_duration_seconds", "Receipt signing latency")
1547 .observe_us(elapsed_us(sign_started));
1548 let receipt = receipt_result.map_err(FinalizeError::receipt_source)?;
1549 self.persist_receipt(&session.id, &receipt).await?;
1550 *session.receipt.lock() = Some(receipt.clone());
1551 receipt
1552 };
1553 self.emit_receipt_event(&session, &receipt).await?;
1554 claim.committed = true;
1555 session.finish_promotion();
1556 remove_outbox(&marker).await?;
1557 self.remove_lifecycle_outbox(&session.id, crate::journal::RECEIPT_OUTBOX_KIND)
1558 .await?;
1559 self.metrics
1560 .counter("av_sessions_promoted_total", "Unsigned sessions promoted")
1561 .inc();
1562 Ok(receipt)
1563 }
1564
1565 #[tracing::instrument(name = "agentvisor.recovery", skip_all)]
1571 pub async fn recover_spooled_sessions(
1572 &self,
1573 sessions: &SessionRegistry,
1574 breaker: &av_loopdetect::BreakerConfig,
1575 ) -> Result<usize, FinalizeError> {
1576 let _recovery = self.recovery_lock.lock().await;
1577 let warn_once = |path: PathBuf| self.warn_once(path);
1583 let ctx = crate::recovery::ReconcilerContext {
1584 spool_dir: &self.spool_dir,
1585 metrics: &self.metrics,
1586 sessions,
1587 journal_key: &self.journal_key,
1588 quarantined_sessions: &self.quarantined_sessions,
1589 bridge: self.bridge.as_ref(),
1590 warn_once: &warn_once,
1591 };
1592 let passes: [&dyn crate::recovery::RecoveryPass; 6] = [
1609 &crate::recovery::QuarantineIncompleteEffectsPass,
1610 &crate::recovery::ReplayLifecycleOutboxesPass,
1611 &crate::recovery::RecoverSignedJournalsPass {
1612 finalizer: self,
1613 breaker,
1614 },
1615 &crate::recovery::ConsolidateStepJournalsPass {
1616 finalizer: self,
1617 breaker,
1618 },
1619 &crate::recovery::QuarantineOrphanJsonPass,
1620 &crate::recovery::AdoptStrictAtifPass {
1621 finalizer: self,
1622 breaker,
1623 },
1624 ];
1625 let mut recovered = 0usize;
1626 for pass in passes {
1627 recovered = recovered.saturating_add(crate::recovery::run_pass(pass, &ctx).await?.recovered);
1628 }
1629 self.ensure_unresolved_tool_metadata(sessions).await?;
1630 for session in sessions.open_sessions_including_closed() {
1634 if session.capture_failed() {
1635 let _lifecycle = self.acquire_lifecycle(&session.id).await;
1636 if let Err(error) = self.emit_quarantine_event(&session).await {
1637 tracing::warn!(session = %session.id, %error, "quarantine notification pending; will retry");
1638 }
1639 }
1640 }
1641 Ok(recovered)
1642 }
1643
1644 async fn ensure_unresolved_tool_metadata(&self, sessions: &SessionRegistry) -> Result<(), FinalizeError> {
1650 let unresolved = crate::routes::unresolved_tool_sessions(&self.spool_dir, &self.journal_key)
1651 .await
1652 .map_err(FinalizeError::atif)?;
1653 for session_id in unresolved {
1654 if sessions.get(&session_id).is_some() {
1655 continue;
1656 }
1657 let digest = av_core::digest::sha256_hex(session_id.as_bytes());
1658 let stem = digest
1659 .get(..32)
1660 .ok_or_else(|| FinalizeError::Invariant("invalid session digest".to_owned()))?;
1661 let metadata_path = self.spool_dir.join(format!("{stem}.session.json"));
1662 let invalid = || {
1663 FinalizeError::atif(
1664 "unresolved tool execution has no recoverable authenticated session metadata".to_owned(),
1665 )
1666 };
1667 let metadata = self
1668 .read_journal_metadata(&metadata_path)
1669 .await?
1670 .ok_or_else(invalid)?;
1671 let recognized_workflow = metadata
1672 .get("workflow")
1673 .and_then(serde_json::Value::as_str)
1674 .and_then(Workflow::parse)
1675 .is_some();
1676 let identity = metadata
1677 .get("identity")
1678 .cloned()
1679 .and_then(|value| serde_json::from_value::<av_events::AgentIdentity>(value).ok());
1680 if metadata
1681 .get("journal_version")
1682 .and_then(serde_json::Value::as_u64)
1683 != Some(2)
1684 || metadata.get("session_id").and_then(serde_json::Value::as_str) != Some(session_id.as_str())
1685 || !recognized_workflow
1686 || identity.is_none()
1687 {
1688 return Err(invalid());
1689 }
1690 }
1691 Ok(())
1692 }
1693
1694 pub(crate) async fn adopt_strict_atif_artifacts(
1699 &self,
1700 sessions: &SessionRegistry,
1701 breaker: &av_loopdetect::BreakerConfig,
1702 ) -> Result<usize, FinalizeError> {
1703 let mut entries = match tokio::fs::read_dir(&self.spool_dir).await {
1704 Ok(entries) => entries,
1705 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(0),
1706 Err(error) => return Err(FinalizeError::atif_source(error)),
1707 };
1708 let known_stems: std::collections::HashSet<String> = sessions
1718 .open_sessions_including_closed()
1719 .iter()
1720 .map(|session| {
1721 let digest = av_core::digest::sha256_hex(session.id.as_bytes());
1722 digest.get(..32).unwrap_or(&digest).to_owned()
1723 })
1724 .collect();
1725 let mut recovered = 0usize;
1726 let mut examined = 0usize;
1727 let mut dirents_seen = 0usize;
1728 while let Some(entry) = entries.next_entry().await.map_err(FinalizeError::atif_source)? {
1729 dirents_seen = dirents_seen.saturating_add(1);
1730 if dirents_seen > MAX_RECOVERY_DIRENTS_PER_TICK {
1731 bump_recovery_scan_cap(&self.metrics, "adopt_strict_atif", examined, dirents_seen);
1732 break;
1733 }
1734 let path = entry.path();
1735 if path.extension().and_then(std::ffi::OsStr::to_str) != Some("json") {
1736 continue;
1737 }
1738 if path
1743 .file_name()
1744 .and_then(std::ffi::OsStr::to_str)
1745 .is_some_and(|name| name.ends_with(".session.json"))
1746 {
1747 continue;
1748 }
1749 if path
1753 .file_stem()
1754 .and_then(std::ffi::OsStr::to_str)
1755 .is_some_and(|stem| known_stems.contains(stem))
1756 {
1757 continue;
1758 }
1759 if !path.with_extension("atif-auth").exists() {
1775 continue;
1776 }
1777 examined = examined.saturating_add(1);
1786 if examined > MAX_RECOVERY_ENTRIES_PER_TICK {
1787 bump_recovery_scan_cap(&self.metrics, "adopt_strict_atif", examined, dirents_seen);
1788 break;
1789 }
1790 let metadata = match tokio::fs::metadata(&path).await {
1796 Ok(m) => m,
1797 Err(error) => {
1798 tracing::warn!(%error, path = %av_core::fsutil::basename(&path), "skipping ATIF spool file whose metadata is unreadable");
1799 continue;
1800 }
1801 };
1802 if metadata.len() > MAX_ATIF_RECOVERY_BYTES {
1803 self.metrics
1804 .counter(
1805 "av_atif_recovery_skipped_total{reason=\"too_large\"}",
1806 "ATIF spool files skipped during recovery",
1807 )
1808 .inc();
1809 tracing::warn!(
1810 path = %av_core::fsutil::basename(&path),
1811 size = metadata.len(),
1812 max = MAX_ATIF_RECOVERY_BYTES,
1813 "ignoring oversize ATIF spool file",
1814 );
1815 continue;
1816 }
1817 let bytes = match read_capped_async(path.clone(), av_core::fsutil::MAX_ATIF_BYTES).await {
1825 Ok(bytes) => bytes,
1826 Err(error) => {
1827 self.metrics
1828 .counter(
1829 "av_atif_recovery_skipped_total{reason=\"read_error\"}",
1830 "ATIF spool files skipped during recovery",
1831 )
1832 .inc();
1833 tracing::warn!(%error, path = %av_core::fsutil::basename(&path), "skipping unreadable ATIF spool file");
1834 continue;
1835 }
1836 };
1837 let trajectory: av_atif::Trajectory = match serde_json::from_slice(&bytes) {
1838 Ok(trajectory) => trajectory,
1839 Err(error) => {
1840 self.metrics
1841 .counter(
1842 "av_atif_recovery_skipped_total{reason=\"invalid_json\"}",
1843 "ATIF spool files skipped during recovery",
1844 )
1845 .inc();
1846 tracing::warn!(%error, path = %av_core::fsutil::basename(&path), "ignoring invalid ATIF spool file");
1847 continue;
1848 }
1849 };
1850 let bytes_issues = match av_atif::validate_bytes(&bytes, av_atif::Mode::Strict) {
1854 Ok(issues) => issues,
1855 Err(reason) => {
1856 self.metrics
1857 .counter(
1858 "av_atif_recovery_skipped_total{reason=\"invalid_json\"}",
1859 "ATIF spool files skipped during recovery",
1860 )
1861 .inc();
1862 tracing::warn!(
1863 reason = %reason,
1864 path = %av_core::fsutil::basename(&path),
1865 "ignoring ATIF spool file rejected at bytes level"
1866 );
1867 continue;
1868 }
1869 };
1870 if !bytes_issues.is_empty()
1871 || !av_atif::validate_trajectory(&trajectory, av_atif::Mode::Strict).is_empty()
1872 {
1873 self.metrics
1874 .counter(
1875 "av_atif_recovery_skipped_total{reason=\"nonconformant\"}",
1876 "ATIF spool files skipped during recovery",
1877 )
1878 .inc();
1879 tracing::warn!(path = %av_core::fsutil::basename(&path), "ignoring nonconformant ATIF spool file");
1880 continue;
1881 }
1882 let Some(session_id) = trajectory.session_id.clone() else {
1883 continue;
1884 };
1885 if let Err(error) = self
1890 .ensure_atif_provenance_from_bytes(&path, &session_id, &bytes)
1891 .await
1892 {
1893 self.metrics
1894 .counter(
1895 "av_atif_recovery_skipped_total{reason=\"provenance\"}",
1896 "ATIF spool files skipped during recovery",
1897 )
1898 .inc();
1899 if self.warn_once(path.clone()) {
1900 tracing::warn!(
1901 %error,
1902 path = %av_core::fsutil::basename(&path),
1903 "ignoring ATIF spool file whose provenance does not verify"
1904 );
1905 }
1906 if matches!(
1910 &error,
1911 FinalizeError::Atif { source: Some(source), .. }
1912 if source.downcast_ref::<std::io::Error>().is_some()
1913 ) {
1914 continue;
1915 }
1916 const MIN_ORPHAN_AGE: std::time::Duration = std::time::Duration::from_secs(60);
1933 let age = tokio::fs::metadata(&path)
1934 .await
1935 .ok()
1936 .and_then(|meta| meta.modified().ok())
1937 .and_then(|mtime| std::time::SystemTime::now().duration_since(mtime).ok());
1938 if !age.is_some_and(|age| age >= MIN_ORPHAN_AGE) {
1939 continue;
1940 }
1941 let mut quarantine = path.clone();
1942 let stem = quarantine
1943 .file_name()
1944 .and_then(std::ffi::OsStr::to_str)
1945 .unwrap_or("provenance-fail-atif")
1946 .to_owned();
1947 let new_name = format!("{stem}.corrupt-{}", av_core::new_event_uid());
1948 quarantine.set_file_name(new_name);
1949 if let Err(rename_error) = tokio::fs::rename(&path, &quarantine).await {
1950 if self.warn_once(path.clone()) {
1951 tracing::warn!(
1952 %rename_error,
1953 path = %av_core::fsutil::basename(&path),
1954 "failed to quarantine ATIF file with bad provenance; will retry next tick"
1955 );
1956 }
1957 continue;
1958 }
1959 let sidecar = path.with_extension("atif-auth");
1970 let mut quarantined_sidecar = quarantine.clone();
1971 quarantined_sidecar.as_mut_os_string().push(".atif-auth");
1972 match tokio::fs::rename(&sidecar, &quarantined_sidecar).await {
1973 Ok(()) => {}
1974 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
1975 Err(error) => {
1976 tracing::warn!(
1977 %error,
1978 path = %av_core::fsutil::basename(&sidecar),
1979 "failed to quarantine the stale provenance sidecar; future closes \
1980 of this session id will fail provenance until it is removed"
1981 );
1982 }
1983 }
1984 continue;
1985 }
1986 if sessions.get(&session_id).is_some() {
1987 continue;
1988 }
1989 enum AtifCandidateOutcome {
2001 Recovered,
2002 Skipped,
2003 }
2004 let outcome: Result<AtifCandidateOutcome, FinalizeError> = async {
2005 let _lifecycle = self.acquire_lifecycle(&session_id).await;
2009 let extra = trajectory.agent.extra.as_ref();
2010 let instance_uid = extra
2011 .and_then(|value| value.get("instance_uid"))
2012 .and_then(serde_json::Value::as_str)
2013 .unwrap_or("recovered")
2014 .to_owned();
2015 let charter = extra
2016 .and_then(|value| value.get("charter"))
2017 .and_then(|value| {
2018 serde_json::from_value::<av_events::CharterFile>(value.clone())
2019 .ok()
2020 .or_else(|| value.as_str().map(Into::into))
2021 })
2022 .unwrap_or_else(|| "recovered".into());
2023 let ttl_remaining_s = extra
2024 .and_then(|value| value.get("ttl_remaining_s"))
2025 .and_then(serde_json::Value::as_u64);
2026 let recovered_session = match sessions.try_insert_recovered(
2027 Session::recover_unsigned_with_principal(
2028 session_id.clone(),
2029 av_events::AgentIdentity {
2030 version: trajectory.agent.version.clone(),
2031 charter,
2032 instance_uid,
2033 ttl_remaining_s,
2034 },
2035 breaker.clone(),
2036 path.clone(),
2037 trajectory.final_metrics.as_ref(),
2038 trajectory.steps.len() as u64,
2046 extra.and_then(|value| value.get("principal_binding")).and_then(serde_json::Value::as_str),
2047 )
2048 .map_err(FinalizeError::atif)?,
2049 ) {
2050 Ok(inserted) => inserted,
2051 Err(_active) => {
2052 tracing::info!(session = %av_core::fsutil::basename(&path), "unsigned recovery skipped: session already active");
2053 return Ok(AtifCandidateOutcome::Skipped);
2054 }
2055 };
2056 let marker_path = close_complete_marker_path(&path);
2067 match read_capped_async(marker_path.clone(), av_core::fsutil::MAX_CONTROL_BYTES).await {
2068 Ok(sealed) => {
2069 match crate::journal::open::<CloseCompleteMarker>(
2070 &self.journal_key,
2071 CLOSE_COMPLETE_DOMAIN,
2072 0,
2073 &sealed,
2074 ) {
2075 Ok(marker)
2076 if marker.session_id == session_id
2077 && marker.digest == av_core::digest::sha256_hex(&bytes) =>
2078 {
2079 recovered_session.mark_close_complete();
2080 recovered_session.advance_seq_past(trajectory.steps.len() as u64);
2090 }
2091 Ok(_) | Err(_) => {
2092 tracing::warn!(
2096 session = %av_core::fsutil::basename(&path),
2097 "close-complete marker does not verify for this artifact; removing it"
2098 );
2099 if let Err(error) = tokio::fs::remove_file(&marker_path).await {
2100 if error.kind() != std::io::ErrorKind::NotFound {
2101 tracing::warn!(%error, "failed to remove stale close-complete marker");
2102 }
2103 }
2104 }
2105 }
2106 }
2107 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
2108 Err(error) => {
2109 tracing::warn!(
2110 %error,
2111 session = %av_core::fsutil::basename(&path),
2112 "close-complete marker unreadable; the pending-close sweep will re-drive the tail"
2113 );
2114 }
2115 }
2116 if self.quarantined_sessions.lock().contains(&session_id) {
2123 recovered_session.mark_capture_failed();
2124 }
2125 let receipt_path = self.receipt_path(&recovered_session.id);
2126 let receipt_restore: Result<(), FinalizeError> = async {
2146 match tokio::fs::metadata(&receipt_path).await {
2147 Ok(_) => {
2148 let bytes_receipt = read_capped_async(
2149 receipt_path.clone(),
2150 av_core::fsutil::MAX_RECEIPT_BYTES,
2151 )
2152 .await
2153 .map_err(|error| {
2154 FinalizeError::Receipt {
2155 context: format!("existing receipt unreadable: {error}"),
2156 source: Some(Box::new(error)),
2157 }
2158 })?;
2159 let receipt = Receipt::from_json_slice(&bytes_receipt)
2170 .map_err(FinalizeError::receipt_source)?;
2171 self.verify_configured_receipt(&receipt)?;
2172 let atif_digest = av_core::digest::sha256_hex(&bytes);
2181 let receipt_matches = matches!(
2182 &receipt.body.subject,
2183 ReceiptSubject::AtifTrajectory { trajectory_digest, .. }
2184 if *trajectory_digest == atif_digest
2185 );
2186 if !receipt_matches {
2187 tracing::warn!(
2188 session = %av_core::fsutil::basename(&path),
2189 "prior receipt does not attest the recovered ATIF trajectory; ignoring as prior-incarnation evidence"
2190 );
2191 } else {
2192 sync_existing_file_parent(&receipt_path)
2193 .await
2194 .map_err(FinalizeError::receipt_source)?;
2195 if path.with_extension("promote").exists() {
2196 recovered_session.restore_pending_receipt(receipt);
2197 } else {
2198 recovered_session.restore_receipt(receipt);
2199 }
2200 }
2201 }
2202 Err(error) if error.kind() == std::io::ErrorKind::NotFound
2203 || error.kind() == std::io::ErrorKind::NotADirectory =>
2204 {
2205 }
2207 Err(error) => {
2208 return Err(FinalizeError::Receipt {
2209 context: format!("existing receipt stat failed: {error}"),
2210 source: Some(Box::new(error)),
2211 });
2212 }
2213 }
2214 Ok(())
2215 }.await;
2216 if let Err(error) = receipt_restore {
2217 sessions.remove(&recovered_session.id);
2222 return Err(error);
2223 }
2224 Ok(AtifCandidateOutcome::Recovered)
2225 }.await;
2226 match outcome {
2227 Ok(AtifCandidateOutcome::Recovered) => recovered += 1,
2228 Ok(AtifCandidateOutcome::Skipped) => {}
2229 Err(error) => {
2230 self.metrics
2231 .counter(
2232 "av_atif_trajectory_recovery_skipped_total",
2233 "ATIF trajectories skipped during recovery due to per-session errors",
2234 )
2235 .inc();
2236 if self.warn_once(path.clone()) {
2237 tracing::warn!(
2238 %error,
2239 path = %av_core::fsutil::basename(&path),
2240 "skipping ATIF trajectory recovery due to per-session error; other sessions continue"
2241 );
2242 }
2243 }
2244 }
2245 }
2246 self.remove_acked_lifecycle_outboxes(sessions).await?;
2247 Ok(recovered)
2248 }
2249
2250 pub(crate) async fn recover_signed_journals(
2263 &self,
2264 sessions: &SessionRegistry,
2265 breaker: &av_loopdetect::BreakerConfig,
2266 ) -> Result<usize, FinalizeError> {
2267 enum SignedCandidateOutcome {
2271 Recovered,
2275 Skipped,
2279 }
2280 let mut entries = match tokio::fs::read_dir(&self.spool_dir).await {
2281 Ok(entries) => entries,
2282 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(0),
2283 Err(error) => return Err(FinalizeError::atif_source(error)),
2284 };
2285 let mut recovered = 0usize;
2286 let mut examined = 0usize;
2287 let mut dirents_seen = 0usize;
2288 while let Some(entry) = entries.next_entry().await.map_err(FinalizeError::atif_source)? {
2289 dirents_seen = dirents_seen.saturating_add(1);
2290 if dirents_seen > MAX_RECOVERY_DIRENTS_PER_TICK {
2291 bump_recovery_scan_cap(&self.metrics, "recover_signed_journals", examined, dirents_seen);
2292 break;
2293 }
2294 let metadata_path = entry.path();
2295 let Some(name) = metadata_path.file_name().and_then(std::ffi::OsStr::to_str) else {
2296 continue;
2297 };
2298 let Some(stem) = name.strip_suffix(".session.json") else {
2299 continue;
2300 };
2301 examined = examined.saturating_add(1);
2302 if examined > MAX_RECOVERY_ENTRIES_PER_TICK {
2303 bump_recovery_scan_cap(&self.metrics, "recover_signed_journals", examined, dirents_seen);
2304 break;
2305 }
2306 let outcome: Result<SignedCandidateOutcome, FinalizeError> = async {
2315 let Some(metadata) = self.read_journal_metadata(&metadata_path).await? else {
2318 return Ok(SignedCandidateOutcome::Skipped);
2319 };
2320 if metadata
2321 .get("journal_version")
2322 .and_then(serde_json::Value::as_u64)
2323 != Some(2)
2324 {
2325 self.metrics
2345 .counter(
2346 "av_journal_version_stranded_total",
2347 "Session journals skipped by recovery due to an unsupported journal_version",
2348 )
2349 .inc();
2350 if self.warn_once(metadata_path.clone()) {
2351 tracing::warn!(
2352 path = %av_core::fsutil::basename(&metadata_path),
2353 version = ?metadata.get("journal_version"),
2354 "sidecar has unsupported journal_version; skipping this session so recovery can proceed for the rest"
2355 );
2356 }
2357 return Ok(SignedCandidateOutcome::Skipped);
2358 }
2359 if metadata.get("workflow").and_then(serde_json::Value::as_str) != Some(Workflow::Signed.as_str())
2360 {
2361 return Ok(SignedCandidateOutcome::Skipped);
2362 }
2363 let session_id = metadata
2364 .get("session_id")
2365 .and_then(serde_json::Value::as_str)
2366 .ok_or_else(|| FinalizeError::atif("journal metadata has no session_id"))?;
2367 if sessions.get(session_id).is_some() {
2368 return Ok(SignedCandidateOutcome::Skipped);
2369 }
2370 let _lifecycle = self.acquire_lifecycle(session_id).await;
2375 let identity: av_events::AgentIdentity = serde_json::from_value(
2376 metadata
2377 .get("identity")
2378 .cloned()
2379 .ok_or_else(|| FinalizeError::atif("journal metadata has no identity"))?,
2380 )
2381 .map_err(FinalizeError::atif_source)?;
2382 let journal_path = self.spool_dir.join(format!("{stem}.events.ndjson"));
2383 let journal = if journal_path.exists() {
2384 read_complete_journal(&journal_path).await?
2385 } else {
2386 Vec::new()
2387 };
2388 if journal.is_empty() {
2389 if quarantine_sibling_exists(&self.spool_dir, stem).await? {
2396 let quarantine_metadata = self
2397 .spool_dir
2398 .join(format!("{stem}.session.json.corrupt-{}", av_core::new_event_uid()));
2399 match tokio::fs::rename(&metadata_path, &quarantine_metadata).await {
2400 Ok(()) => tracing::warn!(
2401 metadata = %av_core::fsutil::basename(&metadata_path),
2402 quarantine = %av_core::fsutil::basename(&quarantine_metadata),
2403 "sealed metadata sidecar quarantined alongside its torn signed journal"
2404 ),
2405 Err(error) => tracing::error!(
2406 metadata = %av_core::fsutil::basename(&metadata_path),
2407 %error,
2408 "failed to quarantine metadata sidecar; leaving in place so a future recovery can try again"
2409 ),
2410 }
2411 return Ok(SignedCandidateOutcome::Skipped);
2412 }
2413 tokio::fs::remove_file(metadata_path.clone())
2414 .await
2415 .map_err(FinalizeError::atif_source)?;
2416 return Ok(SignedCandidateOutcome::Skipped);
2417 }
2418 let session = Arc::new(Session::new(
2419 session_id.to_owned(),
2420 Workflow::Signed,
2421 identity,
2422 breaker.clone(),
2423 ));
2424 session.bind_principal(metadata.get("principal_binding").and_then(serde_json::Value::as_str));
2425 let mut next_sequence = 0u64;
2426 let mut folded = crate::worker::RecoveredTotals::default();
2427 let mut pending_responses = std::collections::HashSet::new();
2428 let domain = format!("{}:active", session.id);
2429 for (index, line) in journal.into_iter().enumerate() {
2430 let index = u64::try_from(index)
2431 .map_err(|_| FinalizeError::atif("active journal index overflow".to_owned()))?;
2432 let record: crate::worker::ActiveJournalRecord =
2433 crate::journal::open(&self.journal_key, &domain, index, line.as_bytes())
2434 .map_err(FinalizeError::atif)?;
2435 let event: av_events::OcsfEvent = serde_json::from_value(record.event.clone())
2436 .map_err(FinalizeError::atif_source)?;
2437 if event.session_uid != session.id {
2438 return Err(FinalizeError::atif(format!(
2439 "signed journal event belongs to session {:?}, expected {:?}",
2440 event.session_uid, session.id
2441 )));
2442 }
2443 if record.atif_step.is_some() || record.identity != event.ai_agent {
2444 return Err(FinalizeError::atif(
2445 "signed active record has inconsistent workflow or identity".to_owned(),
2446 ));
2447 }
2448 track_response_attempt(&mut pending_responses, record.response_attempt.as_ref())?;
2449 if event.metadata.sequence != index {
2450 return Err(FinalizeError::atif(
2451 "signed event sequence does not match active journal index".to_owned(),
2452 ));
2453 }
2454 if record.identity.version != session.identity.version
2455 || record.identity.charter != session.identity.charter
2456 || record.identity.instance_uid != session.identity.instance_uid
2457 {
2458 return Err(FinalizeError::atif(
2459 "active journal changed the session identity".to_owned(),
2460 ));
2461 }
2462 session.refresh_identity(&record.identity);
2463 next_sequence = index
2464 .checked_add(1)
2465 .ok_or_else(|| FinalizeError::atif("event sequence overflow".to_owned()))?;
2466 session
2467 .chain
2468 .lock()
2469 .append(&record.event)
2470 .map_err(FinalizeError::receipt_source)?;
2471 record.fold_into(&mut folded).map_err(FinalizeError::atif)?;
2474 if let Some(id) = record.stop_reason_id {
2475 let reason = av_events::StopReason::from_id(id);
2476 if reason != av_events::StopReason::Unknown {
2477 session.record_stop_reason(reason);
2478 }
2479 }
2480 self.ensure_active_event_published(&session.id, &event, &record.event)
2481 .await?;
2482 }
2483 folded
2484 .validate_tool_accounting()
2485 .map_err(|reason| FinalizeError::atif(format!("signed {reason}")))?;
2486 folded.store_on(&session.totals);
2487 let inconsistent_responses = !pending_responses.is_empty();
2488 session.restore_next_seq(next_sequence);
2489 let expected_subject = {
2490 let chain = session.chain.lock();
2491 ReceiptSubject::EventChain {
2492 chain_head: chain.head_hex(),
2493 event_count: chain.count(),
2494 }
2495 };
2496 let receipt_path = self.receipt_path(&session.id);
2497 match tokio::fs::metadata(&receipt_path).await {
2512 Ok(_) => {
2513 let bytes = read_capped_async(
2514 receipt_path.clone(),
2515 av_core::fsutil::MAX_RECEIPT_BYTES,
2516 )
2517 .await
2518 .map_err(|error| {
2519 FinalizeError::Receipt {
2520 context: format!("existing receipt unreadable: {error}"),
2521 source: Some(Box::new(error)),
2522 }
2523 })?;
2524 let receipt = Receipt::from_json_slice(&bytes)
2528 .map_err(FinalizeError::receipt_source)?;
2529 self.verify_configured_receipt(&receipt)?;
2530 if receipt.body.subject != expected_subject {
2531 return Err(FinalizeError::receipt(
2532 "persisted receipt does not attest the recovered signed journal".to_owned(),
2533 ));
2534 }
2535 sync_existing_file_parent(&receipt_path)
2536 .await
2537 .map_err(FinalizeError::receipt_source)?;
2538 *session.receipt.lock() = Some(receipt);
2539 }
2540 Err(error) if error.kind() == std::io::ErrorKind::NotFound
2541 || error.kind() == std::io::ErrorKind::NotADirectory =>
2542 {
2543 }
2549 Err(error) => {
2550 return Err(FinalizeError::Receipt {
2551 context: format!("existing receipt stat failed: {error}"),
2552 source: Some(Box::new(error)),
2553 });
2554 }
2555 }
2556 let unwrapped = Arc::try_unwrap(session)
2557 .map_err(|_| FinalizeError::Invariant("signed recovery retained session".to_owned()))?;
2558 unwrapped.mark_artifact_committed();
2574 let session = match sessions.try_insert_recovered(unwrapped) {
2575 Ok(inserted) => inserted,
2576 Err(_active) => {
2577 tracing::info!(session = %session_id, "signed recovery skipped: session already active");
2578 return Ok(SignedCandidateOutcome::Skipped);
2579 }
2580 };
2581 if inconsistent_responses {
2582 self.quarantined_sessions.lock().insert(session.id.clone());
2585 session.mark_capture_failed();
2586 self.metrics
2594 .counter(
2595 "av_signed_recovery_quarantined_total",
2596 "Signed sessions re-adopted as capture-failed quarantines during recovery",
2597 )
2598 .inc();
2599 tracing::warn!(
2600 session = %session.id,
2601 "signed recovery re-adopted a capture-failed session (inconsistent response \
2602 journal); it will refuse all requests and its spool files are retained as \
2603 evidence until removed by an operator"
2604 );
2605 return Ok(SignedCandidateOutcome::Recovered);
2606 }
2607 if self.quarantined_sessions.lock().contains(&session.id) {
2608 session.mark_capture_failed();
2609 return Ok(SignedCandidateOutcome::Recovered);
2610 }
2611 match self
2628 .close_session_locked(Arc::clone(&session), StopReason::SessionClosed)
2629 .await
2630 {
2631 Ok(_) => Ok(SignedCandidateOutcome::Recovered),
2632 Err(FinalizeError::CaptureIncomplete) => {
2640 Ok(SignedCandidateOutcome::Recovered)
2641 }
2642 Err(error) => {
2643 sessions.remove(&session.id);
2644 Err(error)
2645 }
2646 }
2647 }.await;
2648 match outcome {
2649 Ok(SignedCandidateOutcome::Recovered) => recovered += 1,
2650 Ok(SignedCandidateOutcome::Skipped) => {}
2651 Err(error) => {
2652 self.metrics
2653 .counter(
2654 "av_signed_recovery_skipped_total",
2655 "Signed sessions skipped during recovery due to per-session errors",
2656 )
2657 .inc();
2658 if self.warn_once(metadata_path.clone()) {
2659 tracing::warn!(
2660 %error,
2661 path = %av_core::fsutil::basename(&metadata_path),
2662 "skipping signed session recovery due to per-session error; other sessions continue"
2663 );
2664 }
2665 }
2666 }
2667 }
2668 Ok(recovered)
2669 }
2670
2671 async fn ensure_active_event_published(
2672 &self,
2673 session_id: &str,
2674 event: &av_events::OcsfEvent,
2675 value: &serde_json::Value,
2676 ) -> Result<(), FinalizeError> {
2677 let topic = event.class_name.topic();
2678 let event_uid = &event.metadata.uid;
2679 if let Some(ack) =
2680 crate::worker::read_broker_ack(&self.spool_dir, session_id, event_uid, &self.journal_key)
2681 .await
2682 .map_err(FinalizeError::bridge)?
2683 {
2684 if ack.topic != topic {
2685 return Err(FinalizeError::bridge(
2686 "broker acknowledgment topic does not match active event".to_owned(),
2687 ));
2688 }
2689 return Ok(());
2690 }
2691 let bridge = self.bridge.as_ref().map(Arc::clone).ok_or_else(|| {
2692 FinalizeError::bridge("unacknowledged active event has no configured broker".to_owned())
2693 })?;
2694 let topic = topic.to_owned();
2695 let key = event.ai_agent.instance_uid.clone();
2696 let value = value.clone();
2697 let uid = event_uid.clone();
2698 let lookup_bridge = Arc::clone(&bridge);
2699 let lookup_topic = topic.clone();
2700 let lookup_key = key.clone();
2701 let lookup_uid = uid.clone();
2702 if let Some(ack) = tokio::task::spawn_blocking(move || {
2703 lookup_bridge.find_event_by_uid(&lookup_topic, &lookup_key, &lookup_uid)
2704 })
2705 .await
2706 .map_err(FinalizeError::Task)?
2707 .map_err(FinalizeError::from)?
2708 {
2709 crate::worker::persist_broker_ack(
2710 &self.spool_dir,
2711 session_id,
2712 event_uid,
2713 &ack,
2714 &self.journal_key,
2715 )
2716 .await
2717 .map_err(FinalizeError::bridge)?;
2718 return Ok(());
2719 }
2720 let ack = tokio::task::spawn_blocking(move || bridge.publish_idempotent(&topic, &key, &value, &uid))
2721 .await
2722 .map_err(FinalizeError::Task)?
2723 .map_err(FinalizeError::from)?;
2724 crate::worker::persist_broker_ack(&self.spool_dir, session_id, event_uid, &ack, &self.journal_key)
2725 .await
2726 .map_err(FinalizeError::bridge)
2727 }
2728
2729 pub(crate) async fn consolidate_step_journals(
2730 &self,
2731 sessions: &SessionRegistry,
2732 breaker: &av_loopdetect::BreakerConfig,
2733 ) -> Result<(), FinalizeError> {
2734 let mut entries = match tokio::fs::read_dir(&self.spool_dir).await {
2735 Ok(entries) => entries,
2736 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
2737 Err(error) => return Err(FinalizeError::atif_source(error)),
2738 };
2739 let mut examined = 0usize;
2740 let mut dirents_seen = 0usize;
2741 while let Some(entry) = entries.next_entry().await.map_err(FinalizeError::atif_source)? {
2742 dirents_seen = dirents_seen.saturating_add(1);
2743 if dirents_seen > MAX_RECOVERY_DIRENTS_PER_TICK {
2744 bump_recovery_scan_cap(&self.metrics, "consolidate_step_journals", examined, dirents_seen);
2745 break;
2746 }
2747 let metadata_path = entry.path();
2748 let Some(name) = metadata_path.file_name().and_then(std::ffi::OsStr::to_str) else {
2749 continue;
2750 };
2751 let Some(stem) = name.strip_suffix(".session.json") else {
2752 continue;
2753 };
2754 examined = examined.saturating_add(1);
2755 if examined > MAX_RECOVERY_ENTRIES_PER_TICK {
2756 bump_recovery_scan_cap(&self.metrics, "consolidate_step_journals", examined, dirents_seen);
2757 break;
2758 }
2759 let mut claimed_session: Option<String> = None;
2766 let outcome: Result<(), FinalizeError> = async {
2767 let final_path = self.spool_dir.join(format!("{stem}.json"));
2768 let journal_path = self.spool_dir.join(format!("{stem}.events.ndjson"));
2769 let Some(metadata) = self.read_journal_metadata(&metadata_path).await? else {
2772 return Ok(());
2773 };
2774 if metadata
2775 .get("journal_version")
2776 .and_then(serde_json::Value::as_u64)
2777 != Some(2)
2778 {
2779 self.metrics
2785 .counter(
2786 "av_journal_version_stranded_total",
2787 "Session journals skipped by recovery due to an unsupported journal_version",
2788 )
2789 .inc();
2790 if self.warn_once(metadata_path.clone()) {
2791 tracing::warn!(
2792 path = %av_core::fsutil::basename(&metadata_path),
2793 version = ?metadata.get("journal_version"),
2794 "sidecar has unsupported journal_version; skipping this session so recovery can proceed for the rest"
2795 );
2796 }
2797 return Ok(());
2798 }
2799 if metadata.get("workflow").and_then(serde_json::Value::as_str) == Some(Workflow::Signed.as_str())
2800 {
2801 return Ok(());
2802 }
2803 let session_id = metadata
2804 .get("session_id")
2805 .and_then(serde_json::Value::as_str)
2806 .ok_or_else(|| FinalizeError::atif("journal metadata has no session_id"))?;
2807 if sessions.get(session_id).is_some() {
2808 return Ok(());
2809 }
2810 let _lifecycle = self.acquire_lifecycle(session_id).await;
2815 let identity: av_events::AgentIdentity = serde_json::from_value(
2816 metadata
2817 .get("identity")
2818 .cloned()
2819 .ok_or_else(|| FinalizeError::atif("journal metadata has no identity"))?,
2820 )
2821 .map_err(FinalizeError::atif_source)?;
2822 let placeholder = Session::new(
2839 session_id.to_owned(),
2840 Workflow::Unsigned,
2841 identity.clone(),
2842 breaker.clone(),
2843 );
2844 placeholder.bind_principal(metadata.get("principal_binding").and_then(serde_json::Value::as_str));
2845 let placeholder_claimed = placeholder.try_close();
2849 debug_assert!(placeholder_claimed, "placeholder is freshly constructed");
2850 let claimed = match sessions.try_insert_recovered(placeholder) {
2851 Ok(claimed) => claimed,
2852 Err(_active) => {
2853 return Ok(());
2858 }
2859 };
2860 claimed_session = Some(session_id.to_owned());
2861 let journal = if journal_path.exists() {
2862 read_complete_journal(&journal_path).await?
2863 } else {
2864 Vec::new()
2865 };
2866 if self.quarantined_sessions.lock().contains(session_id) {
2867 claimed.restore_journal_index(
2876 u64::try_from(journal.len())
2877 .map_err(|_| FinalizeError::atif("active journal length overflow".to_owned()))?,
2878 );
2879 claimed.restore_next_seq(
2886 u64::try_from(journal.len())
2887 .map_err(|_| FinalizeError::atif("active journal length overflow".to_owned()))?,
2888 );
2889 claimed.mark_capture_failed();
2890 claimed.mark_artifact_committed();
2896 claimed_session = None;
2899 return Ok(());
2900 }
2901 if journal.is_empty() {
2902 if quarantine_sibling_exists(&self.spool_dir, stem).await? {
2907 let quarantine_metadata = self
2908 .spool_dir
2909 .join(format!("{stem}.session.json.corrupt-{}", av_core::new_event_uid()));
2910 match tokio::fs::rename(&metadata_path, &quarantine_metadata).await {
2911 Ok(()) => tracing::warn!(
2912 metadata = %av_core::fsutil::basename(&metadata_path),
2913 quarantine = %av_core::fsutil::basename(&quarantine_metadata),
2914 "sealed metadata sidecar quarantined alongside its torn unsigned journal"
2915 ),
2916 Err(error) => tracing::error!(
2917 metadata = %av_core::fsutil::basename(&metadata_path),
2918 %error,
2919 "failed to quarantine metadata sidecar; leaving in place so a future recovery can try again"
2920 ),
2921 }
2922 return Ok(());
2923 }
2924 tokio::fs::remove_file(metadata_path.clone())
2925 .await
2926 .map_err(FinalizeError::atif_source)?;
2927 return Ok(());
2928 }
2929 let journal_len = u64::try_from(journal.len())
2930 .map_err(|_| FinalizeError::atif("active journal length overflow".to_owned()))?;
2931 let agent = av_atif::Agent {
2932 name: "agentvisor-ai-harness".into(),
2933 version: identity.version.clone(),
2934 model_name: None,
2935 tool_definitions: None,
2936 extra: Some(serde_json::json!({
2937 "charter": identity.charter,
2938 "instance_uid": identity.instance_uid,
2939 })),
2940 };
2941 let mut builder = av_atif::TrajectoryBuilder::new(agent, Some(session_id.to_owned()));
2942 let domain = format!("{session_id}:active");
2943 let mut latest_identity = identity.clone();
2944 let mut folded = crate::worker::RecoveredTotals::default();
2945 let mut stop_reason_id = None;
2946 let mut pending_responses = std::collections::HashSet::new();
2947 for (index, line) in journal.into_iter().enumerate() {
2948 let index = u64::try_from(index)
2949 .map_err(|_| FinalizeError::atif("active journal index overflow".to_owned()))?;
2950 let record: crate::worker::ActiveJournalRecord =
2951 crate::journal::open(&self.journal_key, &domain, index, line.as_bytes())
2952 .map_err(FinalizeError::atif)?;
2953 let event: av_events::OcsfEvent = serde_json::from_value(record.event.clone())
2954 .map_err(FinalizeError::atif_source)?;
2955 if event.session_uid != session_id || event.ai_agent != record.identity {
2956 return Err(FinalizeError::atif(
2957 "unsigned active record has inconsistent session or identity".to_owned(),
2958 ));
2959 }
2960 track_response_attempt(&mut pending_responses, record.response_attempt.as_ref())?;
2961 if event.metadata.sequence != index {
2962 return Err(FinalizeError::atif(
2963 "unsigned event sequence does not match active journal index".to_owned(),
2964 ));
2965 }
2966 if record.identity.version != identity.version
2967 || record.identity.charter != identity.charter
2968 || record.identity.instance_uid != identity.instance_uid
2969 {
2970 return Err(FinalizeError::atif(
2971 "active journal changed the unsigned session identity".to_owned(),
2972 ));
2973 }
2974 self.ensure_active_event_published(session_id, &event, &record.event)
2975 .await?;
2976 record.fold_into(&mut folded).map_err(FinalizeError::atif)?;
2980 let step = record.atif_step.ok_or_else(|| {
2981 FinalizeError::atif("unsigned active record has no ATIF step".to_owned())
2982 })?;
2983 builder
2984 .push_step(step)
2985 .map_err(FinalizeError::atif_source)?;
2986 latest_identity = record.identity;
2987 if record.stop_reason_id.is_some() {
2988 stop_reason_id = record.stop_reason_id;
2989 }
2990 }
2991 folded
2992 .validate_tool_accounting()
2993 .map_err(|reason| FinalizeError::atif(format!("unsigned {reason}")))?;
2994 if !pending_responses.is_empty() {
2995 claimed.refresh_identity(&latest_identity);
3008 claimed.restore_journal_index(journal_len);
3009 claimed.restore_next_seq(journal_len);
3010 folded.store_on(&claimed.totals);
3011 claimed.mark_capture_failed();
3012 claimed.mark_artifact_committed();
3016 self.quarantined_sessions.lock().insert(session_id.to_owned());
3017 claimed_session = None;
3020 return Ok(());
3021 }
3022 let mut trajectory = builder.finish();
3023 trajectory.agent.extra = Some(serde_json::json!({
3024 "charter": latest_identity.charter,
3025 "instance_uid": latest_identity.instance_uid,
3026 "ttl_remaining_s": latest_identity.ttl_remaining_s,
3027 "principal_binding": claimed.principal_binding(),
3028 }));
3029 if let Some(metrics) = trajectory.final_metrics.as_mut() {
3030 metrics.total_prompt_tokens = Some(folded.prompt_tokens);
3031 metrics.total_completion_tokens = Some(folded.completion_tokens);
3032 metrics.total_cached_tokens = Some(folded.cached_tokens);
3033 metrics.total_cost_usd =
3034 Some(folded.cost_usd_micros as f64 / av_core::units::USD_MICROS_PER_DOLLAR as f64);
3035 metrics.extra = Some(serde_json::json!({
3036 "tool_calls": folded.tool_calls,
3037 "tool_allowed": folded.tool_allowed,
3038 "tool_blocked": folded.tool_blocked,
3039 "cost_usd_micros": folded.cost_usd_micros,
3040 "stop_reason_id": stop_reason_id.map_or(0u64, u64::from),
3055 "enforcement_latched": 0u64,
3063 }));
3064 }
3065 if final_path.exists() {
3066 let existing: av_atif::Trajectory = serde_json::from_slice(
3067 &read_capped_async(final_path.clone(), av_core::fsutil::MAX_ATIF_BYTES)
3068 .await
3069 .map_err(FinalizeError::atif_source)?,
3070 )
3071 .map_err(FinalizeError::atif_source)?;
3072 let differs = {
3098 let mut lhs = trajectory.clone();
3099 let mut rhs = existing.clone();
3100 lhs.trajectory_id.clone_from(&rhs.trajectory_id);
3101 normalize_extra_ttl(&mut lhs);
3102 normalize_extra_ttl(&mut rhs);
3103 normalize_extra_enforcement(&mut lhs);
3104 normalize_extra_enforcement(&mut rhs);
3105 lhs != rhs
3106 };
3107 if differs {
3108 let write_path = final_path.clone();
3125 let archive_session = session_id.to_owned();
3126 let new_trajectory_id = trajectory.trajectory_id.clone();
3127 let archive_result = tokio::task::spawn_blocking(move || {
3128 archive_conflicting_atif(
3129 &write_path,
3130 new_trajectory_id.as_deref(),
3131 &archive_session,
3132 )
3133 })
3134 .await
3135 .map_err(FinalizeError::Task)?;
3136 if let Err(error) = archive_result {
3137 return Err(FinalizeError::Atif {
3138 context: format!(
3139 "failed to archive prior-incarnation ATIF during recovery: {error}"
3140 ),
3141 source: Some(Box::new(error)),
3142 });
3143 }
3144 let write_path = final_path.clone();
3145 tokio::task::spawn_blocking(move || av_atif::write_atomic(&trajectory, &write_path))
3146 .await
3147 .map_err(FinalizeError::Task)?
3148 .map_err(FinalizeError::atif_source)?;
3149 }
3150 } else {
3151 let write_path = final_path.clone();
3152 tokio::task::spawn_blocking(move || av_atif::write_atomic(&trajectory, &write_path))
3153 .await
3154 .map_err(FinalizeError::Task)?
3155 .map_err(FinalizeError::atif_source)?;
3156 }
3157 self.ensure_atif_provenance(&final_path, session_id).await?;
3158 self.remove_step_journal(session_id).await?;
3159 Ok(())
3160 }.await;
3161 if let Some(id) = claimed_session {
3165 sessions.remove(&id);
3166 }
3167 if let Err(error) = outcome {
3168 self.metrics
3169 .counter(
3170 "av_unsigned_recovery_skipped_total",
3171 "Unsigned step-journal consolidations skipped during recovery due to per-session errors",
3174 )
3175 .inc();
3176 if self.warn_once(metadata_path.clone()) {
3177 tracing::warn!(
3178 %error,
3179 path = %av_core::fsutil::basename(&metadata_path),
3180 "skipping unsigned session consolidation due to per-session error; other sessions continue"
3181 );
3182 }
3183 }
3184 }
3185 Ok(())
3186 }
3187
3188 async fn rebuild_unsigned_trajectory(
3197 &self,
3198 session: &Session,
3199 ) -> Result<Option<av_atif::Trajectory>, FinalizeError> {
3200 let digest = av_core::digest::sha256_hex(session.id.as_bytes());
3201 let stem = digest.get(..32).unwrap_or(&digest);
3202 let journal_path = self.spool_dir.join(format!("{stem}.events.ndjson"));
3203 if !journal_path.exists() {
3204 return Ok(None);
3205 }
3206 let journal = read_complete_journal(&journal_path).await?;
3207 if journal.is_empty() {
3208 return Ok(None);
3209 }
3210 let identity = session.current_identity();
3211 let agent = av_atif::Agent {
3212 name: "agentvisor-ai-harness".into(),
3213 version: identity.version.clone(),
3214 model_name: None,
3215 tool_definitions: None,
3216 extra: None,
3217 };
3218 let mut builder = av_atif::TrajectoryBuilder::new(agent, Some(session.id.clone()));
3219 let domain = format!("{}:active", session.id);
3220 for (index, line) in journal.into_iter().enumerate() {
3221 let index = u64::try_from(index)
3222 .map_err(|_| FinalizeError::atif("active journal index overflow".to_owned()))?;
3223 let record: crate::worker::ActiveJournalRecord =
3224 crate::journal::open(&self.journal_key, &domain, index, line.as_bytes())
3225 .map_err(FinalizeError::atif)?;
3226 let step = record
3227 .atif_step
3228 .ok_or_else(|| FinalizeError::atif("unsigned active record has no ATIF step".to_owned()))?;
3229 builder.push_step(step).map_err(FinalizeError::atif_source)?;
3230 }
3231 Ok(Some(builder.finish()))
3232 }
3233
3234 async fn remove_step_journal(&self, session_id: &str) -> Result<(), FinalizeError> {
3235 let digest = av_core::digest::sha256_hex(session_id.as_bytes());
3236 let stem = digest.get(..32).unwrap_or(&digest).to_owned();
3237 let spool_dir = self.spool_dir.clone();
3238 tokio::task::spawn_blocking(move || -> Result<(), FinalizeError> {
3239 let mut spool_changed = false;
3240 let mut unlink_error: Option<FinalizeError> = None;
3241 for suffix in ["events.ndjson", "steps.ndjson", "acks.ndjson", "session.json"] {
3252 let path = spool_dir.join(format!("{stem}.{suffix}"));
3253 match std::fs::remove_file(&path) {
3254 Ok(()) => spool_changed = true,
3255 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
3256 Err(error) => {
3257 unlink_error = Some(FinalizeError::atif_source(error));
3258 break;
3272 }
3273 }
3274 }
3275 if spool_changed {
3276 if let Err(error) = av_core::fsutil::sync_directory(&spool_dir) {
3277 if unlink_error.is_none() {
3281 unlink_error = Some(FinalizeError::atif_source(error));
3282 } else {
3283 tracing::warn!(
3284 %error,
3285 "spool sync after partial step-journal removal failed; leaving as-is (retry on next tick)"
3286 );
3287 }
3288 }
3289 }
3290 if let Some(error) = unlink_error {
3291 return Err(error);
3292 }
3293 let ack_parent = spool_dir.join("broker-acks");
3294 let ack_path = ack_parent.join(&stem);
3295 let mut ack_error: Option<FinalizeError> = None;
3296 let mut ack_attempted = false;
3297 match std::fs::remove_dir_all(&ack_path) {
3298 Ok(()) => ack_attempted = true,
3299 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
3300 Err(error) => {
3301 ack_attempted = true;
3321 ack_error = Some(FinalizeError::atif_source(error));
3322 }
3323 }
3324 if ack_attempted {
3325 if let Err(error) = av_core::fsutil::sync_directory(&ack_parent) {
3326 if ack_error.is_none() {
3327 ack_error = Some(FinalizeError::atif_source(error));
3328 } else {
3329 tracing::warn!(
3330 %error,
3331 "broker-acks sync after partial removal failed; leaving as-is (retry on next tick)"
3332 );
3333 }
3334 }
3335 }
3336 if let Some(error) = ack_error {
3337 return Err(error);
3338 }
3339 Ok(())
3340 })
3341 .await
3342 .map_err(FinalizeError::Task)?
3343 }
3344
3345 async fn remove_tool_executions(&self, session_id: &str) -> Result<(), FinalizeError> {
3363 let spool_dir = self.spool_dir.clone();
3364 let session_id = session_id.to_owned();
3365 let control_key = self.journal_key;
3366 tokio::task::spawn_blocking(move || -> Result<(), FinalizeError> {
3367 let directory = spool_dir.join(crate::spool::TOOL_EXECUTIONS);
3368 let entries = match std::fs::read_dir(&directory) {
3369 Ok(entries) => entries,
3370 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
3371 Err(error) => return Err(FinalizeError::atif_source(error)),
3372 };
3373 let mut removed_any = false;
3374 for entry in entries {
3375 let entry = entry.map_err(FinalizeError::atif_source)?;
3376 let path = entry.path();
3377 let Some(name) = path.file_name().and_then(std::ffi::OsStr::to_str) else {
3378 continue;
3379 };
3380 let Some(key) = name.strip_suffix(crate::spool::TOOL_INTENT_SUFFIX) else {
3381 continue;
3382 };
3383 let intent_bytes =
3384 match av_core::fsutil::read_capped(&path, av_core::fsutil::MAX_CONTROL_BYTES) {
3385 Ok(bytes) => bytes,
3386 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
3391 Err(error)
3398 if error.kind() == std::io::ErrorKind::InvalidData
3399 || error.kind() == std::io::ErrorKind::InvalidInput =>
3400 {
3401 tracing::warn!(
3402 file = %av_core::fsutil::basename(&path),
3403 %error,
3404 "skipping oversize or non-regular tool-intent file during close cleanup"
3405 );
3406 continue;
3407 }
3408 Err(error) => return Err(FinalizeError::atif_source(error)),
3409 };
3410 let intent_session_id = crate::journal::open::<serde_json::Value>(
3411 &control_key,
3412 &format!("{}:{key}", crate::journal::TOOL_INTENT_DOMAIN),
3413 0,
3414 &intent_bytes,
3415 )
3416 .ok()
3417 .and_then(|value| {
3418 value
3419 .get("session_id")
3420 .and_then(serde_json::Value::as_str)
3421 .map(str::to_owned)
3422 });
3423 if intent_session_id.as_deref() != Some(session_id.as_str()) {
3424 continue;
3425 }
3426 for suffix in [
3445 crate::spool::TOOL_AUDITED_SUFFIX,
3446 crate::spool::TOOL_OUTCOME_SUFFIX,
3447 crate::spool::TOOL_INTENT_SUFFIX,
3448 ".intent.torn",
3449 ] {
3450 let file = directory.join(format!("{key}{suffix}"));
3451 match std::fs::remove_file(&file) {
3452 Ok(()) => removed_any = true,
3453 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
3454 Err(error) => return Err(FinalizeError::atif_source(error)),
3455 }
3456 }
3457 }
3458 if removed_any {
3459 av_core::fsutil::sync_directory(&directory).map_err(FinalizeError::atif_source)?;
3460 }
3461 Ok(())
3462 })
3463 .await
3464 .map_err(FinalizeError::Task)?
3465 }
3466
3467 async fn quarantine_tool_executions(&self, session_id: &str) {
3473 let spool_dir = self.spool_dir.clone();
3474 let session_id = session_id.to_owned();
3475 let control_key = self.journal_key;
3476 let outcome = tokio::task::spawn_blocking(move || {
3477 let directory = spool_dir.join(crate::spool::TOOL_EXECUTIONS);
3478 let entries = match std::fs::read_dir(&directory) {
3479 Ok(entries) => entries,
3480 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
3481 Err(error) => return Err(error.to_string()),
3482 };
3483 for entry in entries {
3484 let Ok(entry) = entry else { continue };
3485 let path = entry.path();
3486 let Some(name) = path.file_name().and_then(std::ffi::OsStr::to_str) else {
3487 continue;
3488 };
3489 let Some(key) = name.strip_suffix(crate::spool::TOOL_INTENT_SUFFIX) else {
3490 continue;
3491 };
3492 let intent_bytes =
3499 match av_core::fsutil::read_capped(&path, av_core::fsutil::MAX_CONTROL_BYTES) {
3500 Ok(bytes) => bytes,
3501 Err(_) => continue,
3502 };
3503 let intent_session_id = crate::journal::open::<serde_json::Value>(
3504 &control_key,
3505 &format!("{}:{key}", crate::journal::TOOL_INTENT_DOMAIN),
3506 0,
3507 &intent_bytes,
3508 )
3509 .ok()
3510 .and_then(|value| {
3511 value
3512 .get("session_id")
3513 .and_then(serde_json::Value::as_str)
3514 .map(str::to_owned)
3515 });
3516 if intent_session_id.as_deref() != Some(session_id.as_str()) {
3517 continue;
3518 }
3519 if !crate::routes::tool_completion_is_authenticated(&directory, key, &control_key) {
3520 continue;
3524 }
3525 let uid = av_core::new_event_uid();
3526 for suffix in [
3530 crate::spool::TOOL_AUDITED_SUFFIX,
3531 crate::spool::TOOL_OUTCOME_SUFFIX,
3532 crate::spool::TOOL_INTENT_SUFFIX,
3533 ] {
3534 let file = directory.join(format!("{key}{suffix}"));
3535 let quarantine = directory.join(format!("{key}{suffix}.capturefailed-{uid}"));
3536 match std::fs::rename(&file, &quarantine) {
3537 Ok(()) | Err(_) => {}
3538 }
3539 }
3540 tracing::warn!(
3541 session = %session_id,
3542 key = %key,
3543 "tool-execution triple quarantined as capture-failed incident evidence"
3544 );
3545 }
3546 let _ = av_core::fsutil::sync_directory(&directory);
3547 Ok(())
3548 })
3549 .await;
3550 match outcome {
3551 Ok(Ok(())) => {}
3552 Ok(Err(error)) => {
3553 tracing::warn!(%error, "failed to quarantine capture-failed tool executions")
3556 }
3557 Err(error) => {
3558 tracing::warn!(
3559 error = &error as &dyn std::error::Error,
3560 "failed to quarantine capture-failed tool executions"
3561 )
3562 }
3563 }
3564 }
3565
3566 pub async fn retry_marked_promotions(&self, sessions: &SessionRegistry) -> Result<usize, FinalizeError> {
3568 let mut entries = match tokio::fs::read_dir(&self.spool_dir).await {
3569 Ok(entries) => entries,
3570 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(0),
3571 Err(error) => return Err(FinalizeError::atif_source(error)),
3572 };
3573 let mut promoted = 0usize;
3574 let mut examined = 0usize;
3575 let mut dirents_seen = 0usize;
3576 while let Some(entry) = entries.next_entry().await.map_err(FinalizeError::atif_source)? {
3577 dirents_seen = dirents_seen.saturating_add(1);
3578 if dirents_seen > MAX_RECOVERY_DIRENTS_PER_TICK {
3579 bump_recovery_scan_cap(&self.metrics, "retry_marked_promotions", examined, dirents_seen);
3580 break;
3581 }
3582 let path = entry.path();
3583 if path.extension().and_then(std::ffi::OsStr::to_str) != Some("promote") {
3584 continue;
3585 }
3586 examined = examined.saturating_add(1);
3587 if examined > MAX_RECOVERY_ENTRIES_PER_TICK {
3588 bump_recovery_scan_cap(&self.metrics, "retry_marked_promotions", examined, dirents_seen);
3589 break;
3590 }
3591 let sealed = match read_capped_async(path.clone(), av_core::fsutil::MAX_CONTROL_BYTES).await {
3599 Ok(bytes) => bytes,
3600 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
3604 Err(error) => {
3605 tracing::warn!(
3606 %error,
3607 path = %av_core::fsutil::basename(&path),
3608 "skipping unreadable promotion marker"
3609 );
3610 continue;
3611 }
3612 };
3613 let marker: PromotionMarker =
3614 match crate::journal::open(&self.journal_key, "promotion-marker", 0, &sealed) {
3615 Ok(m) => m,
3616 Err(error) => {
3617 tracing::warn!(
3618 %error,
3619 path = %av_core::fsutil::basename(&path),
3620 "skipping unauthenticated promotion marker"
3621 );
3622 continue;
3623 }
3624 };
3625 let artifact = path.with_extension("json");
3636 if !artifact.exists() && !path.with_extension("atif-auth").exists() {
3637 let quarantined = path.with_extension("promote.orphaned");
3638 match std::fs::rename(&path, &quarantined) {
3639 Ok(()) => tracing::error!(
3640 session = %marker.session_id,
3641 marker = %av_core::fsutil::basename(&path),
3642 "promotion marker orphaned: artifact and provenance are gone; \
3643 quarantined marker — the promised receipt can never be issued"
3644 ),
3645 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
3646 Err(error) => tracing::warn!(
3647 %error,
3648 marker = %av_core::fsutil::basename(&path),
3649 "failed to quarantine orphaned promotion marker"
3650 ),
3651 }
3652 continue;
3653 }
3654 let Some(session) = sessions.get(&marker.session_id) else {
3655 continue;
3656 };
3657 if !session.is_closed() {
3662 tracing::info!(
3663 session = %marker.session_id,
3664 "promotion retry skipped: session is currently active",
3665 );
3666 continue;
3667 }
3668 match self.promote(session).await {
3669 Ok(_) => promoted += 1,
3670 Err(error) => {
3671 tracing::warn!(
3672 error = &error as &dyn std::error::Error,
3673 path = %av_core::fsutil::basename(&path),
3674 "promotion retry failed"
3675 );
3676 }
3677 }
3678 }
3679 Ok(promoted)
3680 }
3681
3682 fn receipt_path(&self, session_id: &str) -> PathBuf {
3683 self.spool_dir.join("receipts").join(format!(
3684 "{}.json",
3685 &av_core::digest::sha256_hex(session_id.as_bytes())[..32]
3686 ))
3687 }
3688
3689 async fn ensure_atif_provenance(
3690 &self,
3691 path: &std::path::Path,
3692 session_id: &str,
3693 ) -> Result<AtifProvenance, FinalizeError> {
3694 let bytes = read_capped_async(path.to_path_buf(), av_core::fsutil::MAX_ATIF_BYTES)
3697 .await
3698 .map_err(FinalizeError::atif_source)?;
3699 self.ensure_atif_provenance_from_bytes(path, session_id, &bytes)
3700 .await
3701 }
3702
3703 async fn ensure_atif_provenance_from_bytes(
3706 &self,
3707 path: &std::path::Path,
3708 session_id: &str,
3709 bytes: &[u8],
3710 ) -> Result<AtifProvenance, FinalizeError> {
3711 let expected = AtifProvenance {
3712 session_id: session_id.to_owned(),
3713 digest: av_core::digest::sha256_hex(bytes),
3714 };
3715 let provenance_path = path.with_extension("atif-auth");
3716 if provenance_path.exists() {
3717 let sealed = read_capped_async(provenance_path.clone(), av_core::fsutil::MAX_CONTROL_BYTES)
3718 .await
3719 .map_err(FinalizeError::atif_source)?;
3720 let actual: AtifProvenance =
3721 crate::journal::open(&self.journal_key, "atif-provenance", 0, &sealed)
3722 .map_err(FinalizeError::atif)?;
3723 if actual.session_id != expected.session_id || actual.digest != expected.digest {
3724 return Err(FinalizeError::atif(
3725 "ATIF provenance does not match artifact bytes and session".to_owned(),
3726 ));
3727 }
3728 sync_existing_file_parent(&provenance_path)
3732 .await
3733 .map_err(FinalizeError::atif_source)?;
3734 return Ok(actual);
3735 }
3736 let sealed = crate::journal::seal(&self.journal_key, "atif-provenance", 0, &expected)
3737 .map_err(FinalizeError::atif)?;
3738 persist_marker(&provenance_path, &sealed).await?;
3739 Ok(expected)
3740 }
3741
3742 async fn persist_close_complete_marker(&self, session: &Session) {
3747 let atif_path: Option<std::path::PathBuf> = { session.atif_path.lock().clone() };
3748 let Some(atif_path) = atif_path else {
3749 return;
3750 };
3751 let digest = match self.ensure_atif_provenance(&atif_path, &session.id).await {
3752 Ok(provenance) => provenance.digest,
3753 Err(error) => {
3754 tracing::warn!(
3755 session = %session.id,
3756 %error,
3757 "close-complete marker skipped: ATIF provenance unavailable; \
3758 the next restart may re-emit one SESSION_CLOSE for this session"
3759 );
3760 return;
3761 }
3762 };
3763 let marker = CloseCompleteMarker {
3764 session_id: session.id.clone(),
3765 digest,
3766 };
3767 let sealed = match crate::journal::seal(&self.journal_key, CLOSE_COMPLETE_DOMAIN, 0, &marker) {
3768 Ok(sealed) => sealed,
3769 Err(error) => {
3770 tracing::warn!(session = %session.id, %error, "close-complete marker seal failed");
3771 return;
3772 }
3773 };
3774 if let Err(error) = persist_marker(&close_complete_marker_path(&atif_path), &sealed).await {
3775 tracing::warn!(
3776 session = %session.id,
3777 %error,
3778 "close-complete marker persist failed; \
3779 the next restart may re-emit one SESSION_CLOSE for this session"
3780 );
3781 }
3782 }
3783
3784 async fn read_journal_metadata(
3791 &self,
3792 path: &std::path::Path,
3793 ) -> Result<Option<serde_json::Value>, FinalizeError> {
3794 let bytes = match read_capped_async(path.to_path_buf(), av_core::fsutil::MAX_CONTROL_BYTES).await {
3798 Ok(bytes) => bytes,
3799 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
3800 Err(error) => return Err(FinalizeError::atif_source(error)),
3801 };
3802 crate::journal::open(&self.journal_key, "metadata", 0, &bytes)
3803 .map(Some)
3804 .map_err(FinalizeError::atif)
3805 }
3806
3807 fn verify_configured_receipt(&self, receipt: &Receipt) -> Result<(), FinalizeError> {
3808 if receipt.body.key_id != self.signer.key_id() {
3809 return Err(FinalizeError::receipt(format!(
3810 "receipt key {:?} does not match configured key {:?}",
3811 receipt.body.key_id,
3812 self.signer.key_id()
3813 )));
3814 }
3815 let mut keyring = av_receipts::Keyring::new();
3816 keyring
3817 .add_signer(self.signer.as_ref())
3818 .map_err(FinalizeError::receipt_source)?;
3819 receipt.verify(&keyring).map_err(FinalizeError::receipt_source)
3820 }
3821
3822 async fn persist_receipt(&self, session_id: &str, receipt: &Receipt) -> Result<(), FinalizeError> {
3823 let path = self.receipt_path(session_id);
3824 let bytes = serde_json::to_vec_pretty(receipt).map_err(FinalizeError::receipt_source)?;
3825 let receipt_id = receipt.body.receipt_id.clone();
3826 let session = session_id.to_owned();
3827 tokio::task::spawn_blocking(move || {
3828 archive_conflicting_receipt(&path, &receipt_id, &session)?;
3829 av_core::fsutil::write_atomic(&path, &bytes)
3830 })
3831 .await
3832 .map_err(FinalizeError::Task)?
3833 .map_err(FinalizeError::receipt_source)
3834 }
3835
3836 async fn has_quarantine_notice(&self, session: &Session) -> Result<bool, FinalizeError> {
3837 let notice = quarantine_notice_path(&self.spool_dir, &session.id);
3840 match read_capped_async(notice, av_core::fsutil::MAX_CONTROL_BYTES).await {
3841 Ok(sealed) => {
3842 let recorded: LifecycleOutbox = crate::journal::open(
3843 &self.journal_key,
3844 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
3845 0,
3846 &sealed,
3847 )
3848 .map_err(FinalizeError::bridge)?;
3849 if recorded.session_id != session.id
3850 || recorded.kind != crate::journal::SESSION_QUARANTINE_OUTBOX_KIND
3851 || recorded.ack.is_none()
3852 {
3853 return Err(FinalizeError::bridge(
3854 "invalid acknowledged quarantine notification".to_owned(),
3855 ));
3856 }
3857 Ok(true)
3858 }
3859 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
3860 Err(error) => Err(FinalizeError::bridge_source(error)),
3861 }
3862 }
3863
3864 async fn emit_quarantine_event(&self, session: &Session) -> Result<(), FinalizeError> {
3865 if self.has_quarantine_notice(session).await? {
3866 return Ok(());
3867 }
3868 self.emit_bridge_event(
3869 session,
3870 av_events::EventClass::Session,
3871 serde_json::json!({
3872 "action": "quarantined",
3873 "workflow": session.workflow.as_str(),
3874 "reason": "incomplete_capture",
3875 "evidence_complete": false,
3876 }),
3877 crate::journal::SESSION_QUARANTINE_OUTBOX_KIND,
3878 )
3879 .await
3880 }
3881
3882 async fn emit_receipt_event(&self, session: &Session, receipt: &Receipt) -> Result<(), FinalizeError> {
3883 self.emit_bridge_event(
3884 session,
3885 av_events::EventClass::Receipt,
3886 serde_json::json!({
3887 "receipt_id": receipt.body.receipt_id,
3888 "key_id": receipt.body.key_id,
3889 "subject": receipt.body.subject,
3890 "receipt": receipt,
3891 }),
3892 crate::journal::RECEIPT_OUTBOX_KIND,
3893 )
3894 .await
3895 }
3896
3897 async fn emit_bridge_event(
3898 &self,
3899 session: &Session,
3900 class: av_events::EventClass,
3901 payload: serde_json::Value,
3902 kind: &str,
3903 ) -> Result<(), FinalizeError> {
3904 let Some(bridge) = self.bridge.as_ref().map(Arc::clone) else {
3905 return Ok(());
3906 };
3907 let path = self.lifecycle_outbox_path(&session.id, kind);
3908 let current_instance_uid = session.current_identity().instance_uid;
3909 let mut outbox = if path.exists() {
3910 let sealed = read_capped_async(path.clone(), av_core::fsutil::MAX_CONTROL_BYTES)
3911 .await
3912 .map_err(FinalizeError::bridge_source)?;
3913 let outbox: LifecycleOutbox = crate::journal::open(
3914 &self.journal_key,
3915 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
3916 0,
3917 &sealed,
3918 )
3919 .map_err(FinalizeError::bridge)?;
3920 if outbox.session_id != session.id || outbox.kind != kind {
3921 return Err(FinalizeError::bridge(
3922 "lifecycle outbox does not match its session and kind".to_owned(),
3923 ));
3924 }
3925 let stale_incarnation = kind != crate::journal::SESSION_QUARANTINE_OUTBOX_KIND
3958 && (outbox.key != current_instance_uid
3959 || outbox
3960 .session_scope
3961 .as_ref()
3962 .is_some_and(|scope| scope != &session.session_scope));
3963 if stale_incarnation {
3964 remove_outbox(&path).await?;
3965 None
3971 } else {
3972 sync_existing_file_parent(&path)
3975 .await
3976 .map_err(FinalizeError::bridge_source)?;
3977 if let Some(persisted_seq) = outbox
3983 .value
3984 .get("metadata")
3985 .and_then(|metadata| metadata.get("sequence"))
3986 .and_then(serde_json::Value::as_u64)
3987 {
3988 if session.peek_seq() <= persisted_seq {
3989 session.advance_seq_past(persisted_seq);
3990 }
3991 }
3992 Some(outbox)
3993 }
3994 } else {
3995 None
3996 };
3997 if outbox.is_none() {
4000 if kind == crate::journal::SESSION_QUARANTINE_OUTBOX_KIND
4004 && self.has_quarantine_notice(session).await?
4005 {
4006 return Ok(());
4007 }
4008 let event_seq = session.peek_seq();
4013 let event = av_events::OcsfEventBuilder::new(
4014 class,
4015 session.id.clone(),
4016 session.current_identity(),
4017 event_seq,
4018 )
4019 .payload(payload)
4020 .build()
4021 .map_err(FinalizeError::bridge_source)?;
4022 let fresh = LifecycleOutbox {
4023 schema_version: LIFECYCLE_OUTBOX_SCHEMA_V1,
4024 session_id: session.id.clone(),
4025 kind: kind.to_owned(),
4026 topic: class.topic().to_owned(),
4027 key: current_instance_uid.clone(),
4028 value: serde_json::to_value(event).map_err(FinalizeError::bridge_source)?,
4029 ack: None,
4030 session_scope: Some(session.session_scope.clone()),
4031 };
4032 persist_outbox(&path, &fresh, &self.journal_key).await?;
4033 session.advance_seq_past(event_seq);
4034 outbox = Some(fresh);
4035 }
4036 let Some(mut outbox) = outbox else {
4042 return Err(FinalizeError::bridge(
4043 "internal invariant: emit_bridge_event fresh-emit branch left outbox unset".to_owned(),
4044 ));
4045 };
4046 if outbox.ack.is_some() {
4047 if kind == crate::journal::SESSION_QUARANTINE_OUTBOX_KIND {
4048 archive_quarantine_notice(&self.spool_dir, &outbox, &self.journal_key).await?;
4049 }
4050 return Ok(());
4051 }
4052 let topic = outbox.topic.clone();
4053 let key = outbox.key.clone();
4054 let value = outbox.value.clone();
4055 let event_uid = lifecycle_event_uid(&value)?;
4056 let ack = match resolve_lifecycle_ack(bridge, topic, key, value, event_uid).await {
4057 Ok(ack) => ack,
4058 Err(error) => {
4059 self.metrics
4060 .counter(
4061 "av_lifecycle_event_errors_total",
4062 "Lifecycle events not published",
4063 )
4064 .inc();
4065 return Err(error);
4066 }
4067 };
4068 outbox.ack = Some(ack);
4069 persist_outbox(&path, &outbox, &self.journal_key).await?;
4070 if kind == crate::journal::SESSION_QUARANTINE_OUTBOX_KIND {
4071 archive_quarantine_notice(&self.spool_dir, &outbox, &self.journal_key).await?;
4072 }
4073 Ok(())
4074 }
4075
4076 async fn replay_unacked_lifecycle_outbox(
4083 &self,
4084 session: &Session,
4085 kind: &str,
4086 ) -> Result<(), FinalizeError> {
4087 let Some(bridge) = self.bridge.as_ref().map(Arc::clone) else {
4088 return Ok(());
4089 };
4090 let path = self.lifecycle_outbox_path(&session.id, kind);
4091 let sealed = match read_capped_async(path.clone(), av_core::fsutil::MAX_CONTROL_BYTES).await {
4092 Ok(sealed) => sealed,
4093 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
4094 Err(error) => return Err(FinalizeError::bridge_source(error)),
4095 };
4096 let mut outbox: LifecycleOutbox = crate::journal::open(
4097 &self.journal_key,
4098 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
4099 0,
4100 &sealed,
4101 )
4102 .map_err(FinalizeError::bridge)?;
4103 if outbox.session_id != session.id || outbox.kind != kind {
4104 return Err(FinalizeError::bridge(
4105 "lifecycle outbox does not match its session and kind".to_owned(),
4106 ));
4107 }
4108 if outbox.ack.is_some() {
4109 return Ok(());
4110 }
4111 sync_existing_file_parent(&path)
4112 .await
4113 .map_err(FinalizeError::bridge_source)?;
4114 let event_uid = lifecycle_event_uid(&outbox.value)?;
4115 let ack = resolve_lifecycle_ack(
4116 bridge,
4117 outbox.topic.clone(),
4118 outbox.key.clone(),
4119 outbox.value.clone(),
4120 event_uid,
4121 )
4122 .await?;
4123 outbox.ack = Some(ack);
4124 persist_outbox(&path, &outbox, &self.journal_key).await
4125 }
4126
4127 #[cfg(test)]
4131 async fn replay_lifecycle_outboxes(&self) -> Result<usize, FinalizeError> {
4132 let Some(bridge) = self.bridge.as_ref().map(Arc::clone) else {
4133 return Ok(0);
4134 };
4135 replay_lifecycle_outboxes_in(&self.spool_dir, &self.journal_key, bridge, &self.metrics).await
4136 }
4137
4138 pub(crate) async fn complete_pending_closes(
4162 &self,
4163 sessions: &SessionRegistry,
4164 ) -> Result<usize, FinalizeError> {
4165 let pending = sessions.pending_close_sessions();
4166 let mut completed = 0usize;
4167 for session in pending {
4168 let _lifecycle = self.acquire_lifecycle(&session.id).await;
4190 if !session.artifact_committed_flag() || session.close_complete_flag() {
4195 continue;
4196 }
4197 let workflow = session.workflow.as_str();
4198 let outcome: Result<(), FinalizeError> = async {
4199 if session.workflow == Workflow::Signed {
4208 let receipt_outbox =
4209 self.lifecycle_outbox_path(&session.id, crate::journal::RECEIPT_OUTBOX_KIND);
4210 if receipt_outbox.exists() {
4211 self.replay_unacked_lifecycle_outbox(&session, crate::journal::RECEIPT_OUTBOX_KIND)
4212 .await?;
4213 } else if !self
4214 .lifecycle_outbox_path(&session.id, crate::journal::SESSION_CLOSE_OUTBOX_KIND)
4215 .exists()
4216 {
4217 let receipt = session.receipt.lock().clone();
4226 if let Some(receipt) = receipt {
4227 self.emit_receipt_event(&session, &receipt).await?;
4228 } else {
4229 tracing::warn!(
4230 session = %session.id,
4231 "pending-close sweep found no receipt outbox and no in-memory receipt; \
4232 the receipt lifecycle event cannot be reconstructed (the signed receipt \
4233 file itself is on disk)"
4234 );
4235 }
4236 }
4237 }
4238 self.emit_bridge_event(
4239 &session,
4240 av_events::EventClass::Session,
4241 serde_json::json!({"action": "closed", "workflow": workflow}),
4242 crate::journal::SESSION_CLOSE_OUTBOX_KIND,
4243 )
4244 .await?;
4245 self.remove_step_journal(&session.id).await?;
4246 self.remove_tool_executions(&session.id).await?;
4247 self.remove_lifecycle_outbox(&session.id, crate::journal::RECEIPT_OUTBOX_KIND)
4248 .await?;
4249 self.remove_lifecycle_outbox(&session.id, crate::journal::SESSION_CLOSE_OUTBOX_KIND)
4250 .await?;
4251 self.metrics
4252 .counter("av_sessions_finalized_total", "Sessions finalized")
4253 .inc();
4254 if session.workflow == Workflow::Unsigned {
4257 self.persist_close_complete_marker(&session).await;
4258 }
4259 self.clear_budget_state(&session.id);
4268 session.mark_close_complete();
4269 session.drain_trajectory_builder();
4277 session.loop_state.release_embedding();
4280 Ok(())
4281 }
4282 .await;
4283 match outcome {
4284 Ok(()) => completed = completed.saturating_add(1),
4285 Err(error) => {
4286 self.metrics
4287 .counter(
4288 "av_pending_close_completion_failed_total",
4289 "Pending-close completions that failed to finish their tail",
4290 )
4291 .inc();
4292 let key = self.spool_dir.join(format!("pending-close::{}", session.id));
4293 if self.warn_once(key) {
4294 tracing::warn!(
4295 session = %session.id,
4296 %error,
4297 "pending-close completion failed; will retry next tick",
4298 );
4299 }
4300 }
4301 }
4302 }
4303 Ok(completed)
4304 }
4305
4306 async fn remove_acked_lifecycle_outboxes(&self, sessions: &SessionRegistry) -> Result<(), FinalizeError> {
4307 let directory = self.spool_dir.join(crate::spool::OUTBOX);
4308 let mut entries = match tokio::fs::read_dir(&directory).await {
4309 Ok(entries) => entries,
4310 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
4311 self.metrics
4316 .gauge("av_lifecycle_outbox_pending", LIFECYCLE_OUTBOX_PENDING_HELP)
4317 .set(0);
4318 return Ok(());
4319 }
4320 Err(error) => return Err(FinalizeError::bridge_source(error)),
4321 };
4322 let mut examined = 0usize;
4323 let mut dirents_seen = 0usize;
4324 let mut pending_count: u64 = 0;
4325 while let Some(entry) = entries.next_entry().await.map_err(FinalizeError::bridge_source)? {
4326 dirents_seen = dirents_seen.saturating_add(1);
4327 if dirents_seen > MAX_RECOVERY_DIRENTS_PER_TICK {
4328 bump_recovery_scan_cap(&self.metrics, "remove_acked_outboxes", examined, dirents_seen);
4329 break;
4330 }
4331 let path = entry.path();
4332 if path.extension().and_then(std::ffi::OsStr::to_str) != Some("json") {
4333 continue;
4334 }
4335 examined = examined.saturating_add(1);
4336 if examined > MAX_RECOVERY_ENTRIES_PER_TICK {
4337 bump_recovery_scan_cap(&self.metrics, "remove_acked_outboxes", examined, dirents_seen);
4338 break;
4339 }
4340 let sealed = match read_capped_async(path.clone(), av_core::fsutil::MAX_CONTROL_BYTES).await {
4341 Ok(bytes) => bytes,
4342 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
4345 Err(error) => {
4346 tracing::warn!(%error, path = %av_core::fsutil::basename(&path), "skipping unreadable outbox during acked-outbox GC");
4347 continue;
4348 }
4349 };
4350 let outbox: LifecycleOutbox = match crate::journal::open(
4356 &self.journal_key,
4357 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
4358 0,
4359 &sealed,
4360 ) {
4361 Ok(outbox) => outbox,
4362 Err(error) => {
4363 tracing::warn!(%error, path = %av_core::fsutil::basename(&path), "skipping malformed outbox during acked-outbox GC");
4364 continue;
4365 }
4366 };
4367 if path != self.lifecycle_outbox_path(&outbox.session_id, &outbox.kind) {
4368 tracing::warn!(
4369 path = %av_core::fsutil::basename(&path),
4370 "skipping outbox whose filename does not match its authenticated session_id/kind"
4371 );
4372 continue;
4373 }
4374 if outbox.ack.is_some() {
4375 if outbox.kind == crate::journal::SESSION_QUARANTINE_OUTBOX_KIND {
4378 archive_quarantine_notice(&self.spool_dir, &outbox, &self.journal_key).await?;
4379 continue;
4380 }
4381 if sessions.get(&outbox.session_id).is_some_and(|session| {
4395 !session.close_complete_flag()
4396 && !session.capture_failed()
4397 && !session.is_empty_unsigned_quarantine()
4398 }) {
4399 continue;
4400 }
4401 if outbox.kind == crate::journal::RECEIPT_OUTBOX_KIND {
4411 let session_hash = &av_core::digest::sha256_hex(outbox.session_id.as_bytes())[..32];
4412 if self.spool_dir.join(format!("{session_hash}.promote")).exists() {
4413 continue;
4414 }
4415 }
4416 remove_outbox(&path).await?;
4417 } else {
4418 pending_count = pending_count.saturating_add(1);
4422 }
4423 }
4424 self.metrics
4425 .gauge("av_lifecycle_outbox_pending", LIFECYCLE_OUTBOX_PENDING_HELP)
4426 .set(pending_count);
4427 Ok(())
4428 }
4429
4430 fn lifecycle_outbox_path(&self, session_id: &str, kind: &str) -> PathBuf {
4431 lifecycle_outbox_path_in(&self.spool_dir, session_id, kind)
4432 }
4433
4434 async fn remove_lifecycle_outbox(&self, session_id: &str, kind: &str) -> Result<(), FinalizeError> {
4435 remove_outbox(&self.lifecycle_outbox_path(session_id, kind)).await
4436 }
4437}
4438
4439fn normalize_extra_ttl(trajectory: &mut av_atif::Trajectory) {
4458 if let Some(extra) = trajectory.agent.extra.as_mut() {
4459 if let Some(object) = extra.as_object_mut() {
4460 if object.contains_key("ttl_remaining_s") {
4461 object.insert("ttl_remaining_s".to_owned(), serde_json::Value::Null);
4462 }
4463 }
4464 }
4465}
4466
4467fn normalize_extra_enforcement(trajectory: &mut av_atif::Trajectory) {
4477 if let Some(metrics) = trajectory.final_metrics.as_mut() {
4478 if let Some(extra) = metrics.extra.as_mut() {
4479 if let Some(object) = extra.as_object_mut() {
4480 object.remove("enforcement_latched");
4481 }
4482 }
4483 }
4484}
4485
4486#[cfg(test)]
4487mod normalize_tests {
4488 #![allow(clippy::unwrap_used)]
4489
4490 use super::*;
4491
4492 #[test]
4500 fn enforcement_normalization_is_symmetric_for_absent_keys() {
4501 let base = serde_json::json!({
4502 "schema_version": "ATIF-v1.7",
4503 "session_id": "s",
4504 "agent": {"name": "a", "version": "1"},
4505 "steps": [{"step_id": 1, "source": "user", "message": "hi"}],
4506 "final_metrics": {
4507 "total_steps": 1,
4508 "extra": {"stop_reason_id": 1u64, "enforcement_latched": 0u64},
4509 },
4510 });
4511 let mut rebuilt: av_atif::Trajectory = serde_json::from_value(base).unwrap();
4512 let mut legacy = rebuilt.clone();
4513 if let Some(metrics) = legacy.final_metrics.as_mut() {
4514 if let Some(extra) = metrics.extra.as_mut() {
4515 extra.as_object_mut().unwrap().remove("enforcement_latched");
4516 }
4517 }
4518 assert_ne!(rebuilt, legacy, "precondition: raw copies differ on the key");
4519 normalize_extra_enforcement(&mut rebuilt);
4520 normalize_extra_enforcement(&mut legacy);
4521 assert_eq!(
4522 rebuilt, legacy,
4523 "normalized copies must be equal regardless of key presence"
4524 );
4525 }
4526}
4527
4528fn archive_conflicting_atif(
4529 path: &std::path::Path,
4530 new_trajectory_id: Option<&str>,
4531 session_id: &str,
4532) -> std::io::Result<()> {
4533 let existing_id = match av_core::fsutil::read_capped(path, av_core::fsutil::MAX_ATIF_BYTES) {
4534 Ok(bytes) => serde_json::from_slice::<serde_json::Value>(&bytes)
4535 .ok()
4536 .and_then(|existing| {
4537 existing
4538 .get("trajectory_id")
4539 .and_then(serde_json::Value::as_str)
4540 .map(str::to_owned)
4541 }),
4542 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
4554 return archive_orphaned_atif_siblings(path, session_id);
4555 }
4556 Err(_) => None,
4559 };
4560 if existing_id.is_some() && existing_id.as_deref() == new_trajectory_id {
4561 return Ok(());
4562 }
4563 let mut suffix: String = existing_id
4572 .unwrap_or_default()
4573 .chars()
4574 .filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_'))
4575 .take(64)
4576 .collect();
4577 if suffix.is_empty() {
4578 suffix = format!("corrupt-{}", av_core::new_event_uid());
4579 }
4580 let mut archived = path.with_extension(format!("archived-{suffix}"));
4581 if archived.exists() {
4582 archived = path.with_extension(format!("archived-{suffix}-{}", av_core::new_event_uid()));
4583 }
4584 tracing::warn!(
4585 session = %session_id,
4586 archived = %av_core::fsutil::basename(&archived),
4587 "ATIF path collision (recycled session id); archiving previous incarnation's trajectory",
4588 );
4589 std::fs::rename(path, &archived)?;
4590 let sidecar = path.with_extension("atif-auth");
4595 let mut archived_sidecar = archived.clone();
4596 archived_sidecar.as_mut_os_string().push(".atif-auth");
4597 match std::fs::rename(&sidecar, &archived_sidecar) {
4598 Ok(()) => {}
4599 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
4600 Err(error) => return Err(error),
4601 }
4602 let close_marker = close_complete_marker_path(path);
4606 let mut archived_marker = archived.clone();
4607 archived_marker.as_mut_os_string().push(".close-complete");
4608 match std::fs::rename(&close_marker, &archived_marker) {
4609 Ok(()) => {}
4610 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
4611 Err(error) => return Err(error),
4612 }
4613 let promote_marker = path.with_extension("promote");
4635 let mut archived_promote = archived.clone();
4636 archived_promote.as_mut_os_string().push(".promote-archived");
4637 match std::fs::rename(&promote_marker, &archived_promote) {
4638 Ok(()) => Ok(()),
4639 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
4640 Err(error) => Err(error),
4641 }
4642}
4643
4644fn archive_orphaned_atif_siblings(path: &std::path::Path, session_id: &str) -> std::io::Result<()> {
4656 let mut moved: Vec<String> = Vec::new();
4657 for (sibling, archived_ext) in [
4658 (path.with_extension("atif-auth"), "orphaned-atif-auth"),
4659 (path.with_extension("close-complete"), "orphaned-close-complete"),
4660 (path.with_extension("promote"), "orphaned-promote"),
4661 ] {
4662 if !sibling.exists() {
4663 continue;
4664 }
4665 let archived = path.with_extension(format!("{archived_ext}-{}", av_core::new_event_uid()));
4668 match std::fs::rename(&sibling, &archived) {
4669 Ok(()) => moved.push(av_core::fsutil::basename(&sibling).to_owned()),
4670 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
4671 Err(error) => return Err(error),
4672 }
4673 }
4674 if !moved.is_empty() {
4675 tracing::warn!(
4676 session = %session_id,
4677 orphans = ?moved,
4678 "archived orphaned ATIF siblings left by a crashed archive/quarantine; \
4679 the recycled session id can now seal fresh provenance",
4680 );
4681 }
4682 Ok(())
4683}
4684
4685fn archive_conflicting_receipt(
4695 path: &std::path::Path,
4696 new_receipt_id: &str,
4697 session_id: &str,
4698) -> std::io::Result<()> {
4699 let existing_id = match av_core::fsutil::read_capped(path, av_core::fsutil::MAX_RECEIPT_BYTES) {
4700 Ok(bytes) => serde_json::from_slice::<Receipt>(&bytes)
4701 .ok()
4702 .map(|existing| existing.body.receipt_id),
4703 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
4704 Err(_) => None,
4707 };
4708 if existing_id.as_deref() == Some(new_receipt_id) {
4709 return Ok(());
4710 }
4711 let mut suffix: String = existing_id
4715 .unwrap_or_default()
4716 .chars()
4717 .filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'))
4718 .take(64)
4719 .collect();
4720 if suffix.is_empty() {
4721 suffix = format!("corrupt-{}", av_core::new_event_uid());
4722 }
4723 let mut archived = path.with_extension(format!("archived-{suffix}.json"));
4724 if archived.exists() {
4725 archived = path.with_extension(format!("archived-{suffix}-{}.json", av_core::new_event_uid()));
4726 }
4727 tracing::warn!(
4728 session = %session_id,
4729 archived = %av_core::fsutil::basename(&archived),
4730 "receipt path collision (recycled session id); archiving previous incarnation's receipt",
4731 );
4732 std::fs::rename(path, &archived)
4733}
4734
4735pub(crate) async fn replay_lifecycle_outboxes_in(
4742 spool_dir: &std::path::Path,
4743 journal_key: &[u8; 32],
4744 bridge: Arc<dyn EventBus>,
4745 metrics: &Registry,
4746) -> Result<usize, FinalizeError> {
4747 let directory = spool_dir.join(crate::spool::OUTBOX);
4748 let mut entries = match tokio::fs::read_dir(&directory).await {
4749 Ok(entries) => entries,
4750 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(0),
4751 Err(error) => return Err(FinalizeError::bridge_source(error)),
4752 };
4753 let mut replayed = 0usize;
4754 let mut examined = 0usize;
4755 let mut dirents_seen = 0usize;
4756 while let Some(entry) = entries.next_entry().await.map_err(FinalizeError::bridge_source)? {
4757 dirents_seen = dirents_seen.saturating_add(1);
4758 if dirents_seen > MAX_RECOVERY_DIRENTS_PER_TICK {
4759 bump_recovery_scan_cap(metrics, "replay_lifecycle_outboxes", examined, dirents_seen);
4760 break;
4761 }
4762 let path = entry.path();
4763 if path.extension().and_then(std::ffi::OsStr::to_str) != Some("json") {
4764 continue;
4765 }
4766 examined = examined.saturating_add(1);
4767 if examined > MAX_RECOVERY_ENTRIES_PER_TICK {
4768 bump_recovery_scan_cap(metrics, "replay_lifecycle_outboxes", examined, dirents_seen);
4769 break;
4770 }
4771 let sealed = match read_capped_async(path.clone(), av_core::fsutil::MAX_CONTROL_BYTES).await {
4772 Ok(bytes) => bytes,
4773 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
4776 Err(error) => {
4777 tracing::warn!(%error, path = %av_core::fsutil::basename(&path), "skipping unreadable outbox file");
4778 continue;
4779 }
4780 };
4781 let mut outbox: LifecycleOutbox = match crate::journal::open(
4782 journal_key,
4783 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
4784 0,
4785 &sealed,
4786 ) {
4787 Ok(outbox) => outbox,
4788 Err(error) => {
4793 tracing::warn!(%error, path = %av_core::fsutil::basename(&path), "skipping malformed outbox");
4794 continue;
4795 }
4796 };
4797 if path != lifecycle_outbox_path_in(spool_dir, &outbox.session_id, &outbox.kind) {
4798 tracing::warn!(
4799 path = %av_core::fsutil::basename(&path),
4800 "skipping outbox whose filename does not match its authenticated session_id/kind"
4801 );
4802 continue;
4803 }
4804 if outbox.ack.is_none() {
4805 let topic = outbox.topic.clone();
4806 let key = outbox.key.clone();
4807 let value = outbox.value.clone();
4808 let outcome: Result<(), FinalizeError> = async {
4818 sync_existing_file_parent(&path)
4819 .await
4820 .map_err(FinalizeError::bridge_source)?;
4821 let event_uid = lifecycle_event_uid(&value)?;
4822 outbox.ack =
4823 Some(resolve_lifecycle_ack(Arc::clone(&bridge), topic, key, value, event_uid).await?);
4824 persist_outbox(&path, &outbox, journal_key).await
4825 }
4826 .await;
4827 match outcome {
4828 Ok(()) => replayed = replayed.saturating_add(1),
4829 Err(error) => {
4830 tracing::warn!(
4831 %error,
4832 path = %av_core::fsutil::basename(&path),
4833 "outbox replay failed for this session; will retry next tick"
4834 );
4835 }
4836 }
4837 }
4838 if outbox.ack.is_some() && outbox.kind == crate::journal::SESSION_QUARANTINE_OUTBOX_KIND {
4839 if let Err(error) = archive_quarantine_notice(spool_dir, &outbox, journal_key).await {
4840 tracing::warn!(%error, "quarantine notification archival pending; will retry");
4841 }
4842 }
4843 }
4844 Ok(replayed)
4845}
4846
4847fn quarantine_notice_path(spool_dir: &std::path::Path, session_id: &str) -> PathBuf {
4848 let session_hash = &av_core::digest::sha256_hex(session_id.as_bytes())[..32];
4849 spool_dir
4850 .join("quarantine-notices")
4851 .join(format!("{session_hash}.json"))
4852}
4853
4854async fn archive_quarantine_notice(
4855 spool_dir: &std::path::Path,
4856 outbox: &LifecycleOutbox,
4857 journal_key: &[u8; 32],
4858) -> Result<(), FinalizeError> {
4859 persist_outbox(
4862 &quarantine_notice_path(spool_dir, &outbox.session_id),
4863 outbox,
4864 journal_key,
4865 )
4866 .await?;
4867 remove_outbox(&lifecycle_outbox_path_in(
4868 spool_dir,
4869 &outbox.session_id,
4870 &outbox.kind,
4871 ))
4872 .await
4873}
4874
4875fn lifecycle_outbox_path_in(spool_dir: &std::path::Path, session_id: &str, kind: &str) -> PathBuf {
4878 let session_hash = &av_core::digest::sha256_hex(session_id.as_bytes())[..32];
4879 spool_dir
4880 .join(crate::spool::OUTBOX)
4881 .join(format!("{session_hash}.{kind}.json"))
4882}
4883async fn persist_outbox(
4884 path: &std::path::Path,
4885 outbox: &LifecycleOutbox,
4886 journal_key: &[u8; 32],
4887) -> Result<(), FinalizeError> {
4888 let path = path.to_path_buf();
4889 let bytes = crate::journal::seal(journal_key, crate::journal::LIFECYCLE_OUTBOX_DOMAIN, 0, outbox)
4890 .map_err(FinalizeError::bridge)?;
4891 tokio::task::spawn_blocking(move || av_core::fsutil::write_atomic(&path, &bytes))
4892 .await
4893 .map_err(FinalizeError::Task)?
4894 .map_err(FinalizeError::bridge_source)
4895}
4896
4897async fn sync_existing_file_parent(path: &std::path::Path) -> std::io::Result<()> {
4901 let path = path.to_path_buf();
4902 tokio::task::spawn_blocking(move || {
4903 #[cfg(test)]
4904 if PARENT_SYNC_FAILURES
4905 .get()
4906 .is_some_and(|failures| failures.lock().contains(&path))
4907 {
4908 return Err(std::io::Error::other("injected parent directory sync failure"));
4909 }
4910 let parent = path
4911 .parent()
4912 .filter(|parent| !parent.as_os_str().is_empty())
4913 .unwrap_or_else(|| std::path::Path::new("."));
4914 av_core::fsutil::sync_directory_ancestors(parent)?;
4918 av_core::fsutil::sync_directory(parent)
4919 })
4920 .await
4921 .map_err(std::io::Error::other)?
4922}
4923
4924#[cfg(test)]
4926static PARENT_SYNC_FAILURES: std::sync::OnceLock<parking_lot::Mutex<std::collections::HashSet<PathBuf>>> =
4927 std::sync::OnceLock::new();
4928
4929async fn persist_marker(path: &std::path::Path, bytes: &[u8]) -> Result<(), FinalizeError> {
4930 let path = path.to_path_buf();
4931 let bytes = bytes.to_vec();
4932 tokio::task::spawn_blocking(move || av_core::fsutil::write_atomic(&path, &bytes))
4933 .await
4934 .map_err(FinalizeError::Task)?
4935 .map_err(FinalizeError::atif_source)
4936}
4937
4938async fn remove_outbox(path: &std::path::Path) -> Result<(), FinalizeError> {
4939 let path = path.to_path_buf();
4940 tokio::task::spawn_blocking(move || -> Result<(), FinalizeError> {
4941 let parent = path
4942 .parent()
4943 .ok_or_else(|| FinalizeError::bridge("outbox has no parent".to_owned()))?;
4944 match std::fs::remove_file(&path) {
4945 Ok(()) => {
4946 if let Err(error) = av_core::fsutil::sync_directory(parent) {
4966 tracing::warn!(
4970 target: "av_harness::reconciler",
4971 path = %av_core::fsutil::basename(&path),
4972 parent = %av_core::fsutil::basename(parent),
4973 detail = %error,
4974 "remove_outbox: dirent unsync but file removed; caller state is stable, \
4975 relying on reconciler retry to re-sync on next tick"
4976 );
4977 }
4978 Ok(())
4979 }
4980 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
4981 Err(error) => Err(FinalizeError::bridge_source(error)),
4982 }
4983 })
4984 .await
4985 .map_err(FinalizeError::Task)?
4986}
4987
4988pub fn spawn_reconciler(
4992 sessions: Arc<SessionRegistry>,
4993 finalizer: Finalizer,
4994 idle_s: u64,
4995 tick_s: u64,
4996 breaker: av_loopdetect::BreakerConfig,
4997 metrics: Arc<Registry>,
4998 shutdown: Arc<tokio::sync::Notify>,
4999) -> tokio::task::JoinHandle<()> {
5000 use futures::future::FutureExt as _;
5001 let last_tick_completed = metrics.gauge(
5013 "av_reconciler_last_tick_completed_seconds",
5014 "Unix time when the reconciler last completed a full tick (0 until the first completes)",
5015 );
5016 tokio::spawn(async move {
5017 let mut interval = tokio::time::interval(std::time::Duration::from_secs(tick_s.max(1)));
5018 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
5024 loop {
5025 tokio::select! {
5048 biased;
5049 () = shutdown.notified() => {
5050 tracing::debug!("reconciler shutdown notified; exiting loop between ticks");
5051 return;
5052 }
5053 _ = interval.tick() => {}
5054 }
5055 let started = Instant::now();
5056 let outcome = std::panic::AssertUnwindSafe(async {
5063 if let Err(error) = finalizer.recover_spooled_sessions(&sessions, &breaker).await {
5064 tracing::warn!(error = &error as &dyn std::error::Error, "ATIF spool recovery failed");
5065 metrics
5066 .counter("av_reconcile_errors_total", "Reconciliation errors")
5067 .inc();
5068 }
5069 if let Err(error) = finalizer.retry_marked_promotions(&sessions).await {
5070 tracing::warn!(error = &error as &dyn std::error::Error, "durable promotion retry failed");
5071 metrics
5072 .counter("av_reconcile_errors_total", "Reconciliation errors")
5073 .inc();
5074 }
5075 if let Err(error) = finalizer.complete_pending_closes(&sessions).await {
5082 tracing::warn!(error = &error as &dyn std::error::Error, "pending-close completion sweep failed");
5083 metrics
5084 .counter("av_reconcile_errors_total", "Reconciliation errors")
5085 .inc();
5086 }
5087 const MAX_IDLE_CLOSES_PER_TICK: usize = 64;
5096 let idle = sessions.idle_sessions(idle_s);
5097 let deferred = idle.len().saturating_sub(MAX_IDLE_CLOSES_PER_TICK);
5098 if deferred > 0 {
5099 tracing::info!(
5100 closing = MAX_IDLE_CLOSES_PER_TICK,
5101 deferred,
5102 "idle-close batch capped; remaining sessions close on later ticks"
5103 );
5104 }
5105 for session in idle.into_iter().take(MAX_IDLE_CLOSES_PER_TICK) {
5106 let session_id = session.id.clone();
5107 const IDLE_CLOSE_DEADLINE: std::time::Duration =
5122 std::time::Duration::from_secs(90);
5123 let outcome = tokio::time::timeout(
5124 IDLE_CLOSE_DEADLINE,
5125 finalizer.close_session(session, StopReason::SessionClosed),
5126 )
5127 .await;
5128 match outcome {
5129 Ok(Ok(_)) => {}
5130 Ok(Err(error)) => {
5131 tracing::warn!(session = %session_id, error = &error as &dyn std::error::Error, "idle session finalization failed");
5132 metrics
5133 .counter("av_reconcile_errors_total", "Reconciliation errors")
5134 .inc();
5135 }
5136 Err(_elapsed) => {
5137 tracing::warn!(
5138 session = %session_id,
5139 deadline_s = IDLE_CLOSE_DEADLINE.as_secs(),
5140 "idle-close deadline exceeded; deferring to next tick"
5141 );
5142 metrics
5143 .counter(
5144 "av_idle_close_timeouts_total",
5145 crate::pipeline::IDLE_CLOSE_TIMEOUTS_HELP,
5146 )
5147 .inc();
5148 }
5149 }
5150 }
5151 let evicted = sessions.evict_finalized(idle_s);
5152 if !evicted.is_empty() {
5153 tracing::debug!(count = evicted.len(), "evicted finalized signed sessions");
5154 }
5155 })
5156 .catch_unwind()
5157 .await;
5158 let tick_completed = outcome.is_ok();
5159 if let Err(panic) = outcome {
5160 let msg = panic
5161 .downcast_ref::<&'static str>()
5162 .copied()
5163 .or_else(|| panic.downcast_ref::<String>().map(String::as_str))
5164 .unwrap_or("panic payload was not a string");
5165 metrics
5166 .counter(
5167 "av_reconciler_panics_total",
5168 crate::pipeline::supervision_help::RECONCILER_PANICS,
5169 )
5170 .inc();
5171 tracing::error!(
5172 panic = %msg,
5173 "reconciler tick body panicked; continuing on the next tick"
5174 );
5175 }
5176 metrics
5177 .histogram("av_reconcile_duration_seconds", "Idle reconciliation duration")
5178 .observe_us(elapsed_us(started));
5179 if tick_completed {
5187 last_tick_completed.set(av_core::time::now_ms() / av_core::units::MS_PER_SEC);
5188 }
5189 }
5190 })
5191}
5192
5193fn track_response_attempt(
5194 pending: &mut std::collections::HashSet<String>,
5195 attempt: Option<&crate::worker::ResponseAttempt>,
5196) -> Result<(), FinalizeError> {
5197 let Some(attempt) = attempt else {
5198 return Ok(());
5199 };
5200 if attempt.terminal {
5201 if !pending.remove(&attempt.id) {
5202 pending.insert(format!("orphan-terminal:{}", attempt.id));
5203 }
5204 } else if !pending.insert(attempt.id.clone()) {
5205 return Err(FinalizeError::atif(
5206 "active journal repeats a response attempt id".to_owned(),
5207 ));
5208 }
5209 Ok(())
5210}
5211
5212fn lifecycle_event_uid(value: &serde_json::Value) -> Result<String, FinalizeError> {
5213 value
5214 .get("metadata")
5215 .and_then(|metadata| metadata.get("uid"))
5216 .and_then(serde_json::Value::as_str)
5217 .map(str::to_owned)
5218 .ok_or_else(|| FinalizeError::bridge("lifecycle event has no metadata UID".to_owned()))
5219}
5220
5221async fn resolve_lifecycle_ack(
5222 bridge: Arc<dyn EventBus>,
5223 topic: String,
5224 key: String,
5225 value: serde_json::Value,
5226 event_uid: String,
5227) -> Result<av_bridge::PublishAck, FinalizeError> {
5228 let lookup_bridge = Arc::clone(&bridge);
5229 let lookup_topic = topic.clone();
5230 let lookup_key = key.clone();
5231 let lookup_uid = event_uid.clone();
5232 if let Some(ack) = tokio::task::spawn_blocking(move || {
5233 lookup_bridge.find_event_by_uid(&lookup_topic, &lookup_key, &lookup_uid)
5234 })
5235 .await
5236 .map_err(FinalizeError::Task)?
5237 .map_err(FinalizeError::from)?
5238 {
5239 return Ok(ack);
5240 }
5241 tokio::task::spawn_blocking(move || bridge.publish_idempotent(&topic, &key, &value, &event_uid))
5242 .await
5243 .map_err(FinalizeError::Task)?
5244 .map_err(FinalizeError::from)
5245}
5246
5247async fn quarantine_sibling_exists(spool_dir: &std::path::Path, stem: &str) -> Result<bool, FinalizeError> {
5273 let prefix = format!("{stem}.events.ndjson.corrupt-");
5274 let spool_dir = spool_dir.to_path_buf();
5275 tokio::task::spawn_blocking(move || -> Result<bool, FinalizeError> {
5276 let entries = match std::fs::read_dir(&spool_dir) {
5277 Ok(entries) => entries,
5278 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
5279 Err(error) => return Err(FinalizeError::atif_source(error)),
5280 };
5281 let mut examined = 0usize;
5282 for entry in entries {
5283 examined = examined.saturating_add(1);
5284 if examined > MAX_RECOVERY_DIRENTS_PER_TICK {
5285 return Err(FinalizeError::atif_source(std::io::Error::other(format!(
5286 "quarantine_sibling scan exceeded {MAX_RECOVERY_DIRENTS_PER_TICK} \
5287 entries without finding {prefix}* — spool is over-populated; \
5288 preserving metadata rather than risking data loss"
5289 ))));
5290 }
5291 let entry = entry.map_err(FinalizeError::atif_source)?;
5292 let name = entry.file_name();
5293 if let Some(name) = name.to_str() {
5294 if name.starts_with(&prefix) {
5295 return Ok(true);
5296 }
5297 }
5298 }
5299 Ok(false)
5300 })
5301 .await
5302 .map_err(FinalizeError::Task)?
5303}
5304
5305async fn read_capped_async(path: std::path::PathBuf, max_bytes: u64) -> Result<Vec<u8>, std::io::Error> {
5313 tokio::task::spawn_blocking(move || av_core::fsutil::read_capped(&path, max_bytes))
5314 .await
5315 .map_err(|e| std::io::Error::other(e.to_string()))?
5316}
5317
5318async fn read_complete_journal(path: &std::path::Path) -> Result<Vec<String>, FinalizeError> {
5319 let path = path.to_path_buf();
5320 tokio::task::spawn_blocking(move || -> Result<Vec<String>, FinalizeError> {
5321 let bytes = av_core::fsutil::read_capped(&path, av_core::fsutil::MAX_ATIF_BYTES)
5325 .map_err(FinalizeError::atif_source)?;
5326 if bytes.is_empty() {
5327 return Ok(Vec::new());
5328 }
5329 let complete_len = if bytes.last() == Some(&b'\n') {
5330 bytes.len()
5331 } else {
5332 bytes
5333 .iter()
5334 .rposition(|byte| *byte == b'\n')
5335 .map_or(0, |index| index + 1)
5336 };
5337 if complete_len == 0 && !bytes.is_empty() {
5347 let mut quarantine = path.clone();
5348 let stem = quarantine
5349 .file_name()
5350 .and_then(std::ffi::OsStr::to_str)
5351 .unwrap_or("journal");
5352 let new_name = format!("{stem}.corrupt-{}", av_core::new_event_uid());
5353 quarantine.set_file_name(new_name);
5354 let rename_error_message = match std::fs::rename(&path, &quarantine) {
5355 Ok(()) => {
5356 tracing::error!(
5357 original = %av_core::fsutil::basename(&path),
5358 quarantine = %av_core::fsutil::basename(&quarantine),
5359 bytes = bytes.len(),
5360 "journal has no complete lines; quarantined for post-mortem instead of silent 0-truncate"
5361 );
5362 None
5363 }
5364 Err(rename_error) => {
5365 tracing::error!(
5366 path = %av_core::fsutil::basename(&path),
5367 bytes = bytes.len(),
5368 error = %rename_error,
5369 "journal has no complete lines and quarantine rename failed; refusing to truncate"
5370 );
5371 Some(rename_error.to_string())
5372 }
5373 };
5374 let name = av_core::fsutil::basename(&path);
5390 let qname = av_core::fsutil::basename(&quarantine);
5391 return Err(FinalizeError::atif(match rename_error_message {
5392 None => format!(
5393 "journal {name} contained no complete lines ({} bytes); quarantined at {qname}",
5394 bytes.len()
5395 ),
5396 Some(rename_error) => format!(
5397 "journal {name} contained no complete lines ({} bytes); quarantine rename to {qname} failed: {rename_error}; bytes remain at {name}",
5398 bytes.len()
5399 ),
5400 }));
5401 }
5402 if complete_len < bytes.len() {
5403 tracing::warn!(
5404 path = %av_core::fsutil::basename(&path),
5405 stored = bytes.len(),
5406 keeping = complete_len,
5407 dropping = bytes.len() - complete_len,
5408 "trimming partial trailing line from journal recovery"
5409 );
5410 let mut reopen = std::fs::OpenOptions::new();
5411 reopen.write(true);
5412 #[cfg(unix)]
5413 {
5414 use std::os::unix::fs::OpenOptionsExt as _;
5415 reopen.custom_flags(av_core::fsutil::unix_o_nofollow());
5422 }
5423 let file = reopen.open(&path).map_err(FinalizeError::atif_source)?;
5424 file.set_len(complete_len as u64)
5425 .map_err(FinalizeError::atif_source)?;
5426 file.sync_all().map_err(FinalizeError::atif_source)?;
5427 }
5428 let complete = String::from_utf8(bytes.get(..complete_len).unwrap_or_default().to_vec())
5429 .map_err(FinalizeError::atif_source)?;
5430 Ok(complete
5431 .lines()
5432 .filter(|line| !line.trim().is_empty())
5433 .map(str::to_owned)
5434 .collect())
5435 })
5436 .await
5437 .map_err(FinalizeError::Task)?
5438}
5439
5440#[cfg(test)]
5441mod tests {
5442 #![allow(
5443 clippy::expect_used,
5444 clippy::indexing_slicing,
5445 clippy::panic,
5446 clippy::unwrap_used
5447 )]
5448
5449 use super::*;
5450 use av_bridge::{BusError, PublishAck, StoredEvent};
5451 use av_events::AgentIdentity;
5452 use av_receipts::Ed25519Signer;
5453
5454 struct FailFirstReceiptBus {
5455 fail: std::sync::atomic::AtomicBool,
5456 attempts: parking_lot::Mutex<Vec<(String, serde_json::Value)>>,
5457 }
5458
5459 struct ParentSyncFailure(PathBuf);
5460
5461 impl ParentSyncFailure {
5462 fn new(path: &std::path::Path) -> Self {
5463 assert!(PARENT_SYNC_FAILURES
5464 .get_or_init(Default::default)
5465 .lock()
5466 .insert(path.to_path_buf()));
5467 Self(path.to_path_buf())
5468 }
5469 }
5470
5471 impl Drop for ParentSyncFailure {
5472 fn drop(&mut self) {
5473 PARENT_SYNC_FAILURES.get().unwrap().lock().remove(&self.0);
5474 }
5475 }
5476
5477 fn write_visible_file_without_directory_sync(path: &std::path::Path, bytes: &[u8]) {
5480 use std::io::Write as _;
5481 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
5482 let temporary = path.with_extension("pending");
5483 let mut file = std::fs::File::create(&temporary).unwrap();
5484 file.write_all(bytes).unwrap();
5485 file.sync_all().unwrap();
5486 std::fs::rename(temporary, path).unwrap();
5487 }
5488
5489 #[tokio::test]
5490 async fn resync_does_not_recreate_missing_recovery_directory() {
5491 let dir = tempfile::tempdir().unwrap();
5492 let path = dir.path().join("removed/receipt.json");
5493 assert_eq!(
5494 sync_existing_file_parent(&path).await.unwrap_err().kind(),
5495 std::io::ErrorKind::NotFound
5496 );
5497 assert!(!path.parent().unwrap().exists());
5498 }
5499
5500 #[tokio::test]
5501 async fn lifecycle_retry_resync_failure_blocks_publish_and_isolates_replay() {
5502 let directory = tempfile::tempdir().unwrap();
5503 let bus = Arc::new(FailFirstReceiptBus {
5504 fail: std::sync::atomic::AtomicBool::new(false),
5505 attempts: parking_lot::Mutex::new(Vec::new()),
5506 });
5507 let finalizer = Finalizer::with_bridge(
5508 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
5509 directory.path().to_path_buf(),
5510 Arc::new(Registry::new()),
5511 bus.clone(),
5512 );
5513 let session = session(Workflow::Signed);
5514 let failed_uid = av_core::new_event_uid();
5515 let healthy_uid = av_core::new_event_uid();
5516 for (kind, uid, class) in [
5517 (
5518 crate::journal::RECEIPT_OUTBOX_KIND,
5519 &failed_uid,
5520 av_events::EventClass::Receipt,
5521 ),
5522 (
5523 crate::journal::SESSION_CLOSE_OUTBOX_KIND,
5524 &healthy_uid,
5525 av_events::EventClass::Session,
5526 ),
5527 ] {
5528 let outbox = LifecycleOutbox {
5529 schema_version: LIFECYCLE_OUTBOX_SCHEMA_V1,
5530 session_id: session.id.clone(),
5531 kind: kind.to_owned(),
5532 topic: class.topic().to_owned(),
5533 key: session.identity.instance_uid.clone(),
5534 value: serde_json::json!({"metadata": {"uid": uid, "sequence": 42}}),
5535 ack: None,
5536 session_scope: Some(session.session_scope.clone()),
5537 };
5538 let sealed = crate::journal::seal(
5539 &finalizer.journal_key,
5540 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
5541 0,
5542 &outbox,
5543 )
5544 .unwrap();
5545 write_visible_file_without_directory_sync(
5546 &finalizer.lifecycle_outbox_path(&session.id, kind),
5547 &sealed,
5548 );
5549 }
5550 let failed_path = finalizer.lifecycle_outbox_path(&session.id, crate::journal::RECEIPT_OUTBOX_KIND);
5551 let original_bytes = std::fs::read(&failed_path).unwrap();
5552 let failure = ParentSyncFailure::new(&failed_path);
5553 let seq_before = session.peek_seq();
5554 assert!(finalizer
5555 .emit_bridge_event(
5556 &session,
5557 av_events::EventClass::Receipt,
5558 serde_json::json!({}),
5559 crate::journal::RECEIPT_OUTBOX_KIND,
5560 )
5561 .await
5562 .is_err());
5563 assert_eq!(session.peek_seq(), seq_before);
5564 assert!(finalizer
5565 .replay_unacked_lifecycle_outbox(&session, crate::journal::RECEIPT_OUTBOX_KIND)
5566 .await
5567 .is_err());
5568 assert!(bus.attempts.lock().is_empty());
5569 assert_eq!(finalizer.replay_lifecycle_outboxes().await.unwrap(), 1);
5572 assert_eq!(bus.attempts.lock().len(), 1);
5573 assert_eq!(bus.attempts.lock()[0].1["metadata"]["uid"], healthy_uid);
5574 assert_eq!(std::fs::read(&failed_path).unwrap(), original_bytes);
5575 drop(failure);
5576 finalizer
5577 .emit_bridge_event(
5578 &session,
5579 av_events::EventClass::Receipt,
5580 serde_json::json!({}),
5581 crate::journal::RECEIPT_OUTBOX_KIND,
5582 )
5583 .await
5584 .unwrap();
5585 assert_eq!(bus.attempts.lock().len(), 2);
5586 assert_eq!(bus.attempts.lock()[1].1["metadata"]["uid"], failed_uid);
5587 finalizer
5588 .replay_unacked_lifecycle_outbox(&session, crate::journal::RECEIPT_OUTBOX_KIND)
5589 .await
5590 .unwrap();
5591 assert_eq!(finalizer.replay_lifecycle_outboxes().await.unwrap(), 0);
5592 assert_eq!(
5593 bus.attempts.lock().len(),
5594 2,
5595 "retry must retain the original event UID and acknowledgment"
5596 );
5597 }
5598
5599 #[tokio::test]
5600 async fn unsigned_recovery_resync_failures_preserve_provenance_and_receipt() {
5601 let directory = tempfile::tempdir().unwrap();
5602 let finalizer = finalizer(directory.path());
5603 let original = session(Workflow::Unsigned);
5604 original
5605 .atif
5606 .lock()
5607 .push_step(av_atif::Step {
5608 step_id: 0,
5609 timestamp: None,
5610 source: av_atif::Source::User,
5611 message: serde_json::json!("retained evidence"),
5612 reasoning_effort: None,
5613 reasoning_content: None,
5614 model_name: None,
5615 tool_calls: None,
5616 observation: None,
5617 metrics: None,
5618 is_copied_context: None,
5619 llm_call_count: None,
5620 extra: None,
5621 })
5622 .unwrap();
5623 let FinalizeOutcome::Atif { path } = finalizer
5624 .close_session(original.clone(), StopReason::SessionClosed)
5625 .await
5626 .unwrap()
5627 else {
5628 panic!("expected unsigned artifact")
5629 };
5630 let receipt = finalizer.promote(original.clone()).await.unwrap();
5631 let proof = path.with_extension("atif-auth");
5632 let receipt_path = finalizer.receipt_path(&original.id);
5633 let originals: Vec<_> = [&path, &proof, &receipt_path]
5634 .into_iter()
5635 .map(|path| (path.clone(), std::fs::read(path).unwrap()))
5636 .collect();
5637 let old = std::time::SystemTime::now() - std::time::Duration::from_secs(120);
5638 std::fs::File::options()
5639 .write(true)
5640 .open(&path)
5641 .unwrap()
5642 .set_times(std::fs::FileTimes::new().set_modified(old))
5643 .unwrap();
5644 let registry = SessionRegistry::new();
5645 for retry_path in [&proof, &receipt_path] {
5646 let bytes = std::fs::read(retry_path).unwrap();
5647 write_visible_file_without_directory_sync(retry_path, &bytes);
5648 let failure = ParentSyncFailure::new(retry_path);
5649 assert_eq!(
5650 finalizer
5651 .recover_spooled_sessions(®istry, &Default::default())
5652 .await
5653 .unwrap(),
5654 0
5655 );
5656 assert!(registry.get(&original.id).is_none());
5657 for (path, bytes) in &originals {
5658 assert_eq!(
5659 &std::fs::read(path).unwrap(),
5660 bytes,
5661 "a sync error must not quarantine or overwrite authenticated evidence"
5662 );
5663 }
5664 drop(failure);
5665 }
5666 assert_eq!(
5667 finalizer
5668 .recover_spooled_sessions(®istry, &Default::default())
5669 .await
5670 .unwrap(),
5671 1
5672 );
5673 let recovered = registry.get(&original.id).unwrap();
5674 assert_eq!(
5675 recovered.receipt.lock().as_ref().unwrap().body.receipt_id,
5676 receipt.body.receipt_id
5677 );
5678 }
5679
5680 #[tokio::test]
5681 async fn signed_receipt_recovery_resync_failure_preserves_source_journal() {
5682 let directory = tempfile::tempdir().unwrap();
5683 let finalizer = finalizer(directory.path());
5684 let original = session(Workflow::Signed);
5685 let event = av_events::OcsfEventBuilder::new(
5686 av_events::EventClass::Compression,
5687 original.id.clone(),
5688 original.identity.clone(),
5689 0,
5690 )
5691 .payload(serde_json::json!({}))
5692 .build()
5693 .unwrap();
5694 let event_value = serde_json::to_value(&event).unwrap();
5695 original.chain.lock().append(&event_value).unwrap();
5696 let FinalizeOutcome::Receipt { receipt } = finalizer
5697 .close_session(original.clone(), StopReason::SessionClosed)
5698 .await
5699 .unwrap()
5700 else {
5701 panic!("expected signed receipt")
5702 };
5703 let stem = &av_core::digest::sha256_hex(original.id.as_bytes())[..32];
5704 let metadata_path = directory.path().join(format!("{stem}.session.json"));
5705 let journal_path = directory.path().join(format!("{stem}.events.ndjson"));
5706 let metadata = serde_json::json!({
5707 "journal_version": 2, "session_id": original.id,
5708 "identity": original.identity, "workflow": "signed",
5709 });
5710 std::fs::write(
5711 &metadata_path,
5712 crate::journal::seal(&finalizer.journal_key, "metadata", 0, &metadata).unwrap(),
5713 )
5714 .unwrap();
5715 let record = crate::worker::ActiveJournalRecord {
5716 event: event_value,
5717 identity: original.identity.clone(),
5718 atif_step: None,
5719 tool_calls: 0,
5720 tool_allowed: 0,
5721 tool_blocked: 0,
5722 prompt_tokens: 0,
5723 completion_tokens: 0,
5724 cached_tokens: 0,
5725 cost_usd_micros: 0,
5726 prompt_token_correction: 0,
5727 stop_reason_id: None,
5728 response_attempt: None,
5729 };
5730 let mut line = crate::journal::seal(
5731 &finalizer.journal_key,
5732 &format!("{}:active", original.id),
5733 0,
5734 &record,
5735 )
5736 .unwrap();
5737 line.push(b'\n');
5738 std::fs::write(&journal_path, &line).unwrap();
5739 crate::worker::persist_broker_ack(
5740 directory.path(),
5741 &original.id,
5742 &event.metadata.uid,
5743 &PublishAck {
5744 topic: av_events::EventClass::Compression.topic().into(),
5745 partition: 0,
5746 offset: 1,
5747 },
5748 &finalizer.journal_key,
5749 )
5750 .await
5751 .unwrap();
5752 let receipt_path = finalizer.receipt_path(&original.id);
5753 let receipt_bytes = std::fs::read(&receipt_path).unwrap();
5754 write_visible_file_without_directory_sync(&receipt_path, &receipt_bytes);
5755 let failure = ParentSyncFailure::new(&receipt_path);
5756 let registry = SessionRegistry::new();
5757 assert_eq!(
5758 finalizer
5759 .recover_spooled_sessions(®istry, &Default::default())
5760 .await
5761 .unwrap(),
5762 0
5763 );
5764 assert!(registry.get(&original.id).is_none());
5765 assert!(metadata_path.exists());
5766 assert_eq!(std::fs::read(&journal_path).unwrap(), line);
5767 assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt_bytes);
5768 drop(failure);
5769 assert_eq!(
5770 finalizer
5771 .recover_spooled_sessions(®istry, &Default::default())
5772 .await
5773 .unwrap(),
5774 1
5775 );
5776 assert_eq!(
5777 registry
5778 .get(&original.id)
5779 .unwrap()
5780 .receipt
5781 .lock()
5782 .as_ref()
5783 .unwrap()
5784 .body
5785 .receipt_id,
5786 receipt.body.receipt_id
5787 );
5788 assert!(
5789 !journal_path.exists(),
5790 "successful durable adoption completes cleanup"
5791 );
5792 }
5793
5794 #[tokio::test]
5797 async fn quarantine_notification_is_durable_deduplicated_and_never_a_receipt() {
5798 let directory = tempfile::tempdir().unwrap();
5799 let bus = Arc::new(FailFirstReceiptBus {
5800 fail: std::sync::atomic::AtomicBool::new(false),
5801 attempts: parking_lot::Mutex::new(Vec::new()),
5802 });
5803 let make_finalizer = || {
5804 Finalizer::with_bridge(
5805 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
5806 directory.path().to_path_buf(),
5807 Arc::new(Registry::new()),
5808 bus.clone(),
5809 )
5810 };
5811 let first = make_finalizer();
5812 let failed = session(Workflow::Signed);
5813 failed.mark_capture_failed();
5814 assert!(matches!(
5815 first
5816 .close_session(failed.clone(), StopReason::SessionClosed)
5817 .await,
5818 Err(FinalizeError::CaptureIncomplete)
5819 ));
5820 let outbox = first.lifecycle_outbox_path(&failed.id, crate::journal::SESSION_QUARANTINE_OUTBOX_KIND);
5821 let notice = quarantine_notice_path(directory.path(), &failed.id);
5822 assert!(
5823 !outbox.exists(),
5824 "completed notices must not consume active replay capacity"
5825 );
5826 assert!(notice.exists());
5827 assert!(!first.receipt_path(&failed.id).exists());
5828 std::fs::copy(¬ice, &outbox).unwrap();
5831 std::fs::remove_file(¬ice).unwrap();
5832 assert_eq!(
5833 replay_lifecycle_outboxes_in(
5834 directory.path(),
5835 &first.journal_key,
5836 bus.clone(),
5837 &Registry::new()
5838 )
5839 .await
5840 .unwrap(),
5841 0
5842 );
5843 assert!(notice.exists());
5844 assert!(!outbox.exists());
5845 std::fs::copy(¬ice, &outbox).unwrap();
5847 first
5848 .remove_acked_lifecycle_outboxes(&SessionRegistry::new())
5849 .await
5850 .unwrap();
5851 assert!(
5852 notice.exists(),
5853 "the quarantine anchor must survive outbox cleanup"
5854 );
5855 assert!(!outbox.exists());
5856 let restarted = make_finalizer();
5857 let reconstructed = session(Workflow::Signed);
5858 assert_ne!(failed.session_scope, reconstructed.session_scope);
5859 reconstructed.mark_capture_failed();
5860 restarted.emit_quarantine_event(&reconstructed).await.unwrap();
5861 restarted
5864 .emit_bridge_event(
5865 &reconstructed,
5866 av_events::EventClass::Session,
5867 serde_json::json!({"action": "quarantined", "evidence_complete": false}),
5868 crate::journal::SESSION_QUARANTINE_OUTBOX_KIND,
5869 )
5870 .await
5871 .unwrap();
5872 {
5873 let attempts = bus.attempts.lock();
5874 assert_eq!(attempts.len(), 1, "restart must reuse the acknowledged event UID");
5875 let (topic, event) = attempts.first().unwrap();
5876 assert_eq!(topic, "agent.session");
5877 assert_eq!(event.pointer("/payload/action").unwrap(), "quarantined");
5878 assert_eq!(event.pointer("/payload/evidence_complete").unwrap(), false);
5879 assert!(event.pointer("/payload/receipt").is_none());
5880 }
5881 std::fs::write(notice, b"tampered").unwrap();
5883 assert!(restarted.emit_quarantine_event(&reconstructed).await.is_err());
5884 assert_eq!(bus.attempts.lock().len(), 1);
5885 }
5886
5887 #[test]
5889 fn finalize_error_preserves_io_error_kind_through_source() {
5890 let error = FinalizeError::atif_source(std::io::Error::new(
5891 std::io::ErrorKind::NotFound,
5892 "spool file vanished",
5893 ));
5894 assert_eq!(error.to_string(), "ATIF finalization failed: spool file vanished");
5896 let source = std::error::Error::source(&error).expect("typed source must survive");
5897 let io = source
5898 .downcast_ref::<std::io::Error>()
5899 .expect("source must downcast to io::Error");
5900 assert_eq!(io.kind(), std::io::ErrorKind::NotFound);
5901 assert!(std::error::Error::source(&FinalizeError::atif("no step")).is_none());
5903 }
5904
5905 #[test]
5910 fn bus_error_conversion_keeps_classification_and_typed_source() {
5911 let permanent = FinalizeError::from(BusError::UnknownTopic("agent.receipt".to_owned()));
5912 assert!(matches!(permanent, FinalizeError::BridgeConfig { .. }));
5913 assert!(std::error::Error::source(&permanent)
5914 .and_then(|source| source.downcast_ref::<BusError>())
5915 .is_some());
5916
5917 let transient = FinalizeError::from(BusError::Io(std::io::Error::new(
5918 std::io::ErrorKind::PermissionDenied,
5919 "broker socket",
5920 )));
5921 assert!(matches!(transient, FinalizeError::Bridge { .. }));
5922 let bus = std::error::Error::source(&transient)
5923 .and_then(|source| source.downcast_ref::<BusError>())
5924 .expect("BusError source");
5925 let io = std::error::Error::source(bus)
5927 .and_then(|source| source.downcast_ref::<std::io::Error>())
5928 .expect("io::Error inside BusError");
5929 assert_eq!(io.kind(), std::io::ErrorKind::PermissionDenied);
5930 }
5931
5932 impl EventBus for FailFirstReceiptBus {
5933 fn publish(
5934 &self,
5935 topic: &str,
5936 _key: &str,
5937 value: &serde_json::Value,
5938 ) -> Result<PublishAck, BusError> {
5939 self.attempts.lock().push((topic.to_owned(), value.clone()));
5940 if topic == "agent.receipt" && self.fail.swap(false, std::sync::atomic::Ordering::AcqRel) {
5941 return Err(BusError::Backend("injected receipt outage".to_owned()));
5942 }
5943 Ok(PublishAck {
5944 topic: topic.to_owned(),
5945 partition: 0,
5946 offset: self.attempts.lock().len() as u64,
5947 })
5948 }
5949
5950 fn fetch(
5951 &self,
5952 _topic: &str,
5953 _partition: u32,
5954 _offset: u64,
5955 _max: usize,
5956 ) -> Result<Vec<StoredEvent>, BusError> {
5957 Ok(Vec::new())
5958 }
5959
5960 fn partitions(&self, _topic: &str) -> Result<u32, BusError> {
5961 Ok(1)
5962 }
5963
5964 fn topics(&self) -> Vec<String> {
5965 av_events::EventClass::all()
5966 .iter()
5967 .map(|class| class.topic().to_owned())
5968 .collect()
5969 }
5970 }
5971
5972 fn session(workflow: Workflow) -> Arc<Session> {
5973 Arc::new(Session::new(
5974 "lifecycle-session".to_owned(),
5975 workflow,
5976 AgentIdentity {
5977 version: "1".to_owned(),
5978 charter: "test".into(),
5979 instance_uid: "instance-1".to_owned(),
5980 ttl_remaining_s: Some(600),
5981 },
5982 Default::default(),
5983 ))
5984 }
5985
5986 fn finalizer(directory: &std::path::Path) -> Finalizer {
5987 Finalizer::new(
5988 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
5989 directory.to_path_buf(),
5990 Arc::new(Registry::new()),
5991 )
5992 }
5993
5994 #[tokio::test]
5995 async fn unsigned_principal_binding_survives_artifact_restart() {
5996 for principal in [Some("principal-a"), None] {
5997 let directory = tempfile::tempdir().unwrap();
5998 let finalizer = finalizer(directory.path());
5999 let original = session(Workflow::Unsigned);
6000 assert!(original.bind_principal(principal));
6001 original
6002 .atif
6003 .lock()
6004 .push_step(av_atif::Step {
6005 step_id: 0,
6006 timestamp: None,
6007 source: av_atif::Source::User,
6008 message: serde_json::json!("private session content"),
6009 reasoning_effort: None,
6010 reasoning_content: None,
6011 model_name: None,
6012 tool_calls: None,
6013 observation: None,
6014 metrics: None,
6015 is_copied_context: None,
6016 llm_call_count: None,
6017 extra: None,
6018 })
6019 .unwrap();
6020 finalizer
6021 .close_session(Arc::clone(&original), StopReason::SessionClosed)
6022 .await
6023 .unwrap();
6024 let path = original.atif_path.lock().clone().unwrap();
6025 let mut artifact: av_atif::Trajectory =
6026 serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
6027 assert_eq!(
6028 artifact.agent.extra.as_ref().unwrap()["principal_binding"],
6029 serde_json::json!(principal)
6030 );
6031 if principal.is_none() {
6032 artifact
6035 .agent
6036 .extra
6037 .as_mut()
6038 .unwrap()
6039 .as_object_mut()
6040 .unwrap()
6041 .remove("principal_binding");
6042 let bytes = serde_json::to_vec(&artifact).unwrap();
6043 std::fs::write(&path, &bytes).unwrap();
6044 let proof = AtifProvenance {
6045 session_id: original.id.clone(),
6046 digest: av_core::digest::sha256_hex(&bytes),
6047 };
6048 std::fs::write(
6049 path.with_extension("atif-auth"),
6050 crate::journal::seal(&finalizer.journal_key, "atif-provenance", 0, &proof).unwrap(),
6051 )
6052 .unwrap();
6053 let _ = std::fs::remove_file(path.with_extension("close-complete"));
6056 }
6057 let registry = SessionRegistry::new();
6058 finalizer
6059 .recover_spooled_sessions(®istry, &Default::default())
6060 .await
6061 .unwrap();
6062 let recovered = registry
6063 .get(&original.id)
6064 .expect("artifact must be adopted after restart");
6065 assert_eq!(recovered.principal_binding().as_deref(), principal);
6066 assert!(
6067 !recovered.bind_principal(Some("principal-b")),
6068 "a different subject must not acquire a recovered session"
6069 );
6070 assert!(recovered.bind_principal(principal));
6071 }
6072 }
6073
6074 #[tokio::test]
6075 async fn unsigned_principal_binding_survives_journal_consolidation() {
6076 use av_events::{EventClass, OcsfEventBuilder, StatusId};
6077 let directory = tempfile::tempdir().unwrap();
6078 let finalizer = finalizer(directory.path());
6079 let original = session(Workflow::Unsigned);
6080 assert!(original.bind_principal(Some("principal-a")));
6081 let stem = av_core::digest::sha256_hex(original.id.as_bytes());
6082 let stem = &stem[..32];
6083 let metadata = serde_json::json!({
6084 "journal_version": 2, "session_id": original.id,
6085 "identity": original.identity, "workflow": "unsigned",
6086 "principal_binding": original.principal_binding(),
6087 });
6088 std::fs::write(
6089 directory.path().join(format!("{stem}.session.json")),
6090 crate::journal::seal(&finalizer.journal_key, "metadata", 0, &metadata).unwrap(),
6091 )
6092 .unwrap();
6093 let event = OcsfEventBuilder::new(
6094 EventClass::Compression,
6095 original.id.clone(),
6096 original.identity.clone(),
6097 0,
6098 )
6099 .status(StatusId::Success)
6100 .payload(serde_json::json!({}))
6101 .build()
6102 .unwrap();
6103 let record = crate::worker::ActiveJournalRecord {
6104 event: serde_json::to_value(&event).unwrap(),
6105 identity: original.identity.clone(),
6106 atif_step: Some(av_atif::Step {
6107 step_id: 0,
6108 timestamp: None,
6109 source: av_atif::Source::User,
6110 message: serde_json::json!("private journal content"),
6111 reasoning_effort: None,
6112 reasoning_content: None,
6113 model_name: None,
6114 tool_calls: None,
6115 observation: None,
6116 metrics: None,
6117 is_copied_context: None,
6118 llm_call_count: None,
6119 extra: None,
6120 }),
6121 tool_calls: 0,
6122 tool_allowed: 0,
6123 tool_blocked: 0,
6124 prompt_tokens: 0,
6125 completion_tokens: 0,
6126 cached_tokens: 0,
6127 cost_usd_micros: 0,
6128 prompt_token_correction: 0,
6129 stop_reason_id: None,
6130 response_attempt: None,
6131 };
6132 let mut line = crate::journal::seal(
6133 &finalizer.journal_key,
6134 &format!("{}:active", original.id),
6135 0,
6136 &record,
6137 )
6138 .unwrap();
6139 line.push(b'\n');
6140 std::fs::write(directory.path().join(format!("{stem}.events.ndjson")), line).unwrap();
6141 crate::worker::persist_broker_ack(
6142 directory.path(),
6143 &original.id,
6144 &event.metadata.uid,
6145 &PublishAck {
6146 topic: EventClass::Compression.topic().into(),
6147 partition: 0,
6148 offset: 1,
6149 },
6150 &finalizer.journal_key,
6151 )
6152 .await
6153 .unwrap();
6154 let registry = SessionRegistry::new();
6155 finalizer
6156 .recover_spooled_sessions(®istry, &Default::default())
6157 .await
6158 .unwrap();
6159 let recovered = registry
6160 .get(&original.id)
6161 .expect("consolidated artifact must be adopted");
6162 assert_eq!(recovered.principal_binding().as_deref(), Some("principal-a"));
6163 assert!(!recovered.bind_principal(Some("principal-b")));
6164 assert!(recovered.bind_principal(Some("principal-a")));
6165 assert!(
6166 !directory.path().join(format!("{stem}.session.json")).exists(),
6167 "recovery must have moved ownership into the artifact before deleting the journal metadata"
6168 );
6169 }
6170
6171 #[tokio::test]
6178 async fn reconciler_tick_liveness_gauge_is_preregistered_and_advances() {
6179 let directory = tempfile::tempdir().unwrap();
6180 let metrics = Arc::new(Registry::new());
6181 let finalizer = Finalizer::new(
6182 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
6183 directory.path().to_path_buf(),
6184 Arc::clone(&metrics),
6185 );
6186 let sessions = Arc::new(SessionRegistry::new());
6187 let before = av_core::time::now_ms() / av_core::units::MS_PER_SEC;
6188 let handle = spawn_reconciler(
6189 Arc::clone(&sessions),
6190 finalizer,
6191 3600, 1, Default::default(),
6194 Arc::clone(&metrics),
6195 Arc::new(tokio::sync::Notify::new()),
6196 );
6197 assert!(
6200 metrics
6201 .render()
6202 .contains("av_reconciler_last_tick_completed_seconds"),
6203 "the liveness gauge must be registered at spawn, not lazily"
6204 );
6205 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
6207 loop {
6208 let value = metrics
6209 .gauge(
6210 "av_reconciler_last_tick_completed_seconds",
6211 "Unix time when the reconciler last completed a full tick (0 until the first completes)",
6212 )
6213 .get();
6214 if value >= before {
6215 break;
6216 }
6217 assert!(
6218 std::time::Instant::now() < deadline,
6219 "gauge never advanced past {before}; last value {value}"
6220 );
6221 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
6222 }
6223 handle.abort();
6224 }
6225
6226 #[tokio::test]
6233 async fn emit_bridge_event_persist_failure_does_not_burn_a_seq() {
6234 let directory = tempfile::tempdir().unwrap();
6235 std::fs::write(directory.path().join(crate::spool::OUTBOX), b"").unwrap();
6239 let bus = Arc::new(FailFirstReceiptBus {
6240 fail: std::sync::atomic::AtomicBool::new(false),
6241 attempts: parking_lot::Mutex::new(Vec::new()),
6242 });
6243 let finalizer = Finalizer::with_bridge(
6244 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
6245 directory.path().to_path_buf(),
6246 Arc::new(Registry::new()),
6247 bus,
6248 );
6249 let session = session(Workflow::Signed);
6250 let seq_before = session.peek_seq();
6251 let result = finalizer
6254 .close_session(Arc::clone(&session), StopReason::SessionClosed)
6255 .await;
6256 assert!(
6257 result.is_err(),
6258 "close_session must fail when outbox persist is blocked, got {result:?}",
6259 );
6260 assert_eq!(
6261 session.peek_seq(),
6262 seq_before,
6263 "peek_seq must not advance past a failed persist_outbox — the burned \
6264 seq would misalign the journal on retry",
6265 );
6266 }
6267
6268 #[tokio::test]
6276 async fn emit_bridge_event_reading_persisted_outbox_advances_seq_past_it() {
6277 let directory = tempfile::tempdir().unwrap();
6278 let bus = Arc::new(FailFirstReceiptBus {
6279 fail: std::sync::atomic::AtomicBool::new(false),
6280 attempts: parking_lot::Mutex::new(Vec::new()),
6281 });
6282 let finalizer = Finalizer::with_bridge(
6283 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
6284 directory.path().to_path_buf(),
6285 Arc::new(Registry::new()),
6286 bus.clone(),
6287 );
6288 let session = session(Workflow::Signed);
6289 let receipt_seq = 42u64;
6292 let receipt_event_uid = av_core::new_event_uid();
6293 let value = serde_json::json!({
6294 "metadata": { "sequence": receipt_seq, "uid": receipt_event_uid },
6295 "topic": av_events::EventClass::Receipt.topic(),
6296 });
6297 let outbox = LifecycleOutbox {
6298 schema_version: LIFECYCLE_OUTBOX_SCHEMA_V1,
6299 session_id: session.id.clone(),
6300 kind: crate::journal::RECEIPT_OUTBOX_KIND.to_owned(),
6301 topic: av_events::EventClass::Receipt.topic().to_owned(),
6302 key: session.identity.instance_uid.clone(),
6303 session_scope: Some(session.session_scope.clone()),
6304 value,
6305 ack: Some(av_bridge::PublishAck {
6306 topic: av_events::EventClass::Receipt.topic().to_owned(),
6307 partition: 0,
6308 offset: 1,
6309 }),
6310 };
6311 let outbox_path = finalizer.lifecycle_outbox_path(&session.id, crate::journal::RECEIPT_OUTBOX_KIND);
6312 std::fs::create_dir_all(outbox_path.parent().unwrap()).unwrap();
6313 let sealed = crate::journal::seal(
6314 &finalizer.journal_key,
6315 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
6316 0,
6317 &outbox,
6318 )
6319 .unwrap();
6320 std::fs::write(&outbox_path, sealed).unwrap();
6321 assert!(
6322 session.peek_seq() < receipt_seq,
6323 "precondition: in-memory seq must trail the persisted outbox seq",
6324 );
6325 finalizer
6326 .close_session(Arc::clone(&session), StopReason::SessionClosed)
6327 .await
6328 .unwrap();
6329 let attempts = bus.attempts.lock();
6330 let session_close = attempts
6331 .iter()
6332 .find(|(topic, _)| topic == av_events::EventClass::Session.topic())
6333 .expect("SESSION_CLOSE lifecycle event must reach the bridge");
6334 let session_close_seq = session_close
6335 .1
6336 .get("metadata")
6337 .and_then(|metadata| metadata.get("sequence"))
6338 .and_then(serde_json::Value::as_u64)
6339 .expect("published event must carry a numeric metadata.sequence");
6340 assert!(
6341 session_close_seq > receipt_seq,
6342 "SESSION_CLOSE seq ({session_close_seq}) must exceed the persisted RECEIPT_OUTBOX seq ({receipt_seq}) — otherwise consumers see duplicate metadata.sequence values within one session",
6343 );
6344 }
6345
6346 #[tokio::test]
6356 async fn recovered_signed_session_rejects_leases_even_when_finalize_errors() {
6357 use av_events::{EventClass, OcsfEventBuilder, StatusId};
6358 use std::io::Write as _;
6359
6360 let directory = tempfile::tempdir().unwrap();
6361 let signer: Arc<dyn Signer> = Arc::new(Ed25519Signer::from_seed(&[29; 32]));
6362 let journal_key = crate::journal::key_from_signer(signer.as_ref());
6363 let session_id = "signed-recovery-lease-guard";
6364 let identity = AgentIdentity {
6365 version: "1".into(),
6366 charter: "test".into(),
6367 instance_uid: "instance-lease-guard".into(),
6368 ttl_remaining_s: Some(600),
6369 };
6370
6371 let digest = av_core::digest::sha256_hex(session_id.as_bytes());
6373 let stem = &digest[..32];
6374 let metadata_payload = serde_json::json!({
6375 "journal_version": 2,
6376 "session_id": session_id,
6377 "identity": identity,
6378 "workflow": "signed",
6379 });
6380 let metadata_sealed = crate::journal::seal(&journal_key, "metadata", 0, &metadata_payload).unwrap();
6381 std::fs::write(
6382 directory.path().join(format!("{stem}.session.json")),
6383 &metadata_sealed,
6384 )
6385 .unwrap();
6386
6387 let event = OcsfEventBuilder::new(
6389 EventClass::Compression,
6390 session_id.to_owned(),
6391 identity.clone(),
6392 0,
6393 )
6394 .status(StatusId::Success)
6395 .payload(serde_json::json!({}))
6396 .build()
6397 .unwrap();
6398 let event_uid = event.metadata.uid.clone();
6399 let record = crate::worker::ActiveJournalRecord {
6400 event: serde_json::to_value(&event).unwrap(),
6401 identity: identity.clone(),
6402 atif_step: None,
6403 tool_calls: 0,
6404 tool_allowed: 0,
6405 prompt_token_correction: 0,
6406 tool_blocked: 0,
6407 prompt_tokens: 0,
6408 completion_tokens: 0,
6409 cached_tokens: 0,
6410 cost_usd_micros: 0,
6411 stop_reason_id: None,
6412 response_attempt: None,
6413 };
6414 let domain = format!("{session_id}:active");
6415 let sealed = crate::journal::seal(&journal_key, &domain, 0, &record).unwrap();
6416 let mut journal_file =
6417 std::fs::File::create(directory.path().join(format!("{stem}.events.ndjson"))).unwrap();
6418 journal_file.write_all(&sealed).unwrap();
6419 journal_file.write_all(b"\n").unwrap();
6420 drop(journal_file);
6421
6422 crate::worker::persist_broker_ack(
6424 directory.path(),
6425 session_id,
6426 &event_uid,
6427 &PublishAck {
6428 topic: EventClass::Compression.topic().to_owned(),
6429 partition: 0,
6430 offset: 1,
6431 },
6432 &journal_key,
6433 )
6434 .await
6435 .unwrap();
6436
6437 std::fs::write(directory.path().join("receipts"), b"").unwrap();
6445
6446 let registry = crate::session::SessionRegistry::new();
6447 let finalizer = Finalizer::with_bridge(
6448 signer,
6449 directory.path().to_path_buf(),
6450 Arc::new(Registry::new()),
6451 Arc::new(FailFirstReceiptBus {
6452 fail: std::sync::atomic::AtomicBool::new(false),
6453 attempts: parking_lot::Mutex::new(Vec::new()),
6454 }),
6455 );
6456 let result = finalizer
6457 .recover_spooled_sessions(®istry, &Default::default())
6458 .await;
6459 assert!(result.is_ok(), "per-session errors warn+continue; got {result:?}",);
6475
6476 assert!(
6477 registry.get(session_id).is_none(),
6478 "after a transient close error the session must be removed from the registry so the next reconciler tick can re-adopt via the still-present journal sidecar — leaving it in the registry with `is_closed()` = true would starve recovery forever",
6479 );
6480 }
6481
6482 #[tokio::test]
6499 async fn round_41_f1_corrupt_signed_sidecar_does_not_block_other_signed_recovery() {
6500 let directory = tempfile::tempdir().unwrap();
6501 let poison_stem = "poisonpoisonpoisonpoisonpoison32";
6518 std::fs::write(
6519 directory.path().join(format!("{poison_stem}.session.json")),
6520 b"{\"garbage\": true, \"not\": \"a valid sealed metadata\"}",
6521 )
6522 .unwrap();
6523 std::fs::write(
6524 directory.path().join(format!("{poison_stem}.events.ndjson")),
6525 b"{}\n",
6526 )
6527 .unwrap();
6528 let registry = crate::session::SessionRegistry::new();
6529 let finalizer = finalizer(directory.path());
6530 let outcome = finalizer
6531 .recover_spooled_sessions(®istry, &Default::default())
6532 .await;
6533 assert!(
6534 outcome.is_ok(),
6535 "poisoned sidecar must not fail the outer recover_spooled_sessions; got {outcome:?}"
6536 );
6537 assert!(
6538 registry.get(poison_stem).is_none(),
6539 "poisoned session id must NEVER be installed into the registry"
6540 );
6541 }
6542
6543 #[tokio::test]
6544 async fn signed_close_issues_exactly_one_offline_verifiable_receipt() {
6545 let directory = tempfile::tempdir().unwrap();
6546 let finalizer = finalizer(directory.path());
6547 let session = session(Workflow::Signed);
6548
6549 let first = finalizer
6550 .close_session(Arc::clone(&session), StopReason::SessionClosed)
6551 .await
6552 .unwrap();
6553 let FinalizeOutcome::Receipt { receipt } = first else {
6554 panic!("expected receipt")
6555 };
6556 receipt.verify_embedded().unwrap();
6557 assert!(matches!(
6558 finalizer
6559 .close_session(Arc::clone(&session), StopReason::SessionClosed)
6560 .await
6561 .unwrap(),
6562 FinalizeOutcome::AlreadyClosed
6563 ));
6564 assert_eq!(
6565 session.receipt.lock().as_ref().unwrap().body.receipt_id,
6566 receipt.body.receipt_id
6567 );
6568 }
6569
6570 #[tokio::test]
6571 async fn lifecycle_outbox_retries_the_same_receipt_event() {
6572 let directory = tempfile::tempdir().unwrap();
6573 let bus = Arc::new(FailFirstReceiptBus {
6574 fail: std::sync::atomic::AtomicBool::new(true),
6575 attempts: parking_lot::Mutex::new(Vec::new()),
6576 });
6577 let finalizer = Finalizer::with_bridge(
6578 Arc::new(Ed25519Signer::from_seed(&[17; 32])),
6579 directory.path().to_path_buf(),
6580 Arc::new(Registry::new()),
6581 bus.clone(),
6582 );
6583 let session = session(Workflow::Signed);
6584 assert!(matches!(
6585 finalizer
6586 .close_session(Arc::clone(&session), StopReason::SessionClosed)
6587 .await,
6588 Err(FinalizeError::Bridge { .. })
6589 ));
6590 assert!(session.is_closed());
6591 assert!(
6592 session.try_lease().is_none(),
6593 "artifact commit must keep admission closed"
6594 );
6595 let receipt_id = session.receipt.lock().as_ref().unwrap().body.receipt_id.clone();
6596 let outcome = finalizer
6597 .close_session(Arc::clone(&session), StopReason::SessionClosed)
6598 .await
6599 .unwrap();
6600 let FinalizeOutcome::Receipt { receipt } = outcome else {
6601 panic!("expected receipt")
6602 };
6603 assert_eq!(receipt.body.receipt_id, receipt_id);
6604 let attempts = bus.attempts.lock();
6605 let receipt_events: Vec<_> = attempts
6606 .iter()
6607 .filter(|(topic, _)| topic == "agent.receipt")
6608 .collect();
6609 assert_eq!(receipt_events.len(), 2);
6610 assert_eq!(
6611 receipt_events[0].1["metadata"]["uid"],
6612 receipt_events[1].1["metadata"]["uid"]
6613 );
6614 assert!(!finalizer.lifecycle_outbox_path(&session.id, "receipt").exists());
6615 }
6616
6617 #[tokio::test]
6627 async fn round_43_f1_pending_close_completes_via_reconciler_sweep() {
6628 let directory = tempfile::tempdir().unwrap();
6629 let bus = Arc::new(FailFirstReceiptBus {
6630 fail: std::sync::atomic::AtomicBool::new(true),
6631 attempts: parking_lot::Mutex::new(Vec::new()),
6632 });
6633 let finalizer = Finalizer::with_bridge(
6634 Arc::new(Ed25519Signer::from_seed(&[43; 32])),
6635 directory.path().to_path_buf(),
6636 Arc::new(Registry::new()),
6637 bus.clone(),
6638 );
6639 let sessions = SessionRegistry::new();
6640 let session = session(Workflow::Signed);
6641 sessions.insert_recovered(Arc::try_unwrap(session.clone()).unwrap_or_else(|arc| {
6643 Session::new(
6644 arc.id.clone(),
6645 arc.workflow,
6646 arc.current_identity(),
6647 Default::default(),
6648 )
6649 }));
6650 let registered = sessions.get(&session.id).unwrap();
6653
6654 assert!(matches!(
6660 finalizer
6661 .close_session(Arc::clone(®istered), StopReason::SessionClosed)
6662 .await,
6663 Err(FinalizeError::Bridge { .. })
6664 ));
6665 assert!(
6666 registered.artifact_committed_flag(),
6667 "signed close must have persisted the receipt and marked artifact_committed before the emit_receipt_event failure",
6668 );
6669 assert!(
6670 !registered.close_complete_flag(),
6671 "close_complete must NOT be set yet — the finalization tail did not run",
6672 );
6673 let pending = sessions.pending_close_sessions();
6675 assert_eq!(pending.len(), 1, "sweep must see the orphaned session");
6676 assert_eq!(pending[0].id, registered.id);
6677
6678 finalizer.replay_lifecycle_outboxes().await.unwrap();
6683 let completed = finalizer.complete_pending_closes(&sessions).await.unwrap();
6684 assert_eq!(completed, 1, "sweep must complete the orphan");
6685 assert!(
6686 registered.close_complete_flag(),
6687 "close_complete must be set after the sweep — the pending-close invariant",
6688 );
6689 assert!(
6690 sessions.pending_close_sessions().is_empty(),
6691 "no sessions remain in the pending-close set once completed",
6692 );
6693 let digest = av_core::digest::sha256_hex(registered.id.as_bytes());
6695 let stem = digest.get(..32).unwrap();
6696 assert!(
6697 !directory.path().join(format!("{stem}.session.json")).exists(),
6698 "step journal metadata sidecar must be removed by the completion sweep",
6699 );
6700 }
6701
6702 #[tokio::test]
6709 async fn pending_close_sweep_clears_budget_state_like_normal_close() {
6710 let directory = tempfile::tempdir().unwrap();
6711 let bus = Arc::new(FailFirstReceiptBus {
6712 fail: std::sync::atomic::AtomicBool::new(true),
6713 attempts: parking_lot::Mutex::new(Vec::new()),
6714 });
6715 let store: Arc<dyn av_state::StateStore> = Arc::new(av_state::InMemoryStore::new());
6716 let finalizer = Finalizer::with_bridge(
6717 Arc::new(Ed25519Signer::from_seed(&[43; 32])),
6718 directory.path().to_path_buf(),
6719 Arc::new(Registry::new()),
6720 bus,
6721 )
6722 .with_state_store(Arc::clone(&store));
6723 let sessions = SessionRegistry::new();
6724 let session = session(Workflow::Signed);
6725 sessions.insert_recovered(Session::new(
6726 session.id.clone(),
6727 session.workflow,
6728 session.current_identity(),
6729 Default::default(),
6730 ));
6731 let registered = sessions.get(&session.id).unwrap();
6732 let budget_key = format!(
6733 "{}tool_calls",
6734 av_state::ActionBudget::session_prefix(®istered.id)
6735 );
6736 store.add(&budget_key, 5).unwrap();
6737
6738 assert!(matches!(
6741 finalizer
6742 .close_session(Arc::clone(®istered), StopReason::SessionClosed)
6743 .await,
6744 Err(FinalizeError::Bridge { .. })
6745 ));
6746 assert_eq!(
6747 store.get(&budget_key).unwrap(),
6748 5,
6749 "budget counters must survive a failed close (session may still be retried)",
6750 );
6751
6752 finalizer.replay_lifecycle_outboxes().await.unwrap();
6753 let completed = finalizer.complete_pending_closes(&sessions).await.unwrap();
6754 assert_eq!(completed, 1);
6755 assert!(registered.close_complete_flag());
6756 assert_eq!(
6757 store.get(&budget_key).unwrap(),
6758 0,
6759 "sweep must clear budget counters exactly like close_session_locked",
6760 );
6761 assert_eq!(
6762 finalizer
6763 .metrics
6764 .counter("av_sessions_finalized_total", "Sessions finalized")
6765 .get(),
6766 1,
6767 "sweep completion must count as a finalized session",
6768 );
6769 }
6770
6771 #[tokio::test]
6777 async fn persist_receipt_archives_previous_incarnation_on_recycled_id() {
6778 let directory = tempfile::tempdir().unwrap();
6779 let finalizer = finalizer(directory.path());
6780 let session = session(Workflow::Signed);
6781 let subject = || av_receipts::ReceiptSubject::EventChain {
6782 chain_head: "aa".repeat(32),
6783 event_count: 0,
6784 };
6785 let receipts_dir = directory.path().join("receipts");
6786 let json_files = |dir: &std::path::Path| {
6787 std::fs::read_dir(dir)
6788 .unwrap()
6789 .filter_map(Result::ok)
6790 .filter(|e| e.path().extension().and_then(std::ffi::OsStr::to_str) == Some("json"))
6791 .count()
6792 };
6793
6794 let first = Receipt::issue(
6795 session.receipt_body(subject(), StopReason::SessionClosed),
6796 finalizer.signer.as_ref(),
6797 )
6798 .unwrap();
6799 finalizer.persist_receipt(&session.id, &first).await.unwrap();
6800 finalizer.persist_receipt(&session.id, &first).await.unwrap();
6802 assert_eq!(
6803 json_files(&receipts_dir),
6804 1,
6805 "same-id rewrite must stay idempotent"
6806 );
6807
6808 let second = Receipt::issue(
6810 session.receipt_body(subject(), StopReason::SessionClosed),
6811 finalizer.signer.as_ref(),
6812 )
6813 .unwrap();
6814 assert_ne!(first.body.receipt_id, second.body.receipt_id);
6815 finalizer.persist_receipt(&session.id, &second).await.unwrap();
6816 assert_eq!(
6817 json_files(&receipts_dir),
6818 2,
6819 "first incarnation's receipt must be archived, not overwritten",
6820 );
6821 let primary: Receipt =
6822 serde_json::from_slice(&std::fs::read(finalizer.receipt_path(&session.id)).unwrap()).unwrap();
6823 assert_eq!(primary.body.receipt_id, second.body.receipt_id);
6824 let archived = finalizer
6825 .receipt_path(&session.id)
6826 .with_extension(format!("archived-{}.json", first.body.receipt_id));
6827 let archived: Receipt = serde_json::from_slice(&std::fs::read(&archived).unwrap()).unwrap();
6828 assert_eq!(archived.body.receipt_id, first.body.receipt_id);
6829
6830 std::fs::write(finalizer.receipt_path(&session.id), b"{not json").unwrap();
6832 finalizer.persist_receipt(&session.id, &second).await.unwrap();
6833 assert_eq!(
6834 json_files(&receipts_dir),
6835 3,
6836 "corrupt bytes at the receipt path must be archived, not destroyed",
6837 );
6838 let primary: Receipt =
6839 serde_json::from_slice(&std::fs::read(finalizer.receipt_path(&session.id)).unwrap()).unwrap();
6840 assert_eq!(primary.body.receipt_id, second.body.receipt_id);
6841 }
6842
6843 #[tokio::test]
6857 async fn consolidation_archives_prior_incarnation_atif_on_recycled_unsigned_id() {
6858 use av_events::{EventClass, OcsfEventBuilder, StatusId};
6859
6860 let directory = tempfile::tempdir().unwrap();
6861 let finalizer = finalizer(directory.path());
6862 let signer: Arc<dyn Signer> = Arc::new(Ed25519Signer::from_seed(&[7; 32]));
6863 let journal_key = crate::journal::key_from_signer(signer.as_ref());
6864 let session_id = "recycled-unsigned-consolidation";
6865 let identity = AgentIdentity {
6866 version: "1".to_owned(),
6867 charter: "test".into(),
6868 instance_uid: "instance-1".to_owned(),
6869 ttl_remaining_s: Some(600),
6870 };
6871 let digest = av_core::digest::sha256_hex(session_id.as_bytes());
6872 let stem = &digest[..32];
6873 let final_path = directory.path().join(format!("{stem}.json"));
6874 let step = |message: &str| av_atif::Step {
6875 step_id: 1,
6876 timestamp: Some(av_core::time::now_iso8601()),
6877 source: av_atif::Source::User,
6878 message: serde_json::json!(message),
6879 reasoning_effort: None,
6880 reasoning_content: None,
6881 model_name: None,
6882 tool_calls: None,
6883 observation: None,
6884 metrics: None,
6885 is_copied_context: None,
6886 llm_call_count: None,
6887 extra: None,
6888 };
6889
6890 let agent = av_atif::Agent {
6893 name: "agentvisor-ai-harness".into(),
6894 version: identity.version.clone(),
6895 model_name: None,
6896 tool_definitions: None,
6897 extra: None,
6898 };
6899 let mut prior_builder = av_atif::TrajectoryBuilder::new(agent, Some(session_id.to_owned()));
6900 prior_builder.push_step(step("prior-incarnation")).unwrap();
6901 let mut prior = prior_builder.finish();
6902 prior.trajectory_id = Some("t-old".to_owned());
6903 let prior_bytes = serde_json::to_vec(&prior).unwrap();
6904 std::fs::write(&final_path, &prior_bytes).unwrap();
6905 let prior_provenance = AtifProvenance {
6906 session_id: session_id.to_owned(),
6907 digest: av_core::digest::sha256_hex(&prior_bytes),
6908 };
6909 let sealed = crate::journal::seal(&journal_key, "atif-provenance", 0, &prior_provenance).unwrap();
6910 std::fs::write(final_path.with_extension("atif-auth"), &sealed).unwrap();
6911
6912 let metadata_payload = serde_json::json!({
6915 "journal_version": 2,
6916 "session_id": session_id,
6917 "identity": identity,
6918 "workflow": "unsigned",
6919 });
6920 let metadata_sealed = crate::journal::seal(&journal_key, "metadata", 0, &metadata_payload).unwrap();
6921 let metadata_path = directory.path().join(format!("{stem}.session.json"));
6922 std::fs::write(&metadata_path, &metadata_sealed).unwrap();
6923 let event = OcsfEventBuilder::new(
6924 EventClass::Compression,
6925 session_id.to_owned(),
6926 identity.clone(),
6927 0,
6928 )
6929 .status(StatusId::Success)
6930 .payload(serde_json::json!({}))
6931 .build()
6932 .unwrap();
6933 let event_uid = event.metadata.uid.clone();
6934 let record = crate::worker::ActiveJournalRecord {
6935 event: serde_json::to_value(&event).unwrap(),
6936 identity: identity.clone(),
6937 atif_step: Some(step("second-incarnation")),
6938 tool_calls: 0,
6939 tool_allowed: 0,
6940 prompt_token_correction: 0,
6941 tool_blocked: 0,
6942 prompt_tokens: 0,
6943 completion_tokens: 0,
6944 cached_tokens: 0,
6945 cost_usd_micros: 0,
6946 stop_reason_id: None,
6947 response_attempt: None,
6948 };
6949 let domain = format!("{session_id}:active");
6950 let record_sealed = crate::journal::seal(&journal_key, &domain, 0, &record).unwrap();
6951 let journal_path = directory.path().join(format!("{stem}.events.ndjson"));
6952 let mut journal_line = record_sealed.clone();
6953 journal_line.push(b'\n');
6954 std::fs::write(&journal_path, &journal_line).unwrap();
6955 crate::worker::persist_broker_ack(
6956 directory.path(),
6957 session_id,
6958 &event_uid,
6959 &PublishAck {
6960 topic: EventClass::Compression.topic().to_owned(),
6961 partition: 0,
6962 offset: 1,
6963 },
6964 &journal_key,
6965 )
6966 .await
6967 .unwrap();
6968 let ack_path = directory.path().join(format!("{stem}.acks.ndjson"));
6969 let ack_bytes = std::fs::read(&ack_path).unwrap();
6970
6971 let registry = crate::session::SessionRegistry::new();
6972 finalizer
6973 .consolidate_step_journals(®istry, &Default::default())
6974 .await
6975 .unwrap();
6976
6977 let archived_path = final_path.with_extension("archived-t-old");
6978 let archived: av_atif::Trajectory =
6979 serde_json::from_slice(&std::fs::read(&archived_path).unwrap()).unwrap();
6980 assert_eq!(
6981 archived.trajectory_id.as_deref(),
6982 Some("t-old"),
6983 "prior incarnation's artifact must be archived, not destroyed",
6984 );
6985 let mut archived_sidecar = archived_path.clone();
6986 archived_sidecar.as_mut_os_string().push(".atif-auth");
6987 assert!(
6988 archived_sidecar.exists(),
6989 "stale sidecar must move with the archive"
6990 );
6991 let new_bytes = std::fs::read(&final_path).unwrap();
6992 let consolidated: av_atif::Trajectory = serde_json::from_slice(&new_bytes).unwrap();
6993 assert_ne!(
6994 consolidated.trajectory_id.as_deref(),
6995 Some("t-old"),
6996 "consolidated trajectory keeps its fresh id so the archive branch stays live",
6997 );
6998 let resealed: AtifProvenance = crate::journal::open(
6999 &journal_key,
7000 "atif-provenance",
7001 0,
7002 &std::fs::read(final_path.with_extension("atif-auth")).unwrap(),
7003 )
7004 .unwrap();
7005 assert_eq!(
7006 resealed.digest,
7007 av_core::digest::sha256_hex(&new_bytes),
7008 "primary provenance must re-seal over the consolidated bytes",
7009 );
7010 assert!(
7011 !journal_path.exists(),
7012 "step journal must be cleaned after consolidation"
7013 );
7014 assert!(
7015 !metadata_path.exists(),
7016 "metadata sidecar must be cleaned after consolidation"
7017 );
7018 assert!(
7019 registry.get(session_id).is_none(),
7020 "consolidation claim must be released for the adoption scan",
7021 );
7022
7023 std::fs::write(&metadata_path, &metadata_sealed).unwrap();
7029 std::fs::write(&journal_path, &journal_line).unwrap();
7030 std::fs::write(&ack_path, &ack_bytes).unwrap();
7031 finalizer
7032 .consolidate_step_journals(®istry, &Default::default())
7033 .await
7034 .unwrap();
7035 assert_eq!(
7036 std::fs::read(&final_path).unwrap(),
7037 new_bytes,
7038 "identical-content replay must not rewrite the artifact",
7039 );
7040 let archives = std::fs::read_dir(directory.path())
7041 .unwrap()
7042 .filter_map(Result::ok)
7043 .filter(|e| e.file_name().to_string_lossy().contains("archived-"))
7044 .count();
7045 assert_eq!(
7046 archives, 2,
7047 "replay must not archive again (primary + sidecar from the first pass only)",
7048 );
7049 assert!(!journal_path.exists(), "replayed journal must still be cleaned");
7050 }
7051
7052 #[test]
7060 fn archive_conflicting_atif_preserves_previous_incarnation() {
7061 let directory = tempfile::tempdir().unwrap();
7062 let path = directory.path().join("abc123.json");
7063 archive_conflicting_atif(&path, Some("traj-2"), "s").unwrap();
7065 std::fs::write(&path, br#"{"trajectory_id": "traj-1"}"#).unwrap();
7066 let sidecar = path.with_extension("atif-auth");
7067 std::fs::write(&sidecar, b"sealed-old").unwrap();
7068 let close_marker = path.with_extension("close-complete");
7074 std::fs::write(&close_marker, b"sealed-close").unwrap();
7075 let promote_marker = path.with_extension("promote");
7076 std::fs::write(&promote_marker, b"sealed-promote").unwrap();
7077 archive_conflicting_atif(&path, Some("traj-1"), "s").unwrap();
7079 assert!(path.exists() && sidecar.exists());
7080 archive_conflicting_atif(&path, Some("traj-2"), "s").unwrap();
7082 assert!(!path.exists(), "old trajectory must move off the primary path");
7083 assert!(
7084 !sidecar.exists(),
7085 "stale sidecar must not survive to fail the new incarnation"
7086 );
7087 assert!(
7088 !close_marker.exists(),
7089 "stale close-complete marker must not survive to suppress the new incarnation's close"
7090 );
7091 assert!(
7092 !promote_marker.exists(),
7093 "stale promotion marker must not survive to hard-fail the new incarnation's promote"
7094 );
7095 let archived = path.with_extension("archived-traj-1");
7096 assert!(archived.exists(), "previous incarnation must be preserved");
7097 assert!(
7098 archived.extension().and_then(std::ffi::OsStr::to_str) != Some("json"),
7099 "archived artifact must not match the recovery scan's .json glob"
7100 );
7101 let mut archived_sidecar = archived.clone().into_os_string();
7102 archived_sidecar.push(".atif-auth");
7103 assert_eq!(
7104 std::fs::read(std::path::PathBuf::from(archived_sidecar)).unwrap(),
7105 b"sealed-old",
7106 "sidecar must be archived alongside so the pair stays verifiable"
7107 );
7108 std::fs::write(&path, b"{not json").unwrap();
7110 archive_conflicting_atif(&path, Some("traj-3"), "s").unwrap();
7111 assert!(!path.exists());
7112 let preserved = std::fs::read_dir(directory.path())
7113 .unwrap()
7114 .filter_map(Result::ok)
7115 .filter(|e| {
7116 e.file_name()
7117 .to_str()
7118 .is_some_and(|n| n.starts_with("abc123.archived-corrupt-"))
7119 })
7120 .count();
7121 assert_eq!(preserved, 1, "corrupt bytes must be archived, not destroyed");
7122 }
7123
7124 #[test]
7133 fn archive_conflicting_atif_rescues_orphaned_siblings_of_a_missing_primary() {
7134 let directory = tempfile::tempdir().unwrap();
7135 let path = directory.path().join("abc123.json");
7136 let sidecar = path.with_extension("atif-auth");
7137 let close_marker = path.with_extension("close-complete");
7138 let promote_marker = path.with_extension("promote");
7139 std::fs::write(&sidecar, b"sealed-old").unwrap();
7140 std::fs::write(&close_marker, b"sealed-close").unwrap();
7141 std::fs::write(&promote_marker, b"sealed-promote").unwrap();
7142 archive_conflicting_atif(&path, Some("traj-2"), "s").unwrap();
7144 assert!(
7145 !sidecar.exists(),
7146 "orphaned sidecar must be archived so the recycled id can seal fresh provenance"
7147 );
7148 assert!(
7149 !close_marker.exists(),
7150 "orphaned close-complete marker must be archived"
7151 );
7152 assert!(
7153 !promote_marker.exists(),
7154 "orphaned promotion marker must be archived"
7155 );
7156 let names: Vec<String> = std::fs::read_dir(directory.path())
7157 .unwrap()
7158 .filter_map(Result::ok)
7159 .filter_map(|e| e.file_name().to_str().map(str::to_owned))
7160 .collect();
7161 for marker in [
7162 "orphaned-atif-auth",
7163 "orphaned-close-complete",
7164 "orphaned-promote",
7165 ] {
7166 assert_eq!(
7167 names.iter().filter(|n| n.contains(marker)).count(),
7168 1,
7169 "expected exactly one archived {marker} orphan, got {names:?}"
7170 );
7171 }
7172 for name in &names {
7173 let extension = std::path::Path::new(name)
7174 .extension()
7175 .and_then(std::ffi::OsStr::to_str);
7176 assert_ne!(
7177 extension,
7178 Some("json"),
7179 "orphan archive re-enters the recovery scan: {name}"
7180 );
7181 assert_ne!(
7182 extension,
7183 Some("promote"),
7184 "orphan archive re-enters the promote scan: {name}"
7185 );
7186 }
7187 let clean = directory.path().join("def456.json");
7189 archive_conflicting_atif(&clean, Some("traj-1"), "s").unwrap();
7190 }
7191
7192 #[tokio::test]
7200 async fn acked_receipt_outbox_is_retained_while_promotion_marker_exists() {
7201 let directory = tempfile::tempdir().unwrap();
7202 let finalizer = finalizer(directory.path());
7203 let session_id = "promo-retain";
7204 let topic = av_events::EventClass::Receipt.topic();
7205 let outbox = LifecycleOutbox {
7206 schema_version: LIFECYCLE_OUTBOX_SCHEMA_V1,
7207 session_id: session_id.to_owned(),
7208 kind: crate::journal::RECEIPT_OUTBOX_KIND.to_owned(),
7209 topic: topic.to_owned(),
7210 key: "instance-1".to_owned(),
7211 session_scope: None,
7212 value: serde_json::json!({
7213 "metadata": { "sequence": 0, "uid": av_core::new_event_uid() },
7214 "topic": topic,
7215 }),
7216 ack: Some(av_bridge::PublishAck {
7217 topic: topic.to_owned(),
7218 partition: 0,
7219 offset: 1,
7220 }),
7221 };
7222 let outbox_path = finalizer.lifecycle_outbox_path(session_id, crate::journal::RECEIPT_OUTBOX_KIND);
7223 std::fs::create_dir_all(outbox_path.parent().unwrap()).unwrap();
7224 let sealed = crate::journal::seal(
7225 &finalizer.journal_key,
7226 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
7227 0,
7228 &outbox,
7229 )
7230 .unwrap();
7231 std::fs::write(&outbox_path, sealed).unwrap();
7232 let stem = &av_core::digest::sha256_hex(session_id.as_bytes())[..32];
7233 let marker = directory.path().join(format!("{stem}.promote"));
7234 std::fs::write(&marker, b"pending-promotion").unwrap();
7235
7236 let registry = SessionRegistry::new();
7239 finalizer
7240 .remove_acked_lifecycle_outboxes(®istry)
7241 .await
7242 .unwrap();
7243 assert!(
7244 outbox_path.exists(),
7245 "acked receipt outbox must survive while its promotion marker exists"
7246 );
7247
7248 std::fs::remove_file(&marker).unwrap();
7250 finalizer
7251 .remove_acked_lifecycle_outboxes(®istry)
7252 .await
7253 .unwrap();
7254 assert!(
7255 !outbox_path.exists(),
7256 "with no promotion pending, the orphan acked outbox must be collected"
7257 );
7258 }
7259
7260 #[tokio::test]
7271 async fn recycled_session_id_re_emits_lifecycle_event_over_stale_acked_outbox() {
7272 let directory = tempfile::tempdir().unwrap();
7273 let concrete_bus = Arc::new(FailFirstReceiptBus {
7274 fail: std::sync::atomic::AtomicBool::new(false),
7275 attempts: parking_lot::Mutex::new(Vec::new()),
7276 });
7277 let bus_probe = Arc::clone(&concrete_bus);
7278 let bus: Arc<dyn EventBus> = concrete_bus;
7279 let finalizer = Finalizer::with_bridge(
7280 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
7281 directory.path().to_path_buf(),
7282 Arc::new(Registry::new()),
7283 bus,
7284 );
7285
7286 let session_id = "recycled-id";
7290 let topic = av_events::EventClass::Receipt.topic();
7291 let stale = LifecycleOutbox {
7292 schema_version: LIFECYCLE_OUTBOX_SCHEMA_V1,
7293 session_id: session_id.to_owned(),
7294 kind: crate::journal::RECEIPT_OUTBOX_KIND.to_owned(),
7295 topic: topic.to_owned(),
7296 key: "instance-old".to_owned(),
7297 session_scope: None,
7298 value: serde_json::json!({
7299 "metadata": { "sequence": 0, "uid": av_core::new_event_uid() },
7300 "topic": topic,
7301 }),
7302 ack: Some(av_bridge::PublishAck {
7303 topic: topic.to_owned(),
7304 partition: 0,
7305 offset: 1,
7306 }),
7307 };
7308 let outbox_path = finalizer.lifecycle_outbox_path(session_id, crate::journal::RECEIPT_OUTBOX_KIND);
7309 std::fs::create_dir_all(outbox_path.parent().unwrap()).unwrap();
7310 let sealed = crate::journal::seal(
7311 &finalizer.journal_key,
7312 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
7313 0,
7314 &stale,
7315 )
7316 .unwrap();
7317 std::fs::write(&outbox_path, sealed).unwrap();
7318
7319 let session = Arc::new(Session::new(
7321 session_id.to_owned(),
7322 Workflow::Signed,
7323 AgentIdentity {
7324 version: "1".to_owned(),
7325 charter: "test".into(),
7326 instance_uid: "instance-fresh".to_owned(),
7327 ttl_remaining_s: Some(600),
7328 },
7329 Default::default(),
7330 ));
7331 let payload = serde_json::json!({
7332 "receipt_id": "01936000-0000-7000-8000-000000000001",
7333 "chain_head": "0000",
7334 });
7335
7336 finalizer
7337 .emit_bridge_event(
7338 &session,
7339 av_events::EventClass::Receipt,
7340 payload,
7341 crate::journal::RECEIPT_OUTBOX_KIND,
7342 )
7343 .await
7344 .expect("recycled-id emit succeeds");
7345
7346 let attempts = bus_probe.attempts.lock();
7348 assert!(
7349 attempts.iter().any(|(t, _)| t == topic),
7350 "the fresh incarnation MUST publish a receipt lifecycle event even when a stale \
7351 acked outbox for the recycled id exists (pre-R28 this silently returned Ok(()) \
7352 without publishing, dropping the audit event); got attempts: {attempts:?}"
7353 );
7354
7355 let bytes = std::fs::read(&outbox_path).unwrap();
7358 let outbox: LifecycleOutbox = crate::journal::open(
7359 &finalizer.journal_key,
7360 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
7361 0,
7362 &bytes,
7363 )
7364 .unwrap();
7365 assert_eq!(
7366 outbox.key, "instance-fresh",
7367 "persisted outbox key must reflect the fresh incarnation's instance_uid; \
7368 stale key would let the next tick's `remove_acked_lifecycle_outboxes` see \
7369 it as a valid ack and never re-publish"
7370 );
7371 }
7372
7373 #[tokio::test]
7381 async fn same_instance_recycled_session_id_re_emits_lifecycle_event() {
7382 let directory = tempfile::tempdir().unwrap();
7383 let concrete_bus = Arc::new(FailFirstReceiptBus {
7384 fail: std::sync::atomic::AtomicBool::new(false),
7385 attempts: parking_lot::Mutex::new(Vec::new()),
7386 });
7387 let bus_probe = Arc::clone(&concrete_bus);
7388 let bus: Arc<dyn EventBus> = concrete_bus;
7389 let finalizer = Finalizer::with_bridge(
7390 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
7391 directory.path().to_path_buf(),
7392 Arc::new(Registry::new()),
7393 bus,
7394 );
7395 let session_id = "same-instance-recycled";
7396 let identity = AgentIdentity {
7397 version: "1".to_owned(),
7398 charter: "test".into(),
7399 instance_uid: "instance-stable".to_owned(),
7400 ttl_remaining_s: Some(600),
7401 };
7402 let topic = av_events::EventClass::Receipt.topic();
7403 let stale = LifecycleOutbox {
7406 schema_version: LIFECYCLE_OUTBOX_SCHEMA_V1,
7407 session_id: session_id.to_owned(),
7408 kind: crate::journal::RECEIPT_OUTBOX_KIND.to_owned(),
7409 topic: topic.to_owned(),
7410 key: "instance-stable".to_owned(),
7411 session_scope: Some(format!("{session_id}#previous-incarnation")),
7412 value: serde_json::json!({
7413 "metadata": { "sequence": 0, "uid": av_core::new_event_uid() },
7414 "topic": topic,
7415 }),
7416 ack: Some(av_bridge::PublishAck {
7417 topic: topic.to_owned(),
7418 partition: 0,
7419 offset: 1,
7420 }),
7421 };
7422 let outbox_path = finalizer.lifecycle_outbox_path(session_id, crate::journal::RECEIPT_OUTBOX_KIND);
7423 std::fs::create_dir_all(outbox_path.parent().unwrap()).unwrap();
7424 let sealed = crate::journal::seal(
7425 &finalizer.journal_key,
7426 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
7427 0,
7428 &stale,
7429 )
7430 .unwrap();
7431 std::fs::write(&outbox_path, sealed).unwrap();
7432
7433 let session = Arc::new(Session::new(
7435 session_id.to_owned(),
7436 Workflow::Signed,
7437 identity,
7438 Default::default(),
7439 ));
7440 finalizer
7441 .emit_bridge_event(
7442 &session,
7443 av_events::EventClass::Receipt,
7444 serde_json::json!({
7445 "receipt_id": "01936000-0000-7000-8000-000000000002",
7446 "chain_head": "0000",
7447 }),
7448 crate::journal::RECEIPT_OUTBOX_KIND,
7449 )
7450 .await
7451 .expect("same-instance recycled emit succeeds");
7452 let attempts = bus_probe.attempts.lock();
7453 assert!(
7454 attempts.iter().any(|(t, _)| t == topic),
7455 "a stale acked outbox from the SAME instance_uid must not swallow the new \
7456 incarnation's lifecycle event; got attempts: {attempts:?}"
7457 );
7458 let bytes = std::fs::read(&outbox_path).unwrap();
7459 let outbox: LifecycleOutbox = crate::journal::open(
7460 &finalizer.journal_key,
7461 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
7462 0,
7463 &bytes,
7464 )
7465 .unwrap();
7466 assert_eq!(
7467 outbox.session_scope.as_deref(),
7468 Some(session.session_scope.as_str()),
7469 "persisted outbox must carry the fresh incarnation's session_scope"
7470 );
7471 }
7472
7473 #[tokio::test]
7483 async fn round_44_f1_sidecar_less_atif_is_quarantined_without_reading_bytes() {
7484 let directory = tempfile::tempdir().unwrap();
7485 let finalizer = finalizer(directory.path());
7486 let orphan = directory.path().join("hostileplant0000000000000000.json");
7494 std::fs::write(&orphan, b"{not valid json at all - this MUST NOT be parsed").unwrap();
7495 let old = std::time::SystemTime::now() - std::time::Duration::from_secs(120);
7501 let file = std::fs::OpenOptions::new().append(true).open(&orphan).unwrap();
7502 file.set_times(std::fs::FileTimes::new().set_modified(old))
7503 .unwrap();
7504 drop(file);
7505
7506 let registry = SessionRegistry::new();
7507 let outcome = finalizer
7508 .recover_spooled_sessions(®istry, &Default::default())
7509 .await;
7510 assert!(
7511 outcome.is_ok(),
7512 "orphan must not fail the outer scan; got {outcome:?}"
7513 );
7514
7515 assert!(
7520 !orphan.exists(),
7521 "sidecar-less ATIF must be quarantined-renamed after first sighting so subsequent ticks don't re-read it",
7522 );
7523 let mut found_quarantine = false;
7526 for entry in std::fs::read_dir(directory.path()).unwrap() {
7527 let entry = entry.unwrap();
7528 if let Some(name) = entry.file_name().to_str() {
7529 if name.contains(".json.corrupt-") {
7530 found_quarantine = true;
7531 }
7532 }
7533 }
7534 assert!(
7535 found_quarantine,
7536 "quarantined file must be preserved on disk under a `.corrupt-<uid>` name for forensic inspection",
7537 );
7538 }
7539
7540 #[tokio::test]
7550 async fn fresh_sidecar_less_atif_is_not_quarantined_out_from_under_a_live_close() {
7551 let directory = tempfile::tempdir().unwrap();
7552 let finalizer = finalizer(directory.path());
7553 let fresh = directory.path().join("freshinflightclose0000000000.json");
7554 std::fs::write(&fresh, b"{}").unwrap();
7555
7556 let registry = SessionRegistry::new();
7557 finalizer
7558 .recover_spooled_sessions(®istry, &Default::default())
7559 .await
7560 .unwrap();
7561 assert!(
7562 fresh.exists(),
7563 "a fresh sidecar-less artifact (in-flight close window) must not be quarantined"
7564 );
7565 let quarantined = std::fs::read_dir(directory.path())
7566 .unwrap()
7567 .filter_map(Result::ok)
7568 .any(|entry| {
7569 entry
7570 .file_name()
7571 .to_str()
7572 .is_some_and(|name| name.contains(".corrupt-"))
7573 });
7574 assert!(
7575 !quarantined,
7576 "no quarantine sibling may be created for a fresh file"
7577 );
7578 }
7579
7580 #[tokio::test]
7589 async fn live_session_stem_survives_orphan_quarantine_even_when_aged() {
7590 let directory = tempfile::tempdir().unwrap();
7591 let finalizer = finalizer(directory.path());
7592
7593 let live = session(Workflow::Unsigned);
7594 let stem = av_core::digest::sha256_hex(live.id.as_bytes())[..32].to_owned();
7595 let inflight = directory.path().join(format!("{stem}.json"));
7596 std::fs::write(&inflight, b"{}").unwrap();
7597 let old = std::time::SystemTime::now() - std::time::Duration::from_secs(120);
7600 let file = std::fs::OpenOptions::new().append(true).open(&inflight).unwrap();
7601 file.set_times(std::fs::FileTimes::new().set_modified(old))
7602 .unwrap();
7603 drop(file);
7604
7605 let registry = SessionRegistry::new();
7606 registry.insert_recovered(Session::new(
7607 live.id.clone(),
7608 Workflow::Unsigned,
7609 AgentIdentity {
7610 version: "1".to_owned(),
7611 charter: "test".into(),
7612 instance_uid: "instance-1".to_owned(),
7613 ttl_remaining_s: Some(600),
7614 },
7615 Default::default(),
7616 ));
7617
7618 finalizer
7619 .recover_spooled_sessions(®istry, &Default::default())
7620 .await
7621 .unwrap();
7622
7623 assert!(
7624 inflight.exists(),
7625 "an in-progress close's fresh `.json` must not be quarantined even after crossing the age threshold, when the stem belongs to a live session"
7626 );
7627 let quarantined = std::fs::read_dir(directory.path())
7628 .unwrap()
7629 .filter_map(Result::ok)
7630 .any(|entry| {
7631 entry
7632 .file_name()
7633 .to_str()
7634 .is_some_and(|name| name.contains(".corrupt-"))
7635 });
7636 assert!(
7637 !quarantined,
7638 "no `.corrupt-*` sibling may be produced when the stem belongs to a live session"
7639 );
7640 }
7641
7642 #[tokio::test]
7651 async fn registered_session_pair_is_skipped_without_reading_bytes() {
7652 let directory = tempfile::tempdir().unwrap();
7653 let finalizer = finalizer(directory.path());
7654
7655 let live = session(Workflow::Unsigned);
7656 let stem = av_core::digest::sha256_hex(live.id.as_bytes())[..32].to_owned();
7657 std::fs::write(
7658 directory.path().join(format!("{stem}.json")),
7659 b"{this is not json and must never be parsed on a steady-state tick",
7660 )
7661 .unwrap();
7662 std::fs::write(directory.path().join(format!("{stem}.atif-auth")), b"sidecar").unwrap();
7663
7664 let registry = SessionRegistry::new();
7665 registry.insert_recovered(Session::new(
7666 live.id.clone(),
7667 Workflow::Unsigned,
7668 AgentIdentity {
7669 version: "1".to_owned(),
7670 charter: "test".into(),
7671 instance_uid: "instance-1".to_owned(),
7672 ttl_remaining_s: Some(600),
7673 },
7674 Default::default(),
7675 ));
7676
7677 finalizer
7678 .recover_spooled_sessions(®istry, &Default::default())
7679 .await
7680 .unwrap();
7681
7682 let invalid_json_skips = finalizer
7683 .metrics
7684 .counter(
7685 "av_atif_recovery_skipped_total{reason=\"invalid_json\"}",
7686 "ATIF spool files skipped during recovery",
7687 )
7688 .get();
7689 assert_eq!(
7690 invalid_json_skips, 0,
7691 "a registered session's sealed pair must be skipped before the bytes are read — steady-state ticks must not pay O(file) per recovered session"
7692 );
7693 }
7694
7695 #[tokio::test]
7708 async fn round_44_f2_empty_unsigned_quarantine_excluded_from_pending_close_sweep() {
7709 let directory = tempfile::tempdir().unwrap();
7710 let finalizer = finalizer(directory.path());
7711 let sessions = SessionRegistry::new();
7712 let empty = session(Workflow::Unsigned);
7713 let empty_id = empty.id.clone();
7715 sessions.insert_recovered(Arc::try_unwrap(empty).unwrap_or_else(|arc| {
7716 Session::new(
7717 arc.id.clone(),
7718 arc.workflow,
7719 arc.current_identity(),
7720 Default::default(),
7721 )
7722 }));
7723 let registered = sessions.get(&empty_id).unwrap();
7724
7725 let result = finalizer
7732 .close_session(Arc::clone(®istered), StopReason::SessionClosed)
7733 .await;
7734 assert!(
7735 matches!(result, Err(FinalizeError::Atif { .. })),
7736 "empty unsigned close must reject; got {result:?}",
7737 );
7738 assert!(
7739 registered.artifact_committed_flag(),
7740 "empty-unsigned reject seals with artifact_committed to stop idle-sweep churn",
7741 );
7742 assert!(
7743 registered.is_empty_unsigned_quarantine(),
7744 "empty-unsigned reject must be recognizable as a quarantine",
7745 );
7746
7747 let pending = sessions.pending_close_sessions();
7753 assert!(
7754 pending.iter().all(|s| s.id != empty_id),
7755 "empty-unsigned quarantine must be excluded from pending_close_sessions()",
7756 );
7757 let completed = finalizer.complete_pending_closes(&sessions).await.unwrap();
7758 assert_eq!(
7759 completed, 0,
7760 "sweep must NOT complete the empty-unsigned quarantine — evidence would be lost",
7761 );
7762 assert!(
7763 !registered.close_complete_flag(),
7764 "close_complete must remain 0 for the empty-unsigned quarantine so it can NOT be reopened by get_or_open (reopen=true) — preserving the incident record",
7765 );
7766 }
7767
7768 #[tokio::test]
7769 async fn concurrent_close_waits_for_failed_lifecycle_attempt() {
7770 let directory = tempfile::tempdir().unwrap();
7771 let bus = Arc::new(FailFirstReceiptBus {
7772 fail: std::sync::atomic::AtomicBool::new(true),
7773 attempts: parking_lot::Mutex::new(Vec::new()),
7774 });
7775 let finalizer = Finalizer::with_bridge(
7776 Arc::new(Ed25519Signer::from_seed(&[23; 32])),
7777 directory.path().to_path_buf(),
7778 Arc::new(Registry::new()),
7779 bus,
7780 );
7781 let session = session(Workflow::Signed);
7782 let first_finalizer = finalizer.clone();
7783 let first_session = Arc::clone(&session);
7784 let first = tokio::spawn(async move {
7785 first_finalizer
7786 .close_session(first_session, StopReason::SessionClosed)
7787 .await
7788 });
7789 tokio::task::yield_now().await;
7790 let second = finalizer.close_session(session, StopReason::SessionClosed).await;
7791 let first = first.await.unwrap();
7792 assert!(matches!(first, Err(FinalizeError::Bridge { .. })));
7793 assert!(matches!(second, Ok(FinalizeOutcome::Receipt { .. })));
7794 }
7795
7796 #[tokio::test]
7797 async fn startup_replays_lifecycle_outbox_without_session_journal() {
7798 let directory = tempfile::tempdir().unwrap();
7799 let bus = Arc::new(FailFirstReceiptBus {
7800 fail: std::sync::atomic::AtomicBool::new(true),
7801 attempts: parking_lot::Mutex::new(Vec::new()),
7802 });
7803 let finalizer = Finalizer::with_bridge(
7804 Arc::new(Ed25519Signer::from_seed(&[19; 32])),
7805 directory.path().to_path_buf(),
7806 Arc::new(Registry::new()),
7807 bus.clone(),
7808 );
7809 let session = session(Workflow::Signed);
7810 assert!(matches!(
7811 finalizer.close_session(session, StopReason::SessionClosed).await,
7812 Err(FinalizeError::Bridge { .. })
7813 ));
7814 assert!(finalizer
7815 .lifecycle_outbox_path("lifecycle-session", "receipt")
7816 .exists());
7817
7818 let sessions = SessionRegistry::new();
7819 assert_eq!(
7820 finalizer
7821 .recover_spooled_sessions(&sessions, &Default::default())
7822 .await
7823 .unwrap(),
7824 0
7825 );
7826 assert!(sessions.get("lifecycle-session").is_none());
7827 assert!(!finalizer
7828 .lifecycle_outbox_path("lifecycle-session", "receipt")
7829 .exists());
7830 assert_eq!(
7831 bus.attempts
7832 .lock()
7833 .iter()
7834 .filter(|(topic, _)| topic == "agent.receipt")
7835 .count(),
7836 2
7837 );
7838 }
7839
7840 #[tokio::test]
7841 async fn unsigned_close_and_promotion_are_strict_and_idempotent() {
7842 let directory = tempfile::tempdir().unwrap();
7843 let finalizer = finalizer(directory.path());
7844 let session = session(Workflow::Unsigned);
7845 session
7846 .atif
7847 .lock()
7848 .push_step(av_atif::Step {
7849 step_id: 0,
7850 timestamp: Some(av_core::time::now_iso8601()),
7851 source: av_atif::Source::Agent,
7852 message: serde_json::json!("done"),
7853 reasoning_effort: None,
7854 reasoning_content: None,
7855 model_name: None,
7856 tool_calls: None,
7857 observation: None,
7858 metrics: Some(av_atif::Metrics {
7859 prompt_tokens: Some(10),
7860 completion_tokens: Some(2),
7861 cached_tokens: Some(4),
7862 cost_usd: Some(0.001),
7863 logprobs: None,
7864 completion_token_ids: None,
7865 prompt_token_ids: None,
7866 extra: None,
7867 }),
7868 is_copied_context: None,
7869 llm_call_count: Some(1),
7870 extra: None,
7871 })
7872 .unwrap();
7873
7874 let outcome = finalizer
7875 .close_session(Arc::clone(&session), StopReason::SessionClosed)
7876 .await
7877 .unwrap();
7878 let FinalizeOutcome::Atif { path } = outcome else {
7879 panic!("expected ATIF artifact")
7880 };
7881 let value: serde_json::Value = serde_json::from_slice(&tokio::fs::read(path).await.unwrap()).unwrap();
7882 assert!(av_atif::validate_value(&value, av_atif::Mode::Strict).is_empty());
7883
7884 let first = finalizer.promote(Arc::clone(&session)).await.unwrap();
7885 let second = finalizer.promote(Arc::clone(&session)).await.unwrap();
7886 assert_eq!(first.body.receipt_id, second.body.receipt_id);
7887 first.verify_embedded().unwrap();
7888 assert!(matches!(
7889 first.body.subject,
7890 ReceiptSubject::AtifTrajectory {
7891 step_count: 1,
7892 retroactive: true,
7893 ..
7894 }
7895 ));
7896 }
7897
7898 #[tokio::test]
7899 async fn unsigned_restart_preserves_receipt_accounting_and_identity() {
7900 let directory = tempfile::tempdir().unwrap();
7901 let finalizer = finalizer(directory.path());
7902 let original = session(Workflow::Unsigned);
7903 original
7904 .totals
7905 .tool_calls
7906 .store(2, std::sync::atomic::Ordering::Release);
7907 original
7908 .totals
7909 .tool_allowed
7910 .store(1, std::sync::atomic::Ordering::Release);
7911 original
7912 .totals
7913 .tool_blocked
7914 .store(1, std::sync::atomic::Ordering::Release);
7915 original
7916 .totals
7917 .prompt_tokens
7918 .store(17, std::sync::atomic::Ordering::Release);
7919 original
7920 .totals
7921 .completion_tokens
7922 .store(9, std::sync::atomic::Ordering::Release);
7923 original
7924 .totals
7925 .cached_tokens
7926 .store(3, std::sync::atomic::Ordering::Release);
7927 original
7928 .totals
7929 .cost_usd_micros
7930 .store(1_234_567, std::sync::atomic::Ordering::Release);
7931 original.record_stop_reason(StopReason::PolicyBlocked);
7932 original
7933 .atif
7934 .lock()
7935 .push_step(av_atif::Step {
7936 step_id: 0,
7937 timestamp: None,
7938 source: av_atif::Source::User,
7939 message: serde_json::json!("test"),
7940 reasoning_effort: None,
7941 reasoning_content: None,
7942 model_name: None,
7943 tool_calls: None,
7944 observation: None,
7945 metrics: None,
7946 is_copied_context: None,
7947 llm_call_count: None,
7948 extra: None,
7949 })
7950 .unwrap();
7951 finalizer
7952 .close_session(original, StopReason::SessionClosed)
7953 .await
7954 .unwrap();
7955
7956 let registry = SessionRegistry::new();
7957 finalizer
7958 .recover_spooled_sessions(®istry, &Default::default())
7959 .await
7960 .unwrap();
7961 let recovered = registry.get("lifecycle-session").unwrap();
7962 assert_eq!(recovered.current_identity().ttl_remaining_s, Some(600));
7963 let receipt = finalizer.promote(recovered).await.unwrap();
7964 assert_eq!(receipt.body.tool_calls.total, 2);
7965 assert_eq!(receipt.body.tool_calls.allowed, 1);
7966 assert_eq!(receipt.body.tool_calls.blocked, 1);
7967 assert_eq!(receipt.body.cost.prompt_tokens, 17);
7968 assert_eq!(receipt.body.cost.completion_tokens, 9);
7969 assert_eq!(receipt.body.cost.cached_tokens, 3);
7970 assert_eq!(receipt.body.cost.cost_usd_micros, 1_234_567);
7971 assert_eq!(receipt.body.stop_reason_id, StopReason::PolicyBlocked.id());
7972 }
7973
7974 #[tokio::test]
7986 async fn restart_does_not_reemit_session_close_for_finalized_unsigned() {
7987 let directory = tempfile::tempdir().unwrap();
7988 let bus = Arc::new(FailFirstReceiptBus {
7989 fail: std::sync::atomic::AtomicBool::new(false),
7990 attempts: parking_lot::Mutex::new(Vec::new()),
7991 });
7992 let finalizer = Finalizer::with_bridge(
7993 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
7994 directory.path().to_path_buf(),
7995 Arc::new(Registry::new()),
7996 bus.clone(),
7997 );
7998 let original = session(Workflow::Unsigned);
7999 original
8000 .atif
8001 .lock()
8002 .push_step(av_atif::Step {
8003 step_id: 0,
8004 timestamp: None,
8005 source: av_atif::Source::User,
8006 message: serde_json::json!("test"),
8007 reasoning_effort: None,
8008 reasoning_content: None,
8009 model_name: None,
8010 tool_calls: None,
8011 observation: None,
8012 metrics: None,
8013 is_copied_context: None,
8014 llm_call_count: None,
8015 extra: None,
8016 })
8017 .unwrap();
8018 finalizer
8019 .close_session(original, StopReason::SessionClosed)
8020 .await
8021 .unwrap();
8022 let session_topic = av_events::EventClass::Session.topic();
8023 let closes_after_close = bus
8024 .attempts
8025 .lock()
8026 .iter()
8027 .filter(|(topic, _)| topic == session_topic)
8028 .count();
8029 assert_eq!(closes_after_close, 1, "exactly one close on the wire");
8030 let stem = &av_core::digest::sha256_hex("lifecycle-session".as_bytes())[..32];
8031 let marker_path = directory.path().join(format!("{stem}.close-complete"));
8032 assert!(
8033 marker_path.exists(),
8034 "the close tail must persist the close-complete marker"
8035 );
8036
8037 let registry = SessionRegistry::new();
8039 finalizer
8040 .recover_spooled_sessions(®istry, &Default::default())
8041 .await
8042 .unwrap();
8043 let recovered = registry.get("lifecycle-session").unwrap();
8044 assert!(
8045 recovered.close_complete_flag(),
8046 "recovery must restore close_complete from the sealed marker"
8047 );
8048 assert!(
8049 registry.pending_close_sessions().is_empty(),
8050 "a finalized Unsigned session must not re-enter the pending-close sweep"
8051 );
8052 finalizer.complete_pending_closes(®istry).await.unwrap();
8053 let closes_after_restart = bus
8054 .attempts
8055 .lock()
8056 .iter()
8057 .filter(|(topic, _)| topic == session_topic)
8058 .count();
8059 assert_eq!(
8060 closes_after_restart, 1,
8061 "restart must not publish a duplicate SESSION_CLOSE for a finalized session"
8062 );
8063
8064 std::fs::remove_file(&marker_path).unwrap();
8070 let registry = SessionRegistry::new();
8071 finalizer
8072 .recover_spooled_sessions(®istry, &Default::default())
8073 .await
8074 .unwrap();
8075 let recovered = registry.get("lifecycle-session").unwrap();
8076 assert!(
8077 !recovered.close_complete_flag(),
8078 "without the marker, recovery must leave the session to the sweep \
8079 (its close may never have been published)"
8080 );
8081 finalizer.complete_pending_closes(®istry).await.unwrap();
8082 let closes_after_sweep = bus
8083 .attempts
8084 .lock()
8085 .iter()
8086 .filter(|(topic, _)| topic == session_topic)
8087 .count();
8088 assert_eq!(
8089 closes_after_sweep, 2,
8090 "the sweep must emit the close exactly once"
8091 );
8092 assert!(
8093 marker_path.exists(),
8094 "the sweep must persist the marker after emitting the close"
8095 );
8096
8097 std::fs::write(&marker_path, b"garbage").unwrap();
8100 let registry = SessionRegistry::new();
8101 finalizer
8102 .recover_spooled_sessions(®istry, &Default::default())
8103 .await
8104 .unwrap();
8105 let recovered = registry.get("lifecycle-session").unwrap();
8106 assert!(
8107 !recovered.close_complete_flag(),
8108 "a marker that fails MAC/digest verification must not restore close_complete"
8109 );
8110 assert!(
8111 !marker_path.exists(),
8112 "an invalid marker must be removed so the sweep can rewrite it"
8113 );
8114 }
8115
8116 #[tokio::test]
8117 async fn incomplete_capture_never_produces_receipt_or_atif() {
8118 let directory = tempfile::tempdir().unwrap();
8119 let finalizer = finalizer(directory.path());
8120 for workflow in [Workflow::Signed, Workflow::Unsigned] {
8121 let session = session(workflow);
8122 session.mark_capture_failed();
8123 assert!(matches!(
8124 finalizer
8125 .close_session(Arc::clone(&session), StopReason::SessionClosed)
8126 .await,
8127 Err(FinalizeError::CaptureIncomplete)
8128 ));
8129 assert!(session.receipt.lock().is_none());
8130 assert!(session.atif_path.lock().is_none());
8131 }
8132 }
8133
8134 #[tokio::test]
8148 async fn close_session_seals_capture_failed_session_so_idle_sweep_stops_churning() {
8149 let directory = tempfile::tempdir().unwrap();
8150 let finalizer = finalizer(directory.path());
8151 for workflow in [Workflow::Signed, Workflow::Unsigned] {
8152 let session = session(workflow);
8153 session.mark_capture_failed();
8154 let result = finalizer
8155 .close_session(Arc::clone(&session), StopReason::SessionClosed)
8156 .await;
8157 assert!(matches!(result, Err(FinalizeError::CaptureIncomplete)));
8158 assert!(
8159 session.is_closed(),
8160 "close_session_locked must seal a capture_failed session (mark_artifact_committed + claim.committed = true) on the CaptureIncomplete return path so subsequent idle-sweep passes skip it via the `!is_closed()` filter — otherwise CloseClaim drops unarmed, reset_close puts `closed` back to 0, is_closed() stays false, and the idle sweeper churns forever on this session (workflow: {workflow:?})",
8161 );
8162 assert!(session.receipt.lock().is_none());
8163 assert!(session.atif_path.lock().is_none());
8164 }
8165 }
8166
8167 #[tokio::test]
8187 async fn close_session_seals_empty_unsigned_session_so_idle_sweep_stops_churning() {
8188 let directory = tempfile::tempdir().unwrap();
8189 let finalizer = finalizer(directory.path());
8190 let session = session(Workflow::Unsigned);
8191 assert_eq!(
8192 session.atif.lock().clone().finish().steps.len(),
8193 0,
8194 "precondition: session has no captured steps",
8195 );
8196 let result = finalizer
8197 .close_session(Arc::clone(&session), StopReason::SessionClosed)
8198 .await;
8199 assert!(
8200 matches!(result, Err(FinalizeError::Atif { .. })),
8201 "empty unsigned close must surface an ATIF error to the caller: {result:?}",
8202 );
8203 assert!(
8204 session.is_closed(),
8205 "close_session_locked must seal an empty unsigned session (mark_artifact_committed + claim.committed = true) when write_atomic's strict validation rejects the empty trajectory — otherwise CloseClaim drops unarmed, reset_close puts `closed` back to 0, is_closed() stays false, and the idle sweeper churns forever on this session (write_atomic → validate → \"must contain at least one step\" → Err → reset_close → picked up next tick → repeat)",
8206 );
8207 assert!(
8208 session.atif_path.lock().is_none(),
8209 "no ATIF file was ever produced for the empty session",
8210 );
8211 }
8212
8213 #[tokio::test]
8221 async fn recovery_tick_does_not_quarantine_live_sessions_with_inflight_markers() {
8222 let directory = tempfile::tempdir().unwrap();
8223 let finalizer = finalizer(directory.path());
8224 let live = session(Workflow::Unsigned);
8225 let registry = crate::session::SessionRegistry::new();
8226 let live = registry.insert_recovered(Arc::try_unwrap(live).map_err(|_| ()).unwrap());
8227 crate::worker::create_response_marker(
8228 directory.path(),
8229 &finalizer.journal_key,
8230 &live.id,
8231 "digest".to_owned(),
8232 )
8233 .await
8234 .unwrap();
8235 finalizer
8236 .recover_spooled_sessions(®istry, &Default::default())
8237 .await
8238 .unwrap();
8239 assert!(
8240 !finalizer.quarantined_sessions.lock().contains(&live.id),
8241 "a live session's in-flight marker must not put it in quarantine",
8242 );
8243 assert!(
8244 !live.capture_failed(),
8245 "a recovery tick must not poison a live session mid-request",
8246 );
8247 }
8248
8249 #[tokio::test]
8258 async fn live_session_with_id_in_quarantine_set_can_still_close() {
8259 let directory = tempfile::tempdir().unwrap();
8260 let finalizer = finalizer(directory.path());
8261 finalizer
8263 .quarantined_sessions
8264 .lock()
8265 .insert("lifecycle-session".to_owned());
8266 let live = session(Workflow::Unsigned);
8267 assert!(!live.capture_failed(), "precondition: live session is clean");
8268 live.atif
8270 .lock()
8271 .push_step(av_atif::Step {
8272 step_id: 0,
8273 timestamp: None,
8274 source: av_atif::Source::Agent,
8275 message: serde_json::json!("live response"),
8276 reasoning_effort: None,
8277 reasoning_content: None,
8278 model_name: Some("test-model".into()),
8279 tool_calls: None,
8280 observation: None,
8281 metrics: Some(av_atif::Metrics {
8282 prompt_tokens: Some(1),
8283 completion_tokens: Some(1),
8284 cached_tokens: Some(0),
8285 cost_usd: Some(0.0),
8286 logprobs: None,
8287 completion_token_ids: None,
8288 prompt_token_ids: None,
8289 extra: None,
8290 }),
8291 is_copied_context: None,
8292 llm_call_count: Some(1),
8293 extra: None,
8294 })
8295 .unwrap();
8296 let outcome = finalizer
8297 .close_session(Arc::clone(&live), StopReason::SessionClosed)
8298 .await
8299 .expect("live session must finalize despite id sharing space with a set entry");
8300 match outcome {
8301 FinalizeOutcome::Atif { .. } => {}
8302 other => panic!("expected FinalizeOutcome::Atif, got {other:?}"),
8303 }
8304 assert!(
8305 !live.capture_failed(),
8306 "close_session must not poison the live session's capture_failed flag",
8307 );
8308 }
8309
8310 #[tokio::test]
8317 async fn recovery_does_not_clobber_a_live_session_with_the_same_id() {
8318 let directory = tempfile::tempdir().unwrap();
8319 let finalizer = finalizer(directory.path());
8320 let closed_session = session(Workflow::Unsigned);
8322 closed_session
8323 .atif
8324 .lock()
8325 .push_step(av_atif::Step {
8326 step_id: 0,
8327 timestamp: None,
8328 source: av_atif::Source::Agent,
8329 message: serde_json::json!("archived response"),
8330 reasoning_effort: None,
8331 reasoning_content: None,
8332 model_name: Some("test-model".into()),
8333 tool_calls: None,
8334 observation: None,
8335 metrics: Some(av_atif::Metrics {
8336 prompt_tokens: Some(1),
8337 completion_tokens: Some(1),
8338 cached_tokens: Some(0),
8339 cost_usd: Some(0.0),
8340 logprobs: None,
8341 completion_token_ids: None,
8342 prompt_token_ids: None,
8343 extra: None,
8344 }),
8345 is_copied_context: None,
8346 llm_call_count: Some(1),
8347 extra: None,
8348 })
8349 .unwrap();
8350 finalizer
8351 .close_session(Arc::clone(&closed_session), StopReason::SessionClosed)
8352 .await
8353 .unwrap();
8354 let registry = SessionRegistry::new();
8356 let live = registry.get_or_open(
8357 "lifecycle-session",
8358 Workflow::Unsigned,
8359 &AgentIdentity {
8360 version: "1".to_owned(),
8361 charter: "test".into(),
8362 instance_uid: "instance-1".to_owned(),
8363 ttl_remaining_s: Some(600),
8364 },
8365 &Default::default(),
8366 );
8367 assert!(!live.is_closed(), "precondition: live session is open");
8368 assert!(live.receipt.lock().is_none(), "precondition: no receipt yet");
8369 assert!(live.atif_path.lock().is_none(), "precondition: no atif path yet");
8370 finalizer
8372 .recover_spooled_sessions(®istry, &Default::default())
8373 .await
8374 .unwrap();
8375 assert!(
8376 !live.is_closed(),
8377 "live session must not be force-closed by recovery"
8378 );
8379 assert!(
8380 live.receipt.lock().is_none(),
8381 "live session's receipt must not be overwritten from the stale artifact",
8382 );
8383 assert!(
8384 live.atif_path.lock().is_none(),
8385 "live session's atif_path must not be reassigned to the stale artifact",
8386 );
8387 assert_eq!(registry.len(), 1, "recovery must not add a duplicate entry");
8388 }
8389
8390 #[tokio::test]
8398 async fn recovery_skips_tampered_artifact_and_still_recovers_the_rest() {
8399 let directory = tempfile::tempdir().unwrap();
8400 let metrics = Arc::new(Registry::new());
8401 let finalizer = Finalizer::new(
8402 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
8403 directory.path().to_path_buf(),
8404 Arc::clone(&metrics),
8405 );
8406 let step = av_atif::Step {
8407 step_id: 0,
8408 timestamp: None,
8409 source: av_atif::Source::Agent,
8410 message: serde_json::json!("archived response"),
8411 reasoning_effort: None,
8412 reasoning_content: None,
8413 model_name: Some("test-model".into()),
8414 tool_calls: None,
8415 observation: None,
8416 metrics: Some(av_atif::Metrics {
8417 prompt_tokens: Some(1),
8418 completion_tokens: Some(1),
8419 cached_tokens: Some(0),
8420 cost_usd: Some(0.0),
8421 logprobs: None,
8422 completion_token_ids: None,
8423 prompt_token_ids: None,
8424 extra: None,
8425 }),
8426 is_copied_context: None,
8427 llm_call_count: Some(1),
8428 extra: None,
8429 };
8430 let identity = AgentIdentity {
8431 version: "1".to_owned(),
8432 charter: "test".into(),
8433 instance_uid: "instance-1".to_owned(),
8434 ttl_remaining_s: Some(600),
8435 };
8436 let mut artifact_paths = Vec::new();
8437 for id in ["tampered-session", "healthy-session"] {
8438 let session = Arc::new(Session::new(
8439 id.to_owned(),
8440 Workflow::Unsigned,
8441 identity.clone(),
8442 Default::default(),
8443 ));
8444 session.atif.lock().push_step(step.clone()).unwrap();
8445 let outcome = finalizer
8446 .close_session(Arc::clone(&session), StopReason::SessionClosed)
8447 .await
8448 .unwrap();
8449 match outcome {
8450 FinalizeOutcome::Atif { path } => artifact_paths.push(path),
8451 other => panic!("expected FinalizeOutcome::Atif, got {other:?}"),
8452 }
8453 }
8454 let tampered = &artifact_paths[0];
8457 let mut bytes = std::fs::read(tampered).unwrap();
8458 let position = bytes
8459 .windows("archived".len())
8460 .position(|window| window == b"archived")
8461 .expect("artifact must contain the step message");
8462 bytes[position] = b'X';
8463 std::fs::write(tampered, &bytes).unwrap();
8464
8465 let registry = SessionRegistry::new();
8466 finalizer
8467 .recover_spooled_sessions(®istry, &Default::default())
8468 .await
8469 .expect("one tampered artifact must not abort the recovery scan");
8470 assert!(
8471 registry.get("healthy-session").is_some(),
8472 "the healthy artifact must still recover when a sibling is tampered",
8473 );
8474 assert!(
8475 registry.get("tampered-session").is_none(),
8476 "the tampered artifact must not recover a session",
8477 );
8478 assert!(
8479 tampered.exists(),
8480 "the tampered artifact must stay on disk as evidence",
8481 );
8482 let rendered = metrics.render();
8483 assert!(
8484 rendered.contains("av_atif_recovery_skipped_total{reason=\"provenance\"}"),
8485 "skips must be visible to operators via metrics; got: {rendered}",
8486 );
8487 }
8488
8489 #[tokio::test]
8498 async fn provenance_quarantine_moves_stale_sidecar_so_recycled_id_can_close() {
8499 let directory = tempfile::tempdir().unwrap();
8500 let metrics = Arc::new(Registry::new());
8501 let finalizer = Finalizer::new(
8502 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
8503 directory.path().to_path_buf(),
8504 Arc::clone(&metrics),
8505 );
8506 let step = av_atif::Step {
8507 step_id: 0,
8508 timestamp: None,
8509 source: av_atif::Source::Agent,
8510 message: serde_json::json!("first incarnation response"),
8511 reasoning_effort: None,
8512 reasoning_content: None,
8513 model_name: Some("test-model".into()),
8514 tool_calls: None,
8515 observation: None,
8516 metrics: Some(av_atif::Metrics {
8517 prompt_tokens: Some(1),
8518 completion_tokens: Some(1),
8519 cached_tokens: Some(0),
8520 cost_usd: Some(0.0),
8521 logprobs: None,
8522 completion_token_ids: None,
8523 prompt_token_ids: None,
8524 extra: None,
8525 }),
8526 is_copied_context: None,
8527 llm_call_count: Some(1),
8528 extra: None,
8529 };
8530 let identity = AgentIdentity {
8531 version: "1".to_owned(),
8532 charter: "test".into(),
8533 instance_uid: "instance-1".to_owned(),
8534 ttl_remaining_s: Some(600),
8535 };
8536 let session = Arc::new(Session::new(
8537 "quarantined-then-recycled".to_owned(),
8538 Workflow::Unsigned,
8539 identity.clone(),
8540 Default::default(),
8541 ));
8542 session.atif.lock().push_step(step.clone()).unwrap();
8543 let FinalizeOutcome::Atif { path } = finalizer
8544 .close_session(Arc::clone(&session), StopReason::SessionClosed)
8545 .await
8546 .unwrap()
8547 else {
8548 panic!("expected FinalizeOutcome::Atif");
8549 };
8550 let mut bytes = std::fs::read(&path).unwrap();
8554 let position = bytes
8555 .windows("first".len())
8556 .position(|window| window == b"first")
8557 .expect("artifact must contain the step message");
8558 bytes[position] = b'X';
8559 std::fs::write(&path, &bytes).unwrap();
8560 let old = std::time::SystemTime::now() - std::time::Duration::from_secs(120);
8561 let file = std::fs::OpenOptions::new().append(true).open(&path).unwrap();
8562 file.set_times(std::fs::FileTimes::new().set_modified(old))
8563 .unwrap();
8564 drop(file);
8565
8566 let registry = SessionRegistry::new();
8567 finalizer
8568 .recover_spooled_sessions(®istry, &Default::default())
8569 .await
8570 .expect("quarantine must not fail the scan");
8571
8572 let sidecar = path.with_extension("atif-auth");
8573 assert!(!path.exists(), "the tampered artifact must be quarantined");
8574 assert!(
8575 !sidecar.exists(),
8576 "the stale provenance sidecar must be quarantined alongside its artifact"
8577 );
8578 let mut quarantined_pair = 0usize;
8579 for entry in std::fs::read_dir(directory.path()).unwrap() {
8580 let name = entry.unwrap().file_name();
8581 if let Some(name) = name.to_str() {
8582 if name.contains(".json.corrupt-") {
8583 quarantined_pair += 1;
8584 }
8585 }
8586 }
8587 assert_eq!(
8588 quarantined_pair, 2,
8589 "both the artifact and its sidecar must survive as an associated forensic pair"
8590 );
8591
8592 let recycled = Arc::new(Session::new(
8596 "quarantined-then-recycled".to_owned(),
8597 Workflow::Unsigned,
8598 identity,
8599 Default::default(),
8600 ));
8601 recycled.atif.lock().push_step(step).unwrap();
8602 let outcome = finalizer
8603 .close_session(Arc::clone(&recycled), StopReason::SessionClosed)
8604 .await;
8605 assert!(
8606 matches!(outcome, Ok(FinalizeOutcome::Atif { .. })),
8607 "a recycled id must close cleanly after its predecessor was quarantined; got {outcome:?}"
8608 );
8609 }
8610
8611 #[tokio::test]
8625 async fn round_42_f1_corrupt_receipt_does_not_block_other_atif_recovery() {
8626 let directory = tempfile::tempdir().unwrap();
8627 let metrics = Arc::new(Registry::new());
8628 let finalizer = Finalizer::new(
8629 Arc::new(Ed25519Signer::from_seed(&[42; 32])),
8630 directory.path().to_path_buf(),
8631 Arc::clone(&metrics),
8632 );
8633 let step = av_atif::Step {
8634 step_id: 0,
8635 timestamp: None,
8636 source: av_atif::Source::Agent,
8637 message: serde_json::json!("archived response"),
8638 reasoning_effort: None,
8639 reasoning_content: None,
8640 model_name: Some("test-model".into()),
8641 tool_calls: None,
8642 observation: None,
8643 metrics: Some(av_atif::Metrics {
8644 prompt_tokens: Some(1),
8645 completion_tokens: Some(1),
8646 cached_tokens: Some(0),
8647 cost_usd: Some(0.0),
8648 logprobs: None,
8649 completion_token_ids: None,
8650 prompt_token_ids: None,
8651 extra: None,
8652 }),
8653 is_copied_context: None,
8654 llm_call_count: Some(1),
8655 extra: None,
8656 };
8657 let identity = AgentIdentity {
8658 version: "1".to_owned(),
8659 charter: "test".into(),
8660 instance_uid: "instance-1".to_owned(),
8661 ttl_remaining_s: Some(600),
8662 };
8663 for id in ["poisoned-receipt-session", "healthy-atif-session"] {
8668 let session = Arc::new(Session::new(
8669 id.to_owned(),
8670 Workflow::Unsigned,
8671 identity.clone(),
8672 Default::default(),
8673 ));
8674 session.atif.lock().push_step(step.clone()).unwrap();
8675 finalizer
8676 .close_session(Arc::clone(&session), StopReason::SessionClosed)
8677 .await
8678 .unwrap();
8679 }
8680
8681 let poisoned_receipt = directory.path().join("receipts").join(format!(
8688 "{}.json",
8689 &av_core::digest::sha256_hex(b"poisoned-receipt-session")[..32]
8690 ));
8691 std::fs::create_dir_all(poisoned_receipt.parent().unwrap()).unwrap();
8692 std::fs::write(&poisoned_receipt, b"{not valid receipt json").unwrap();
8693
8694 let registry = SessionRegistry::new();
8695 let outcome = finalizer
8696 .recover_spooled_sessions(®istry, &Default::default())
8697 .await;
8698 assert!(
8699 outcome.is_ok(),
8700 "corrupt receipt must not abort the ATIF recovery scan; got {outcome:?}",
8701 );
8702 assert!(
8703 registry.get("healthy-atif-session").is_some(),
8704 "the healthy ATIF session MUST recover even when a sibling has a poisoned receipt — this is the HOL-block invariant",
8705 );
8706 let rendered = metrics.render();
8707 assert!(
8708 rendered.contains("av_atif_trajectory_recovery_skipped_total"),
8709 "per-session ATIF recovery skips must be visible to operators via a dedicated counter; got: {rendered}",
8710 );
8711 }
8712
8713 #[tokio::test]
8721 async fn promotion_retry_does_not_force_close_a_live_session() {
8722 let directory = tempfile::tempdir().unwrap();
8723 let finalizer = finalizer(directory.path());
8724 let closed_session = session(Workflow::Unsigned);
8726 closed_session
8727 .atif
8728 .lock()
8729 .push_step(av_atif::Step {
8730 step_id: 0,
8731 timestamp: None,
8732 source: av_atif::Source::Agent,
8733 message: serde_json::json!("archived response"),
8734 reasoning_effort: None,
8735 reasoning_content: None,
8736 model_name: Some("test-model".into()),
8737 tool_calls: None,
8738 observation: None,
8739 metrics: Some(av_atif::Metrics {
8740 prompt_tokens: Some(1),
8741 completion_tokens: Some(1),
8742 cached_tokens: Some(0),
8743 cost_usd: Some(0.0),
8744 logprobs: None,
8745 completion_token_ids: None,
8746 prompt_token_ids: None,
8747 extra: None,
8748 }),
8749 is_copied_context: None,
8750 llm_call_count: Some(1),
8751 extra: None,
8752 })
8753 .unwrap();
8754 let outcome = finalizer
8755 .close_session(Arc::clone(&closed_session), StopReason::SessionClosed)
8756 .await
8757 .unwrap();
8758 let FinalizeOutcome::Atif { path } = outcome else {
8759 panic!("expected ATIF artifact")
8760 };
8761 let trajectory_bytes = tokio::fs::read(&path).await.unwrap();
8762 let promotion_marker = crate::journal::seal(
8763 &finalizer.journal_key,
8764 "promotion-marker",
8765 0,
8766 &PromotionMarker {
8767 session_id: closed_session.id.clone(),
8768 trajectory_digest: av_core::digest::sha256_hex(&trajectory_bytes),
8769 },
8770 )
8771 .unwrap();
8772 tokio::fs::write(path.with_extension("promote"), &promotion_marker)
8773 .await
8774 .unwrap();
8775 let registry = SessionRegistry::new();
8777 let live = registry.get_or_open(
8778 &closed_session.id,
8779 Workflow::Unsigned,
8780 &AgentIdentity {
8781 version: "1".to_owned(),
8782 charter: "test".into(),
8783 instance_uid: "instance-1".to_owned(),
8784 ttl_remaining_s: Some(600),
8785 },
8786 &Default::default(),
8787 );
8788 live.atif
8792 .lock()
8793 .push_step(av_atif::Step {
8794 step_id: 0,
8795 timestamp: None,
8796 source: av_atif::Source::Agent,
8797 message: serde_json::json!("live response"),
8798 reasoning_effort: None,
8799 reasoning_content: None,
8800 model_name: Some("test-model".into()),
8801 tool_calls: None,
8802 observation: None,
8803 metrics: Some(av_atif::Metrics {
8804 prompt_tokens: Some(2),
8805 completion_tokens: Some(3),
8806 cached_tokens: Some(0),
8807 cost_usd: Some(0.0),
8808 logprobs: None,
8809 completion_token_ids: None,
8810 prompt_token_ids: None,
8811 extra: None,
8812 }),
8813 is_copied_context: None,
8814 llm_call_count: Some(1),
8815 extra: None,
8816 })
8817 .unwrap();
8818 assert!(!live.is_closed(), "precondition: live session is open");
8819 let promoted = finalizer.retry_marked_promotions(®istry).await.unwrap();
8820 assert_eq!(
8821 promoted, 0,
8822 "promotion retry must not count a skipped live session as promoted",
8823 );
8824 assert!(
8825 !live.is_closed(),
8826 "live session must not be force-closed by promotion retry",
8827 );
8828 assert!(
8829 live.receipt.lock().is_none(),
8830 "live session must not receive a receipt from a stale promotion marker",
8831 );
8832 assert!(
8833 live.atif_path.lock().is_none(),
8834 "live session's atif_path must not be set by a background promotion retry — that would mean its trajectory was snapshotted to disk out of band",
8835 );
8836 assert_eq!(
8837 tokio::fs::read(&path).await.unwrap(),
8838 trajectory_bytes,
8839 "the archived ATIF artifact must not be overwritten by a background retry that force-finalized the live session",
8840 );
8841 assert!(
8844 path.with_extension("promote").exists(),
8845 "the promotion marker must remain on disk for future retries",
8846 );
8847 }
8848
8849 #[tokio::test]
8868 async fn recovery_marks_quarantined_unsigned_session_finalized_to_stop_idle_sweep_churn() {
8869 let directory = tempfile::tempdir().unwrap();
8870 let finalizer = finalizer(directory.path());
8871 let session_id = "quarantined-unsigned";
8872 let identity = AgentIdentity {
8873 version: "1".into(),
8874 charter: "test".into(),
8875 instance_uid: "instance-1".into(),
8876 ttl_remaining_s: Some(600),
8877 };
8878
8879 let digest = av_core::digest::sha256_hex(session_id.as_bytes());
8880 let stem = &digest[..32];
8881 let metadata_payload = serde_json::json!({
8882 "journal_version": 2,
8883 "session_id": session_id,
8884 "identity": identity,
8885 "workflow": "unsigned",
8886 });
8887 let metadata_sealed =
8888 crate::journal::seal(&finalizer.journal_key, "metadata", 0, &metadata_payload).unwrap();
8889 std::fs::write(
8890 directory.path().join(format!("{stem}.session.json")),
8891 &metadata_sealed,
8892 )
8893 .unwrap();
8894
8895 finalizer
8899 .quarantined_sessions
8900 .lock()
8901 .insert(session_id.to_owned());
8902
8903 let registry = SessionRegistry::new();
8904 finalizer
8905 .recover_spooled_sessions(®istry, &Default::default())
8906 .await
8907 .unwrap();
8908
8909 let recovered = registry
8910 .get(session_id)
8911 .expect("quarantined session must be inserted");
8912 assert!(
8913 recovered.capture_failed(),
8914 "quarantined session must carry the capture-failed verdict",
8915 );
8916 assert!(
8917 recovered.is_closed(),
8918 "the quarantined-unsigned recovery branch (consolidate_step_journals) must also mark the session finalized (artifact_committed) so the idle sweeper's `!is_closed()` filter skips it — otherwise every idle tick calls close_session_locked which returns CaptureIncomplete, CloseClaim resets the close, and the session churns forever burning CPU, log noise, and metrics without ever leaving the registry",
8919 );
8920 }
8921
8922 #[tokio::test]
8930 async fn atif_scan_ignores_live_session_journal_metadata() {
8931 let directory = tempfile::tempdir().unwrap();
8932 let metrics = Arc::new(Registry::new());
8933 let finalizer = Finalizer::new(
8934 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
8935 directory.path().to_path_buf(),
8936 Arc::clone(&metrics),
8937 );
8938 let session_id = "still-open-session";
8939 let identity = AgentIdentity {
8940 version: "1".into(),
8941 charter: "test".into(),
8942 instance_uid: "instance-1".into(),
8943 ttl_remaining_s: Some(600),
8944 };
8945 let digest = av_core::digest::sha256_hex(session_id.as_bytes());
8946 let stem = &digest[..32];
8947 let metadata_payload = serde_json::json!({
8948 "journal_version": 2,
8949 "session_id": session_id,
8950 "identity": identity,
8951 "workflow": "unsigned",
8952 });
8953 let metadata_sealed =
8954 crate::journal::seal(&finalizer.journal_key, "metadata", 0, &metadata_payload).unwrap();
8955 std::fs::write(
8956 directory.path().join(format!("{stem}.session.json")),
8957 &metadata_sealed,
8958 )
8959 .unwrap();
8960
8961 let registry = SessionRegistry::new();
8963 let live = registry.get_or_open(session_id, Workflow::Unsigned, &identity, &Default::default());
8964 assert!(!live.is_closed(), "precondition: session is open");
8965
8966 finalizer
8967 .recover_spooled_sessions(®istry, &Default::default())
8968 .await
8969 .unwrap();
8970
8971 assert!(
8972 !metrics.render().contains("av_atif_recovery_skipped_total"),
8973 "the ATIF scan must skip *.session.json instead of counting it as an invalid spool file",
8974 );
8975 assert!(
8976 directory.path().join(format!("{stem}.session.json")).exists(),
8977 "the live session's journal metadata must survive the pass",
8978 );
8979 }
8980
8981 #[tokio::test]
8982 async fn restart_quarantines_inflight_response_without_stopping_recovery() {
8983 let directory = tempfile::tempdir().unwrap();
8984 let finalizer = finalizer(directory.path());
8985 crate::worker::create_response_marker(
8986 directory.path(),
8987 &finalizer.journal_key,
8988 "uncertain-session",
8989 "request-digest".to_owned(),
8990 )
8991 .await
8992 .unwrap();
8993
8994 let registry = SessionRegistry::new();
8995 assert_eq!(
8996 finalizer
8997 .recover_spooled_sessions(®istry, &Default::default())
8998 .await
8999 .unwrap(),
9000 0
9001 );
9002 assert!(finalizer
9003 .quarantined_sessions
9004 .lock()
9005 .contains("uncertain-session"));
9006 assert!(registry.get("uncertain-session").is_none());
9007 }
9008
9009 #[tokio::test]
9010 async fn mcp_durable_startup_distinguishes_deferred_metadata_from_orphan() {
9011 let directory = tempfile::tempdir().unwrap();
9012 let finalizer = finalizer(directory.path());
9013 let registry = SessionRegistry::new();
9014 let session_id = "deferred-uncertain";
9015 let key = av_core::digest::sha256_hex(b"deferred execution");
9016 let executions = directory.path().join(crate::spool::TOOL_EXECUTIONS);
9017 std::fs::create_dir(&executions).unwrap();
9018 let intent = serde_json::json!({
9019 "execution_key": key, "session_id": session_id, "tool": "read",
9020 "request_digest": "request-digest", "principal_digest": "principal-digest",
9021 });
9022 let sealed_intent = crate::journal::seal(
9023 &finalizer.journal_key,
9024 &format!("{}:{key}", crate::journal::TOOL_INTENT_DOMAIN),
9025 0,
9026 &intent,
9027 )
9028 .unwrap();
9029 let intent_path = executions.join(format!("{key}{}", crate::spool::TOOL_INTENT_SUFFIX));
9030 std::fs::write(&intent_path, &sealed_intent).unwrap();
9031 assert!(finalizer
9032 .ensure_unresolved_tool_metadata(®istry)
9033 .await
9034 .is_err());
9035 let identity = AgentIdentity {
9036 version: "1".to_owned(),
9037 charter: "test".into(),
9038 instance_uid: "instance-1".to_owned(),
9039 ttl_remaining_s: None,
9040 };
9041 let metadata = serde_json::json!({
9042 "journal_version": 2, "session_id": session_id,
9043 "identity": identity, "workflow": "unsigned",
9044 });
9045 let digest = av_core::digest::sha256_hex(session_id.as_bytes());
9046 let metadata_path = directory.path().join(format!("{}.session.json", &digest[..32]));
9047 std::fs::write(
9048 &metadata_path,
9049 crate::journal::seal(&finalizer.journal_key, "metadata", 0, &metadata).unwrap(),
9050 )
9051 .unwrap();
9052 finalizer
9055 .ensure_unresolved_tool_metadata(®istry)
9056 .await
9057 .unwrap();
9058 assert!(registry.get(session_id).is_none());
9059 for (field, value) in [
9060 ("journal_version", serde_json::json!(3)),
9061 ("session_id", serde_json::json!("different-session")),
9062 ("workflow", serde_json::json!("unknown")),
9063 ("identity", serde_json::json!(null)),
9064 ] {
9065 let mut invalid = metadata.clone();
9066 invalid[field] = value;
9067 let sealed = crate::journal::seal(&finalizer.journal_key, "metadata", 0, &invalid).unwrap();
9068 std::fs::write(&metadata_path, &sealed).unwrap();
9069 assert!(
9070 finalizer
9071 .ensure_unresolved_tool_metadata(®istry)
9072 .await
9073 .is_err(),
9074 "authenticated but irrecoverable {field} must not exempt an orphan"
9075 );
9076 assert_eq!(std::fs::read(&metadata_path).unwrap(), sealed);
9077 }
9078 std::fs::write(&metadata_path, b"unauthenticated metadata").unwrap();
9079 assert!(finalizer
9080 .ensure_unresolved_tool_metadata(®istry)
9081 .await
9082 .is_err());
9083 assert_eq!(
9084 std::fs::read(&metadata_path).unwrap(),
9085 b"unauthenticated metadata"
9086 );
9087 assert_eq!(std::fs::read(intent_path).unwrap(), sealed_intent);
9088 }
9089
9090 #[tokio::test]
9091 async fn restart_recovers_atif_and_retries_marked_promotion() {
9092 let directory = tempfile::tempdir().unwrap();
9093 let first = finalizer(directory.path());
9094 let original = session(Workflow::Unsigned);
9095 original
9096 .atif
9097 .lock()
9098 .push_step(av_atif::Step {
9099 step_id: 0,
9100 timestamp: None,
9101 source: av_atif::Source::Agent,
9102 message: serde_json::json!("recovered response"),
9103 reasoning_effort: None,
9104 reasoning_content: None,
9105 model_name: Some("test-model".into()),
9106 tool_calls: None,
9107 observation: None,
9108 metrics: Some(av_atif::Metrics {
9109 prompt_tokens: Some(5),
9110 completion_tokens: Some(2),
9111 cached_tokens: Some(1),
9112 cost_usd: Some(0.001),
9113 logprobs: None,
9114 completion_token_ids: None,
9115 prompt_token_ids: None,
9116 extra: None,
9117 }),
9118 is_copied_context: None,
9119 llm_call_count: Some(1),
9120 extra: None,
9121 })
9122 .unwrap();
9123 let outcome = first
9124 .close_session(Arc::clone(&original), StopReason::SessionClosed)
9125 .await
9126 .unwrap();
9127 let FinalizeOutcome::Atif { path } = outcome else {
9128 panic!("expected ATIF artifact")
9129 };
9130 let trajectory_bytes = tokio::fs::read(&path).await.unwrap();
9131 let promotion_marker = crate::journal::seal(
9132 &first.journal_key,
9133 "promotion-marker",
9134 0,
9135 &PromotionMarker {
9136 session_id: original.id.clone(),
9137 trajectory_digest: av_core::digest::sha256_hex(&trajectory_bytes),
9138 },
9139 )
9140 .unwrap();
9141 tokio::fs::write(path.with_extension("promote"), &promotion_marker)
9142 .await
9143 .unwrap();
9144
9145 let recovered_registry = SessionRegistry::new();
9146 let after_restart = finalizer(directory.path());
9147 assert_eq!(
9148 after_restart
9149 .recover_spooled_sessions(&recovered_registry, &Default::default())
9150 .await
9151 .unwrap(),
9152 1
9153 );
9154 assert_eq!(
9155 after_restart
9156 .retry_marked_promotions(&recovered_registry)
9157 .await
9158 .unwrap(),
9159 1
9160 );
9161 let recovered = recovered_registry.get(&original.id).unwrap();
9162 let receipt = recovered.receipt.lock().clone().unwrap();
9163 receipt.verify_embedded().unwrap();
9164 assert!(recovered.is_promoted());
9165 assert!(!path.with_extension("promote").exists());
9166
9167 tokio::fs::write(path.with_extension("promote"), &promotion_marker)
9168 .await
9169 .unwrap();
9170 let second_registry = SessionRegistry::new();
9171 assert_eq!(
9172 after_restart
9173 .recover_spooled_sessions(&second_registry, &Default::default())
9174 .await
9175 .unwrap(),
9176 1
9177 );
9178 let restored = second_registry.get(&original.id).unwrap();
9179 assert_eq!(
9180 restored.receipt.lock().as_ref().unwrap().body.receipt_id,
9181 receipt.body.receipt_id,
9182 "restart must restore the persisted receipt, not issue a duplicate"
9183 );
9184 assert_eq!(
9185 after_restart
9186 .retry_marked_promotions(&second_registry)
9187 .await
9188 .unwrap(),
9189 1
9190 );
9191 assert!(restored.is_promoted());
9192 assert!(!path.with_extension("promote").exists());
9193 }
9194
9195 #[tokio::test]
9196 async fn close_waits_for_active_response_lease() {
9197 let directory = tempfile::tempdir().unwrap();
9198 let finalizer = finalizer(directory.path());
9199 let session = session(Workflow::Signed);
9200 let lease = crate::session::SessionLease::new(Arc::clone(&session));
9201 let close_session = Arc::clone(&session);
9202 let task = tokio::spawn(async move {
9203 finalizer
9204 .close_session(close_session, StopReason::SessionClosed)
9205 .await
9206 });
9207 tokio::task::yield_now().await;
9208 assert!(!task.is_finished(), "close overtook an active response");
9209 drop(lease);
9210 assert!(matches!(
9211 task.await.unwrap().unwrap(),
9212 FinalizeOutcome::Receipt { .. }
9213 ));
9214 }
9215
9216 #[test]
9225 fn warned_artifacts_clamps_zero_cap_to_one() {
9226 let mut warned = WarnedArtifacts::new(0);
9227 assert!(warned.insert(PathBuf::from("a")));
9229 assert_eq!(warned.len(), 1);
9230 assert!(!warned.insert(PathBuf::from("a")));
9232 assert!(warned.insert(PathBuf::from("b")));
9234 assert_eq!(warned.len(), 1);
9235 }
9236
9237 #[test]
9238 fn warned_artifacts_evicts_one_at_a_time_not_all_at_once() {
9239 let mut warned = WarnedArtifacts::new(3);
9240 assert!(warned.insert(PathBuf::from("a")));
9241 assert!(warned.insert(PathBuf::from("b")));
9242 assert!(warned.insert(PathBuf::from("c")));
9243 assert_eq!(warned.len(), 3);
9244 assert!(!warned.insert(PathBuf::from("b")));
9247 assert!(warned.insert(PathBuf::from("d")));
9250 assert_eq!(warned.len(), 3);
9251 assert!(!warned.insert(PathBuf::from("b")));
9252 assert!(!warned.insert(PathBuf::from("c")));
9253 assert!(!warned.insert(PathBuf::from("d")));
9254 assert!(warned.insert(PathBuf::from("a")));
9259 assert_eq!(warned.len(), 3);
9260 assert!(!warned.insert(PathBuf::from("c")));
9261 assert!(!warned.insert(PathBuf::from("d")));
9262 assert!(!warned.insert(PathBuf::from("a")));
9263 }
9264
9265 #[test]
9271 fn quarantined_sessions_clamps_zero_cap_and_evicts_fifo() {
9272 let mut q = QuarantinedSessions::new(0);
9273 assert!(q.insert("a".to_owned()));
9274 assert!(q.contains("a"));
9275 assert_eq!(q.len(), 1);
9276 assert!(!q.insert("a".to_owned()), "duplicate must not re-insert");
9277 assert!(q.insert("b".to_owned()));
9279 assert_eq!(q.len(), 1);
9280 assert!(!q.contains("a"));
9281 assert!(q.contains("b"));
9282
9283 let mut q = QuarantinedSessions::new(3);
9287 assert!(q.insert("a".to_owned()));
9288 assert!(q.insert("b".to_owned()));
9289 assert!(q.insert("c".to_owned()));
9290 assert_eq!(q.len(), 3);
9291 assert!(q.insert("d".to_owned()));
9292 assert_eq!(q.len(), 3);
9293 assert!(!q.contains("a"), "a evicted");
9294 assert!(q.contains("b") && q.contains("c") && q.contains("d"));
9295 assert!(q.insert("e".to_owned()));
9296 assert!(!q.contains("b"), "b evicted next");
9297 assert!(q.contains("c") && q.contains("d") && q.contains("e"));
9298 }
9299
9300 #[tokio::test]
9301 async fn torn_journal_tail_is_truncated_without_losing_complete_records() {
9302 let directory = tempfile::tempdir().unwrap();
9303 let path = directory.path().join("journal.ndjson");
9304 std::fs::write(&path, b"{\"complete\":true}\n{\"torn\":").unwrap();
9305 let lines = read_complete_journal(&path).await.unwrap();
9306 assert_eq!(lines, vec![r#"{"complete":true}"#]);
9307 assert_eq!(std::fs::read(&path).unwrap(), b"{\"complete\":true}\n");
9308 }
9309
9310 #[tokio::test]
9317 async fn journal_with_no_complete_lines_is_quarantined_not_truncated() {
9318 let directory = tempfile::tempdir().unwrap();
9319 let path = directory.path().join("journal.ndjson");
9320 std::fs::write(&path, b"{\"torn_before_first_newline\":").unwrap();
9321 let outcome = read_complete_journal(&path).await;
9322 let err = outcome.unwrap_err();
9324 assert!(
9325 format!("{err:?}").contains("no complete lines"),
9326 "expected quarantine error, got {err:?}",
9327 );
9328 assert!(!path.exists(), "journal was not moved out of the recovery path");
9330 let entries: Vec<_> = std::fs::read_dir(directory.path())
9333 .unwrap()
9334 .filter_map(Result::ok)
9335 .filter(|entry| entry.file_name().to_string_lossy().contains(".corrupt-"))
9336 .collect();
9337 assert_eq!(entries.len(), 1, "expected exactly one quarantine file");
9338 let bytes = std::fs::read(entries[0].path()).unwrap();
9339 assert_eq!(&bytes, b"{\"torn_before_first_newline\":");
9340 let msg = format!("{err}");
9349 assert!(
9350 !msg.contains(directory.path().to_string_lossy().as_ref()),
9351 "FinalizeError::Atif leaked spool dir absolute path: {msg}"
9352 );
9353 assert!(
9354 !msg.contains(std::path::MAIN_SEPARATOR_STR),
9355 "FinalizeError::Atif still contains a path separator: {msg}"
9356 );
9357 assert!(
9360 msg.contains("journal.ndjson"),
9361 "FinalizeError::Atif should still mention the journal basename: {msg}"
9362 );
9363 }
9364
9365 #[tokio::test]
9366 async fn failed_persistence_reopens_session_for_retry() {
9367 let directory = tempfile::tempdir().unwrap();
9368 let blocking_file = directory.path().join("not-a-directory");
9369 std::fs::write(&blocking_file, b"file").unwrap();
9370 let finalizer = finalizer(&blocking_file);
9371 let session = session(Workflow::Signed);
9372 assert!(finalizer
9373 .close_session(Arc::clone(&session), StopReason::SessionClosed)
9374 .await
9375 .is_err());
9376 assert!(!session.is_closed());
9377 assert!(session.try_close(), "failed close claim was not reset");
9378 }
9379
9380 #[tokio::test]
9381 async fn failed_unsigned_persistence_keeps_steps_for_retry() {
9382 let directory = tempfile::tempdir().unwrap();
9383 let spool = directory.path().join("spool");
9384 std::fs::write(&spool, b"blocking file").unwrap();
9385 let finalizer = finalizer(&spool);
9386 let session = session(Workflow::Unsigned);
9387 session
9388 .atif
9389 .lock()
9390 .push_step(av_atif::Step {
9391 step_id: 0,
9392 timestamp: None,
9393 source: av_atif::Source::User,
9394 message: serde_json::json!("survive"),
9395 reasoning_effort: None,
9396 reasoning_content: None,
9397 model_name: None,
9398 tool_calls: None,
9399 observation: None,
9400 metrics: None,
9401 is_copied_context: None,
9402 llm_call_count: None,
9403 extra: None,
9404 })
9405 .unwrap();
9406 assert!(finalizer
9407 .close_session(Arc::clone(&session), StopReason::SessionClosed)
9408 .await
9409 .is_err());
9410 assert!(!session.is_closed());
9411 std::fs::remove_file(&spool).unwrap();
9412 std::fs::create_dir(&spool).unwrap();
9413 let FinalizeOutcome::Atif { path } = finalizer
9414 .close_session(session, StopReason::SessionClosed)
9415 .await
9416 .unwrap()
9417 else {
9418 panic!("expected ATIF")
9419 };
9420 let trajectory: av_atif::Trajectory = serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap();
9421 assert_eq!(trajectory.steps.len(), 1);
9422 assert_eq!(trajectory.steps[0].message, serde_json::json!("survive"));
9423 }
9424
9425 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
9436 async fn concurrent_closes_across_many_sessions_never_deadlock() {
9437 const SESSIONS: usize = 32;
9438 let directory = tempfile::tempdir().unwrap();
9439 let finalizer = Arc::new(finalizer(directory.path()));
9440 let mut tasks = Vec::with_capacity(SESSIONS);
9441 for i in 0..SESSIONS {
9442 let f = Arc::clone(&finalizer);
9443 let s = Arc::new(Session::new(
9444 format!("lifecycle-{i}"),
9445 Workflow::Signed,
9446 AgentIdentity {
9447 version: "1".to_owned(),
9448 charter: "test".into(),
9449 instance_uid: format!("instance-{i}"),
9450 ttl_remaining_s: Some(600),
9451 },
9452 Default::default(),
9453 ));
9454 tasks.push(tokio::spawn(async move {
9455 f.close_session(s, StopReason::SessionClosed).await
9456 }));
9457 }
9458 let results = tokio::time::timeout(
9459 std::time::Duration::from_secs(15),
9460 futures::future::join_all(tasks),
9461 )
9462 .await
9463 .expect("close_session tasks deadlocked under lifecycle_lock contention");
9464 for result in results {
9465 let outcome = result.expect("task panicked");
9466 assert!(outcome.is_ok(), "close failed: {outcome:?}");
9467 }
9468 }
9469
9470 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
9477 async fn per_session_lock_table_prunes_entries_after_close() {
9478 const SESSIONS: usize = 200;
9479 let directory = tempfile::tempdir().unwrap();
9480 let finalizer = Arc::new(finalizer(directory.path()));
9481 let locks = finalizer.lifecycle_locks();
9482 let mut tasks = Vec::with_capacity(SESSIONS);
9483 for i in 0..SESSIONS {
9484 let f = Arc::clone(&finalizer);
9485 let s = Arc::new(Session::new(
9486 format!("prune-{i}"),
9487 Workflow::Signed,
9488 AgentIdentity {
9489 version: "1".to_owned(),
9490 charter: "test".into(),
9491 instance_uid: format!("instance-{i}"),
9492 ttl_remaining_s: Some(600),
9493 },
9494 Default::default(),
9495 ));
9496 tasks.push(tokio::spawn(async move {
9497 f.close_session(s, StopReason::SessionClosed).await
9498 }));
9499 }
9500 for task in tasks {
9501 let _ = task.await.expect("task panicked");
9502 }
9503 for _ in 0..60 {
9513 if locks.len() == 0 {
9514 return;
9515 }
9516 tokio::time::sleep(std::time::Duration::from_millis(25)).await;
9517 }
9518 panic!(
9519 "SessionLockTable did not prune to empty after {SESSIONS} closes; \
9520 residual entries = {}",
9521 locks.len()
9522 );
9523 }
9524
9525 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
9546 async fn recovery_scan_does_not_head_of_line_block_unrelated_close() {
9547 let directory = tempfile::tempdir().unwrap();
9548 let finalizer = Arc::new(finalizer(directory.path()));
9549 let signer: Arc<dyn Signer> = Arc::new(Ed25519Signer::from_seed(&[7; 32]));
9552 let journal_key = crate::journal::key_from_signer(signer.as_ref());
9553
9554 let blocker_id = "scan-blocker";
9558 let digest = av_core::digest::sha256_hex(blocker_id.as_bytes());
9559 let stem = &digest[..32];
9560 let metadata_payload = serde_json::json!({
9561 "journal_version": 2,
9562 "session_id": blocker_id,
9563 "identity": AgentIdentity {
9564 version: "1".to_owned(),
9565 charter: "test".into(),
9566 instance_uid: "instance-blocker".to_owned(),
9567 ttl_remaining_s: Some(600),
9568 },
9569 "workflow": "signed",
9570 });
9571 let sealed = crate::journal::seal(&journal_key, "metadata", 0, &metadata_payload).unwrap();
9572 std::fs::write(directory.path().join(format!("{stem}.session.json")), &sealed).unwrap();
9573
9574 let locks = finalizer.lifecycle_locks();
9578 let held = locks.arc_for(blocker_id).lock_owned().await;
9579
9580 let f_scan = Arc::clone(&finalizer);
9581 let scan_task = tokio::spawn(async move {
9582 f_scan
9583 .recover_spooled_sessions(&crate::session::SessionRegistry::new(), &Default::default())
9584 .await
9585 });
9586
9587 let session = Arc::new(Session::new(
9588 "unrelated-close".to_owned(),
9589 Workflow::Signed,
9590 AgentIdentity {
9591 version: "1".to_owned(),
9592 charter: "test".into(),
9593 instance_uid: "instance-close".to_owned(),
9594 ttl_remaining_s: Some(600),
9595 },
9596 Default::default(),
9597 ));
9598 let _ = tokio::time::timeout(
9602 std::time::Duration::from_secs(60),
9603 finalizer.close_session(session, StopReason::SessionClosed),
9604 )
9605 .await
9606 .expect("unrelated close blocked behind the in-flight recovery scan — the head-of-line blocking signature of a global lifecycle lock");
9607
9608 assert!(
9612 !scan_task.is_finished(),
9613 "recovery scan finished while its candidate's lifecycle lock was externally held — \
9614 the scan never acquired the per-session lock, so this test lost its teeth"
9615 );
9616
9617 drop(held);
9618 scan_task
9619 .await
9620 .unwrap()
9621 .expect("recovery scan must complete after the blocker lock is released");
9622 }
9623
9624 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
9632 async fn close_latency_scales_reasonably_under_lock_contention() {
9633 let directory = tempfile::tempdir().unwrap();
9634 let finalizer = Arc::new(finalizer(directory.path()));
9635 let warm = Arc::new(Session::new(
9637 "warm".to_owned(),
9638 Workflow::Signed,
9639 AgentIdentity {
9640 version: "1".to_owned(),
9641 charter: "test".into(),
9642 instance_uid: "warm".to_owned(),
9643 ttl_remaining_s: Some(600),
9644 },
9645 Default::default(),
9646 ));
9647 let uncontended = std::time::Instant::now();
9648 finalizer
9649 .close_session(warm, StopReason::SessionClosed)
9650 .await
9651 .unwrap();
9652 let baseline = uncontended.elapsed();
9653
9654 const N: usize = 16;
9656 let mut tasks = Vec::with_capacity(N);
9657 let started = std::time::Instant::now();
9658 for i in 0..N {
9659 let f = Arc::clone(&finalizer);
9660 let s = Arc::new(Session::new(
9661 format!("contended-{i}"),
9662 Workflow::Signed,
9663 AgentIdentity {
9664 version: "1".to_owned(),
9665 charter: "test".into(),
9666 instance_uid: format!("contended-{i}"),
9667 ttl_remaining_s: Some(600),
9668 },
9669 Default::default(),
9670 ));
9671 tasks.push(tokio::spawn(async move {
9672 f.close_session(s, StopReason::SessionClosed).await
9673 }));
9674 }
9675 for t in tasks {
9676 t.await.unwrap().unwrap();
9677 }
9678 let total = started.elapsed();
9679 let multiplier = u32::try_from(N * 10).unwrap_or(u32::MAX);
9683 let budget = baseline
9684 .saturating_mul(multiplier)
9685 .max(std::time::Duration::from_secs(60));
9686 assert!(
9687 total < budget,
9688 "16 contended closes took {total:?}, budget {budget:?} (baseline {baseline:?})",
9689 );
9690 }
9691
9692 #[tokio::test(flavor = "current_thread")]
9706 async fn promotion_retry_quarantines_markers_whose_evidence_is_gone() {
9707 let directory = tempfile::tempdir().unwrap();
9708 let finalizer = finalizer(directory.path());
9709 let spool = directory.path();
9710
9711 let orphan_marker = spool.join("gone.promote");
9712 let sealed = crate::journal::seal(
9713 &finalizer.journal_key,
9714 "promotion-marker",
9715 0,
9716 &PromotionMarker {
9717 session_id: "session-gone".to_owned(),
9718 trajectory_digest: av_core::digest::sha256_hex(b"vanished bytes"),
9719 },
9720 )
9721 .unwrap();
9722 std::fs::write(&orphan_marker, &sealed).unwrap();
9723
9724 let kept_marker = spool.join("kept.promote");
9727 let kept_sealed = crate::journal::seal(
9728 &finalizer.journal_key,
9729 "promotion-marker",
9730 0,
9731 &PromotionMarker {
9732 session_id: "session-kept".to_owned(),
9733 trajectory_digest: av_core::digest::sha256_hex(b"{}"),
9734 },
9735 )
9736 .unwrap();
9737 std::fs::write(&kept_marker, &kept_sealed).unwrap();
9738 std::fs::write(spool.join("kept.json"), b"{}").unwrap();
9739 std::fs::write(spool.join("kept.atif-auth"), b"provenance").unwrap();
9740
9741 let registry = SessionRegistry::new();
9742 let promoted = finalizer.retry_marked_promotions(®istry).await.unwrap();
9743 assert_eq!(promoted, 0);
9744
9745 assert!(
9746 !orphan_marker.exists(),
9747 "an unsatisfiable marker must not keep its .promote name (retry-forever loop)"
9748 );
9749 assert!(
9750 spool.join("gone.promote.orphaned").exists(),
9751 "the sealed marker payload must survive quarantine as forensic evidence"
9752 );
9753 assert!(
9754 kept_marker.exists(),
9755 "a marker whose evidence pair exists is recoverable and must be left alone"
9756 );
9757 assert!(!spool.join("kept.promote.orphaned").exists());
9758
9759 let promoted = finalizer.retry_marked_promotions(®istry).await.unwrap();
9762 assert_eq!(promoted, 0);
9763 assert!(spool.join("gone.promote.orphaned").exists());
9764 }
9765
9766 #[tokio::test(flavor = "current_thread")]
9767 async fn retention_prunes_only_old_paired_evidence() {
9768 let temp = tempfile::tempdir().unwrap();
9769 let spool = temp.path().to_path_buf();
9770 let signer = Arc::new(Ed25519Signer::from_seed(&[7u8; 32])) as Arc<dyn Signer>;
9771 let metrics = Arc::new(Registry::new());
9772 let finalizer = Finalizer::new(signer, spool.clone(), metrics);
9773
9774 let old_json = spool.join("bbb.json");
9777 let old_sc = spool.join("bbb.atif-auth");
9778 std::fs::write(&old_json, b"{}").unwrap();
9779 std::fs::write(&old_sc, b"provenance").unwrap();
9780 let old_marker = spool.join("bbb.close-complete");
9784 std::fs::write(&old_marker, b"sealed").unwrap();
9785
9786 let orphan_marker = spool.join("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.close-complete");
9789 std::fs::write(&orphan_marker, b"sealed").unwrap();
9790
9791 let archived_marker = spool.join("eee.archived-traj9.close-complete");
9794 std::fs::write(&archived_marker, b"sealed").unwrap();
9795
9796 let orphan_json = spool.join("ccc.json");
9799 std::fs::write(&orphan_json, b"{}").unwrap();
9800
9801 let marked_json = spool.join("fff.json");
9807 let marked_sc = spool.join("fff.atif-auth");
9808 let marked_promote = spool.join("fff.promote");
9809 std::fs::write(&marked_json, b"{}").unwrap();
9810 std::fs::write(&marked_sc, b"provenance").unwrap();
9811 std::fs::write(&marked_promote, b"pending promotion").unwrap();
9812
9813 let live_session = spool.join("ddd.session.json");
9815 std::fs::write(&live_session, b"{}").unwrap();
9816
9817 let removed = finalizer
9819 .prune_sealed_atif(std::time::Duration::ZERO)
9820 .await
9821 .unwrap();
9822 assert_eq!(removed, 1, "exactly one sealed pair must be pruned");
9823
9824 assert!(!old_json.exists(), "aged sealed evidence must be pruned");
9825 assert!(!old_sc.exists(), "aged sidecar must be pruned");
9826 assert!(
9827 !old_marker.exists(),
9828 "the pair's close-complete marker must be pruned with the pair"
9829 );
9830 assert!(
9831 !orphan_marker.exists(),
9832 "an orphaned per-close marker past retention must be healed"
9833 );
9834 assert!(
9835 archived_marker.exists(),
9836 "archived collision markers are preserved evidence, never retention targets"
9837 );
9838 assert!(
9839 orphan_json.exists(),
9840 "unpaired remnants belong to quarantine, not retention"
9841 );
9842 assert!(
9843 marked_json.exists() && marked_sc.exists() && marked_promote.exists(),
9844 "a pair with a pending promotion marker is owed a receipt and must survive retention"
9845 );
9846 assert!(live_session.exists(), ".session.json is never a retention target");
9847
9848 let fresh_json = spool.join("aaa.json");
9851 let fresh_sc = spool.join("aaa.atif-auth");
9852 std::fs::write(&fresh_json, b"{}").unwrap();
9853 std::fs::write(&fresh_sc, b"provenance").unwrap();
9854 let fresh_marker = spool.join("aaa.close-complete");
9855 std::fs::write(&fresh_marker, b"sealed").unwrap();
9856 let fresh_orphan_marker = spool.join("cccccccccccccccccccccccccccccccc.close-complete");
9859 std::fs::write(&fresh_orphan_marker, b"sealed").unwrap();
9860
9861 let removed = finalizer
9862 .prune_sealed_atif(std::time::Duration::from_secs(60 * 60))
9863 .await
9864 .unwrap();
9865 assert_eq!(removed, 0, "fresh evidence must be preserved");
9866 assert!(fresh_json.exists());
9867 assert!(fresh_sc.exists());
9868 assert!(fresh_marker.exists(), "a live pair keeps its marker");
9869 assert!(
9870 fresh_orphan_marker.exists(),
9871 "an orphaned marker inside the retention window is left alone"
9872 );
9873 }
9874
9875 #[tokio::test]
9883 async fn recovery_scan_dirent_cap_returns_early_on_wrong_extension_flood() {
9884 let temp = tempfile::tempdir().unwrap();
9885 let spool = temp.path().to_path_buf();
9886 let metrics = Arc::new(Registry::new());
9887 let finalizer = Finalizer::new(
9888 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
9889 spool.clone(),
9890 Arc::clone(&metrics),
9891 );
9892
9893 let overshoot = MAX_RECOVERY_DIRENTS_PER_TICK + 100;
9899 for i in 0..overshoot {
9900 std::fs::write(spool.join(format!("junk-{i}.junk")), b"").unwrap();
9901 }
9902
9903 let sessions = SessionRegistry::new();
9904 let breaker = av_loopdetect::BreakerConfig::default();
9905 let recovered = finalizer
9906 .adopt_strict_atif_artifacts(&sessions, &breaker)
9907 .await
9908 .expect("scan returns cleanly on cap-hit, not error");
9909 assert_eq!(recovered, 0, "wrong-extension entries are never adopted");
9910
9911 let scrape = metrics.render();
9912 assert!(
9913 scrape.contains("av_recovery_scan_capped_total{pass=\"adopt_strict_atif\"} 1"),
9914 "dirent cap must fire exactly once on wrong-extension flood; got scrape:\n{scrape}"
9915 );
9916 }
9917
9918 #[tokio::test]
9927 async fn recovery_scan_entries_cap_fires_on_matching_entries_flood() {
9928 let temp = tempfile::tempdir().unwrap();
9929 let spool = temp.path().to_path_buf();
9930 let metrics = Arc::new(Registry::new());
9931 let finalizer = Finalizer::new(
9932 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
9933 spool.clone(),
9934 Arc::clone(&metrics),
9935 );
9936
9937 let overshoot = MAX_RECOVERY_ENTRIES_PER_TICK + 100;
9943 for i in 0..overshoot {
9944 let stem = format!("{:032x}", i);
9945 std::fs::write(spool.join(format!("{stem}.json")), b"").unwrap();
9946 std::fs::write(spool.join(format!("{stem}.atif-auth")), b"provenance").unwrap();
9947 }
9948 for i in 0..50 {
9953 std::fs::write(spool.join(format!("{:032x}.json", 0x1000_0000 + i)), b"").unwrap();
9954 }
9955
9956 let sessions = SessionRegistry::new();
9957 let breaker = av_loopdetect::BreakerConfig::default();
9958 let _ = finalizer
9959 .adopt_strict_atif_artifacts(&sessions, &breaker)
9960 .await
9961 .expect("scan returns cleanly on cap-hit, not error");
9962
9963 let scrape = metrics.render();
9964 assert!(
9965 scrape.contains("av_recovery_scan_capped_total{pass=\"adopt_strict_atif\"} 1"),
9966 "entries cap must fire exactly once on matching flood; got scrape:\n{scrape}"
9967 );
9968 }
9969
9970 #[tokio::test]
9978 async fn lifecycle_outbox_pending_gauge_reflects_unacked_count() {
9979 let directory = tempfile::tempdir().unwrap();
9980 let metrics = Arc::new(Registry::new());
9981 let finalizer = Finalizer::new(
9982 Arc::new(Ed25519Signer::from_seed(&[7; 32])),
9983 directory.path().to_path_buf(),
9984 Arc::clone(&metrics),
9985 );
9986 let outbox_dir = directory.path().join(crate::spool::OUTBOX);
9987 std::fs::create_dir_all(&outbox_dir).unwrap();
9988
9989 let topic = av_events::EventClass::Receipt.topic();
9990 for i in 0..3 {
9992 let session_id = format!("session-unacked-{i}");
9993 let outbox = LifecycleOutbox {
9994 schema_version: LIFECYCLE_OUTBOX_SCHEMA_V1,
9995 session_id: session_id.clone(),
9996 kind: crate::journal::RECEIPT_OUTBOX_KIND.to_owned(),
9997 topic: topic.to_owned(),
9998 key: format!("instance-{i}"),
9999 session_scope: None,
10000 value: serde_json::json!({
10001 "metadata": { "sequence": 0, "uid": av_core::new_event_uid() },
10002 "topic": topic,
10003 }),
10004 ack: None,
10005 };
10006 let path = finalizer.lifecycle_outbox_path(&session_id, crate::journal::RECEIPT_OUTBOX_KIND);
10007 let sealed = crate::journal::seal(
10008 &finalizer.journal_key,
10009 crate::journal::LIFECYCLE_OUTBOX_DOMAIN,
10010 0,
10011 &outbox,
10012 )
10013 .unwrap();
10014 std::fs::write(path, sealed).unwrap();
10015 }
10016
10017 let sessions = SessionRegistry::new();
10018 finalizer
10019 .remove_acked_lifecycle_outboxes(&sessions)
10020 .await
10021 .unwrap();
10022
10023 let scrape = metrics.render();
10024 assert!(
10025 scrape.contains("av_lifecycle_outbox_pending 3"),
10026 "gauge must reflect the 3 unacked outboxes; scrape:\n{scrape}"
10027 );
10028 }
10029}