1use std::sync::Arc;
24use std::time::Duration;
25
26use serde::{Deserialize, Serialize};
27use zeph_common::task_supervisor::TaskSupervisor;
28use zeph_common::text::truncate_to_bytes_ref;
29
30#[non_exhaustive]
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
43#[serde(rename_all = "snake_case")]
44pub enum PageType {
45 ToolOutput,
47 ConversationTurn,
49 MemoryExcerpt,
51 SystemContext,
53}
54
55impl std::fmt::Display for PageType {
56 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
57 match self {
58 Self::ToolOutput => f.write_str("tool_output"),
59 Self::ConversationTurn => f.write_str("conversation_turn"),
60 Self::MemoryExcerpt => f.write_str("memory_excerpt"),
61 Self::SystemContext => f.write_str("system_context"),
62 }
63 }
64}
65
66#[derive(Debug, Clone, Serialize, Deserialize)]
70#[serde(tag = "kind", rename_all = "snake_case")]
71#[non_exhaustive]
72pub enum PageOrigin {
73 ToolPair {
75 tool_name: String,
77 },
78 Turn {
80 message_id: String,
82 },
83 Excerpt {
85 source_label: String,
87 },
88 System {
90 key: String,
92 },
93}
94
95#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
101#[serde(rename_all = "snake_case")]
102#[non_exhaustive]
103pub enum SchemaHint {
104 Json,
106 Text,
108 Diff,
110 Table,
112 Binary,
114}
115
116#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
132pub struct PageId(pub String);
133
134impl PageId {
135 #[must_use]
137 pub fn compute(page_type: PageType, origin_key: &str, body: &[u8]) -> Self {
138 let mut hasher = blake3::Hasher::new();
139 hasher.update(page_type.to_string().as_bytes());
140 hasher.update(b"|");
141 hasher.update(origin_key.as_bytes());
142 hasher.update(b"|");
143 hasher.update(body);
144 let hash = hasher.finalize();
145 let mut hex = String::with_capacity(32);
147 for b in &hash.as_bytes()[..16] {
148 use std::fmt::Write as _;
149 let _ = write!(hex, "{b:02x}");
150 }
151 Self(format!("blake3:{hex}"))
152 }
153}
154
155#[derive(Debug, Clone)]
163pub struct TypedPage {
164 pub page_id: PageId,
166 pub page_type: PageType,
168 pub origin: PageOrigin,
170 pub tokens: u32,
172 pub body: Arc<str>,
174 pub schema_hint: Option<SchemaHint>,
176}
177
178impl TypedPage {
179 #[must_use]
181 pub fn new(
182 page_type: PageType,
183 origin: PageOrigin,
184 tokens: u32,
185 body: Arc<str>,
186 schema_hint: Option<SchemaHint>,
187 ) -> Self {
188 let origin_key = origin_key_for(&origin);
189 let page_id = PageId::compute(page_type, &origin_key, body.as_bytes());
190 Self {
191 page_id,
192 page_type,
193 origin,
194 tokens,
195 body,
196 schema_hint,
197 }
198 }
199}
200
201fn origin_key_for(origin: &PageOrigin) -> String {
202 match origin {
203 PageOrigin::ToolPair { tool_name } => format!("tool:{tool_name}"),
204 PageOrigin::Turn { message_id } => format!("turn:{message_id}"),
205 PageOrigin::Excerpt { source_label } => format!("excerpt:{source_label}"),
206 PageOrigin::System { key } => format!("system:{key}"),
207 }
208}
209
210#[derive(Debug, Clone)]
217pub struct FidelityContract {
218 pub fidelity_level: &'static str,
220 pub invariant_version: u8,
222 pub required_fields: &'static [&'static str],
224}
225
226#[derive(Debug, Clone, Serialize)]
233pub struct FidelityViolation {
234 pub missing_field: String,
236 pub detail: String,
238}
239
240#[derive(Debug, Clone)]
244pub struct CompactedPage {
245 pub body: Arc<str>,
247 pub tokens: u32,
249}
250
251pub trait PageInvariant: Send + Sync {
270 fn page_type(&self) -> PageType;
272
273 fn minimum_fidelity(&self, page: &TypedPage) -> FidelityContract;
275
276 fn verify(
283 &self,
284 original: &TypedPage,
285 compacted: &CompactedPage,
286 ) -> Result<(), Vec<FidelityViolation>>;
287}
288
289pub struct ToolOutputInvariant;
296
297impl PageInvariant for ToolOutputInvariant {
298 fn page_type(&self) -> PageType {
299 PageType::ToolOutput
300 }
301
302 fn minimum_fidelity(&self, _page: &TypedPage) -> FidelityContract {
303 FidelityContract {
304 fidelity_level: "structured_summary_v1",
305 invariant_version: 1,
306 required_fields: &["tool_name", "exit_status"],
307 }
308 }
309
310 fn verify(
311 &self,
312 original: &TypedPage,
313 compacted: &CompactedPage,
314 ) -> Result<(), Vec<FidelityViolation>> {
315 let body = compacted.body.as_ref();
316 if original.schema_hint == Some(SchemaHint::Binary) {
318 return Ok(());
319 }
320
321 let mut violations = Vec::new();
322
323 let tool_name = match &original.origin {
325 PageOrigin::ToolPair { tool_name } => tool_name.as_str(),
326 _ => "",
327 };
328 if !tool_name.is_empty() && !body.contains(tool_name) {
329 violations.push(FidelityViolation {
330 missing_field: "tool_name".into(),
331 detail: format!("compacted body does not reference tool '{tool_name}'"),
332 });
333 }
334
335 let has_status = body.contains("exit_status")
337 || body.contains("exit_code")
338 || body.contains("status:")
339 || body.contains("Status:")
340 || body.contains("exit:")
341 || body.contains("rc:");
342 if !has_status {
343 violations.push(FidelityViolation {
344 missing_field: "exit_status".into(),
345 detail: "compacted body does not contain an exit status indicator".into(),
346 });
347 }
348
349 if original.schema_hint == Some(SchemaHint::Json) {
353 let original_body = original.body.as_ref();
354 let preserved = check_json_structural_key(original_body, body);
355 if !preserved {
356 violations.push(FidelityViolation {
357 missing_field: "structural_key".into(),
358 detail: "compacted JSON tool output does not reference any top-level field \
359 name from the original output"
360 .into(),
361 });
362 }
363 }
364
365 if violations.is_empty() {
366 Ok(())
367 } else {
368 Err(violations)
369 }
370 }
371}
372
373fn check_json_structural_key(original: &str, compacted: &str) -> bool {
379 let Ok(value) = serde_json::from_str::<serde_json::Value>(original) else {
380 return true;
381 };
382 let Some(obj) = value.as_object() else {
383 return true;
384 };
385 if obj.is_empty() {
386 return true;
387 }
388 obj.keys().any(|k| compacted.contains(k.as_str()))
389}
390
391pub struct ConversationTurnInvariant;
395
396impl PageInvariant for ConversationTurnInvariant {
397 fn page_type(&self) -> PageType {
398 PageType::ConversationTurn
399 }
400
401 fn minimum_fidelity(&self, _page: &TypedPage) -> FidelityContract {
402 FidelityContract {
403 fidelity_level: "semantic_summary_v1",
404 invariant_version: 1,
405 required_fields: &["role"],
406 }
407 }
408
409 fn verify(
410 &self,
411 _original: &TypedPage,
412 compacted: &CompactedPage,
413 ) -> Result<(), Vec<FidelityViolation>> {
414 let body = compacted.body.as_ref();
415 let has_role =
416 body.contains("user") || body.contains("assistant") || body.contains("system");
417 if !has_role {
418 return Err(vec![FidelityViolation {
419 missing_field: "role".into(),
420 detail: "compacted turn does not identify a speaker role".into(),
421 }]);
422 }
423 Ok(())
424 }
425}
426
427pub struct MemoryExcerptInvariant;
431
432impl PageInvariant for MemoryExcerptInvariant {
433 fn page_type(&self) -> PageType {
434 PageType::MemoryExcerpt
435 }
436
437 fn minimum_fidelity(&self, _page: &TypedPage) -> FidelityContract {
438 FidelityContract {
439 fidelity_level: "excerpt_summary_v1",
440 invariant_version: 1,
441 required_fields: &["source_label"],
442 }
443 }
444
445 fn verify(
446 &self,
447 original: &TypedPage,
448 compacted: &CompactedPage,
449 ) -> Result<(), Vec<FidelityViolation>> {
450 let source_label = match &original.origin {
451 PageOrigin::Excerpt { source_label } => source_label.as_str(),
452 _ => return Ok(()),
453 };
454 if !compacted.body.contains(source_label) {
455 return Err(vec![FidelityViolation {
456 missing_field: "source_label".into(),
457 detail: format!("compacted excerpt does not contain source label '{source_label}'"),
458 }]);
459 }
460 Ok(())
461 }
462}
463
464pub struct SystemContextInvariant;
469
470pub const SYSTEM_POINTER_PREFIX: &str = "[system-ptr:";
472
473impl PageInvariant for SystemContextInvariant {
474 fn page_type(&self) -> PageType {
475 PageType::SystemContext
476 }
477
478 fn minimum_fidelity(&self, _page: &TypedPage) -> FidelityContract {
479 FidelityContract {
480 fidelity_level: "pointer_replace_v1",
481 invariant_version: 1,
482 required_fields: &["pointer"],
483 }
484 }
485
486 fn verify(
487 &self,
488 _original: &TypedPage,
489 compacted: &CompactedPage,
490 ) -> Result<(), Vec<FidelityViolation>> {
491 if !compacted.body.starts_with(SYSTEM_POINTER_PREFIX) {
492 return Err(vec![FidelityViolation {
493 missing_field: "pointer".into(),
494 detail: format!(
495 "SystemContext page was not pointer-replaced \
496 (body does not start with '{SYSTEM_POINTER_PREFIX}')"
497 ),
498 }]);
499 }
500 Ok(())
501 }
502}
503
504pub struct InvariantRegistry {
520 tool_output: Box<dyn PageInvariant>,
521 conversation_turn: Box<dyn PageInvariant>,
522 memory_excerpt: Box<dyn PageInvariant>,
523 system_context: Box<dyn PageInvariant>,
524}
525
526impl Default for InvariantRegistry {
527 fn default() -> Self {
528 Self {
529 tool_output: Box::new(ToolOutputInvariant),
530 conversation_turn: Box::new(ConversationTurnInvariant),
531 memory_excerpt: Box::new(MemoryExcerptInvariant),
532 system_context: Box::new(SystemContextInvariant),
533 }
534 }
535}
536
537impl InvariantRegistry {
538 #[must_use]
542 pub fn get(&self, page_type: PageType) -> Option<&dyn PageInvariant> {
543 match page_type {
544 PageType::ToolOutput => Some(self.tool_output.as_ref()),
545 PageType::ConversationTurn => Some(self.conversation_turn.as_ref()),
546 PageType::MemoryExcerpt => Some(self.memory_excerpt.as_ref()),
547 PageType::SystemContext => Some(self.system_context.as_ref()),
548 }
549 }
550
551 pub fn enforce(
562 &self,
563 original: &TypedPage,
564 compacted: &CompactedPage,
565 ) -> Result<(), Vec<FidelityViolation>> {
566 let _span = tracing::info_span!(
567 "context.compaction.typed_page",
568 page_type = %original.page_type,
569 page_id = %original.page_id.0,
570 original_tokens = original.tokens,
571 compacted_tokens = compacted.tokens,
572 )
573 .entered();
574
575 if let Some(inv) = self.get(original.page_type) {
576 inv.verify(original, compacted)
577 } else {
578 tracing::warn!(
579 page_type = %original.page_type,
580 "no invariant registered for page type — skipping verification"
581 );
582 Ok(())
583 }
584 }
585}
586
587#[must_use]
616pub fn classify(body: &str) -> PageType {
617 classify_with_role(body, false)
618}
619
620#[must_use]
640pub fn classify_with_role(body: &str, is_system_role: bool) -> PageType {
641 tracing::info_span!(
642 "context.compaction.typed_page.classify",
643 body_len = body.len()
644 )
645 .in_scope(|| classify_with_role_inner(body, is_system_role))
646}
647
648fn classify_with_role_inner(body: &str, is_system_role: bool) -> PageType {
649 const TOOL_PREFIXES: &[&str] = &["[tool_output]", "[tool:", "[Tool output]"];
651 const MEMORY_PREFIXES: &[&str] = &[
652 "[cross-session context]",
653 "[semantic recall]",
654 "[known facts]",
655 "[conversation summaries]",
656 "[past corrections]",
657 "## Relevant documents",
658 ];
659 const SYSTEM_PREFIXES: &[&str] = &[
660 "[Persona context]",
661 "[Past experience]",
662 "[Memory summary]",
663 "[system",
664 "[skill",
665 "[persona",
666 "[digest",
667 "[compression",
668 ];
669
670 let trimmed = body.trim_start();
671
672 for prefix in TOOL_PREFIXES {
673 if trimmed.starts_with(prefix) {
674 return PageType::ToolOutput;
675 }
676 }
677 for prefix in MEMORY_PREFIXES {
678 if trimmed.starts_with(prefix) {
679 return PageType::MemoryExcerpt;
680 }
681 }
682 for prefix in SYSTEM_PREFIXES {
683 if trimmed.starts_with(prefix) {
684 return PageType::SystemContext;
685 }
686 }
687
688 if is_system_role {
693 return PageType::SystemContext;
694 }
695
696 tracing::warn!(
697 body_prefix = truncate_to_bytes_ref(body, 80),
698 "typed-page classification fallback to ConversationTurn"
699 );
700 PageType::ConversationTurn
701}
702
703#[must_use]
709pub fn detect_schema_hint(body: &str, is_binary: bool) -> SchemaHint {
710 if is_binary || body.contains('\u{FFFD}') {
711 return SchemaHint::Binary;
712 }
713 let trimmed = body.trim_start();
714 if trimmed.starts_with('{') || trimmed.starts_with('[') {
715 return SchemaHint::Json;
716 }
717 if trimmed.starts_with("--- ")
718 || trimmed.starts_with("+++ ")
719 || trimmed.starts_with("diff --git")
720 || trimmed.starts_with("diff -")
721 {
722 return SchemaHint::Diff;
723 }
724 let first_line = trimmed.lines().next().unwrap_or("");
726 if first_line.matches('\t').count() >= 2 || first_line.matches('|').count() >= 2 {
727 return SchemaHint::Table;
728 }
729 SchemaHint::Text
730}
731
732#[derive(Debug, Serialize)]
739pub struct CompactedPageRecord {
740 pub ts: String,
742 pub turn_id: String,
744 pub page_id: String,
746 pub page_type: PageType,
748 pub origin: PageOrigin,
750 pub original_tokens: u32,
752 pub compacted_tokens: u32,
754 pub fidelity_level: String,
756 pub invariant_version: u8,
758 pub provider_name: String,
760 pub violations: Vec<FidelityViolation>,
762 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
764 pub classification_fallback: bool,
765}
766
767#[derive(Debug, Clone, Serialize)]
771pub struct BatchViolation {
772 pub assertion: String,
774 pub detail: String,
776}
777
778#[derive(Debug, Clone, Default)]
802pub struct BatchAssertions {
803 pub tool_names_in_batch: Vec<String>,
805 pub has_memory_excerpt: bool,
807 pub excerpt_labels: Vec<String>,
809}
810
811impl BatchAssertions {
812 #[must_use]
816 pub fn check(&self, summary: &str) -> Vec<BatchViolation> {
817 let mut violations = Vec::new();
818
819 if !self.tool_names_in_batch.is_empty() {
821 let any_tool_mentioned = self
822 .tool_names_in_batch
823 .iter()
824 .any(|name| !name.is_empty() && summary.contains(name.as_str()));
825 if !any_tool_mentioned {
826 violations.push(BatchViolation {
827 assertion: "tool_coverage".into(),
828 detail: format!(
829 "summary mentions none of the {} tool(s) in batch: {:?}",
830 self.tool_names_in_batch.len(),
831 self.tool_names_in_batch
832 ),
833 });
834 }
835 }
836
837 if self.has_memory_excerpt && !self.excerpt_labels.is_empty() {
839 let any_label_mentioned = self
840 .excerpt_labels
841 .iter()
842 .any(|label| !label.is_empty() && summary.contains(label.as_str()));
843 if !any_label_mentioned {
844 violations.push(BatchViolation {
845 assertion: "excerpt_label_coverage".into(),
846 detail: format!(
847 "summary mentions none of the memory excerpt labels: {:?}",
848 self.excerpt_labels
849 ),
850 });
851 }
852 }
853
854 violations
855 }
856}
857
858pub struct TypedPagesState {
865 pub registry: InvariantRegistry,
867 pub audit_sink: Option<CompactionAuditSink>,
869 pub is_active: bool,
872}
873
874enum AuditCommand {
878 Record(CompactedPageRecord),
880 Flush(tokio::sync::oneshot::Sender<()>),
882}
883
884#[derive(Debug, Clone)]
912pub struct CompactionAuditSink {
913 tx: tokio::sync::mpsc::Sender<AuditCommand>,
914 drop_counter: Arc<std::sync::atomic::AtomicU64>,
915}
916
917impl CompactionAuditSink {
918 #[tracing::instrument(name = "context.typed_page.open", skip_all)]
931 pub async fn open(
932 path: &std::path::Path,
933 capacity: usize,
934 supervisor: Option<&TaskSupervisor>,
935 ) -> Result<Self, std::io::Error> {
936 use tokio::io::AsyncWriteExt as _;
937
938 if let Some(parent) = path.parent() {
939 tokio::fs::create_dir_all(parent).await?;
940 }
941 let file = tokio::fs::OpenOptions::new()
942 .create(true)
943 .append(true)
944 .open(path)
945 .await?;
946
947 let (tx, mut rx) = tokio::sync::mpsc::channel::<AuditCommand>(capacity.max(1));
948 let drop_counter = Arc::new(std::sync::atomic::AtomicU64::new(0));
949 let drop_counter_bg = drop_counter.clone();
950
951 let fut = async move {
952 let mut writer = tokio::io::BufWriter::new(file);
953 while let Some(cmd) = rx.recv().await {
954 match cmd {
955 AuditCommand::Record(record) => match serde_json::to_string(&record) {
956 Ok(mut line) => {
957 line.push('\n');
958 if let Err(e) = writer.write_all(line.as_bytes()).await {
959 tracing::error!("compaction audit write failed: {e:#}");
960 }
961 }
962 Err(e) => {
963 tracing::error!("compaction audit serialization failed: {e:#}");
964 }
965 },
966 AuditCommand::Flush(responder) => {
967 let _ = writer.flush().await;
968 let _ = responder.send(());
969 }
970 }
971 }
972 let _ = writer.flush().await;
974
975 let dropped = drop_counter_bg.load(std::sync::atomic::Ordering::Relaxed);
976 if dropped > 0 {
977 tracing::warn!(dropped, "compaction audit sink closed with dropped records");
978 }
979 };
980
981 if let Some(sup) = supervisor {
982 drop(sup.spawn_oneshot(Arc::from("context.audit_sink"), move || fut));
983 } else {
984 tokio::spawn(fut); }
986
987 Ok(Self { tx, drop_counter })
988 }
989
990 pub fn send(&self, record: CompactedPageRecord) {
994 match self.tx.try_send(AuditCommand::Record(record)) {
995 Ok(()) => {}
996 Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
997 let prev = self
998 .drop_counter
999 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1000 tracing::warn!(
1001 dropped_total = prev + 1,
1002 "compaction audit sink full — record dropped (best-effort audit)"
1003 );
1004 }
1005 Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
1006 tracing::error!("compaction audit sink closed unexpectedly");
1007 }
1008 }
1009 }
1010
1011 #[tracing::instrument(name = "context.typed_page.flush", skip_all)]
1017 pub async fn flush(&self) {
1018 let (tx, rx) = tokio::sync::oneshot::channel::<()>();
1019 if self.tx.send(AuditCommand::Flush(tx)).await.is_ok() {
1020 let _ = tokio::time::timeout(Duration::from_millis(100), rx).await;
1021 }
1022 }
1023
1024 #[must_use]
1026 pub fn dropped_count(&self) -> u64 {
1027 self.drop_counter.load(std::sync::atomic::Ordering::Relaxed)
1028 }
1029}
1030
1031#[cfg(test)]
1034mod tests {
1035 use super::*;
1036
1037 #[test]
1040 fn page_id_same_input_same_output() {
1041 let a = PageId::compute(PageType::ToolOutput, "tool:shell", b"exit_code: 0");
1042 let b = PageId::compute(PageType::ToolOutput, "tool:shell", b"exit_code: 0");
1043 assert_eq!(a, b);
1044 }
1045
1046 #[test]
1047 fn page_id_different_type_different_id() {
1048 let a = PageId::compute(PageType::ToolOutput, "tool:shell", b"body");
1049 let b = PageId::compute(PageType::ConversationTurn, "tool:shell", b"body");
1050 assert_ne!(a, b);
1051 }
1052
1053 #[test]
1054 fn page_id_starts_with_blake3_prefix() {
1055 let id = PageId::compute(PageType::SystemContext, "system:persona", b"some content");
1056 assert!(id.0.starts_with("blake3:"));
1057 }
1058
1059 #[test]
1062 fn classify_tool_output_prefix() {
1063 assert_eq!(
1064 classify("[tool_output] shell exit_code: 0"),
1065 PageType::ToolOutput
1066 );
1067 assert_eq!(classify("[tool:shell] result"), PageType::ToolOutput);
1068 }
1069
1070 #[test]
1071 fn classify_memory_prefixes() {
1072 assert_eq!(
1073 classify("[cross-session context]\nsome recall"),
1074 PageType::MemoryExcerpt
1075 );
1076 assert_eq!(
1077 classify("[semantic recall]\n- [user] hello"),
1078 PageType::MemoryExcerpt
1079 );
1080 assert_eq!(classify("[known facts]\n- fact"), PageType::MemoryExcerpt);
1081 assert_eq!(
1082 classify("[conversation summaries]\n- 1-10: summary"),
1083 PageType::MemoryExcerpt
1084 );
1085 }
1086
1087 #[test]
1088 fn classify_system_prefixes() {
1089 assert_eq!(classify("[Persona context]\nfact"), PageType::SystemContext);
1090 assert_eq!(classify("[system prompt]"), PageType::SystemContext);
1091 }
1092
1093 #[test]
1094 fn classify_fallback_is_conversation_turn() {
1095 assert_eq!(classify("Hello, world!"), PageType::ConversationTurn);
1096 assert_eq!(classify(""), PageType::ConversationTurn);
1097 }
1098
1099 #[test]
1102 fn detect_schema_hint_json() {
1103 assert_eq!(
1104 detect_schema_hint(r#"{"key": "val"}"#, false),
1105 SchemaHint::Json
1106 );
1107 assert_eq!(detect_schema_hint("[1,2,3]", false), SchemaHint::Json);
1108 }
1109
1110 #[test]
1111 fn detect_schema_hint_diff() {
1112 assert_eq!(detect_schema_hint("--- a\n+++ b", false), SchemaHint::Diff);
1113 }
1114
1115 #[test]
1116 fn detect_schema_hint_binary() {
1117 assert_eq!(detect_schema_hint("anything", true), SchemaHint::Binary);
1118 }
1119
1120 #[test]
1121 fn detect_schema_hint_text_fallback() {
1122 assert_eq!(detect_schema_hint("plain text", false), SchemaHint::Text);
1123 }
1124
1125 #[test]
1128 fn tool_output_invariant_passes_when_fields_present() {
1129 let inv = ToolOutputInvariant;
1130 let page = TypedPage::new(
1131 PageType::ToolOutput,
1132 PageOrigin::ToolPair {
1133 tool_name: "shell".into(),
1134 },
1135 100,
1136 Arc::from("[tool_output] shell exit_code: 0\nsome output"),
1137 Some(SchemaHint::Text),
1138 );
1139 let compacted = CompactedPage {
1140 body: Arc::from("shell exit_status: 0\nkey: value"),
1141 tokens: 10,
1142 };
1143 assert!(inv.verify(&page, &compacted).is_ok());
1144 }
1145
1146 #[test]
1147 fn tool_output_invariant_fails_missing_tool_name() {
1148 let inv = ToolOutputInvariant;
1149 let page = TypedPage::new(
1150 PageType::ToolOutput,
1151 PageOrigin::ToolPair {
1152 tool_name: "my_tool".into(),
1153 },
1154 100,
1155 Arc::from("[tool_output] my_tool exit_code: 0"),
1156 Some(SchemaHint::Text),
1157 );
1158 let compacted = CompactedPage {
1159 body: Arc::from("exit_status: 0"),
1160 tokens: 5,
1161 };
1162 let result = inv.verify(&page, &compacted);
1163 assert!(result.is_err());
1164 let violations = result.unwrap_err();
1165 assert!(violations.iter().any(|v| v.missing_field == "tool_name"));
1166 }
1167
1168 #[test]
1169 fn tool_output_invariant_passes_for_binary() {
1170 let inv = ToolOutputInvariant;
1171 let page = TypedPage::new(
1172 PageType::ToolOutput,
1173 PageOrigin::ToolPair {
1174 tool_name: "binary_tool".into(),
1175 },
1176 100,
1177 Arc::from("<binary:1024 bytes>"),
1178 Some(SchemaHint::Binary),
1179 );
1180 let compacted = CompactedPage {
1181 body: Arc::from("<binary:1024 bytes> (archived)"),
1182 tokens: 5,
1183 };
1184 assert!(inv.verify(&page, &compacted).is_ok());
1185 }
1186
1187 #[test]
1190 fn system_context_invariant_passes_with_pointer() {
1191 let inv = SystemContextInvariant;
1192 let page = TypedPage::new(
1193 PageType::SystemContext,
1194 PageOrigin::System {
1195 key: "persona".into(),
1196 },
1197 200,
1198 Arc::from("[Persona context]\nsome persona info"),
1199 None,
1200 );
1201 let compacted = CompactedPage {
1202 body: Arc::from("[system-ptr:blake3:abcdef123456]"),
1203 tokens: 3,
1204 };
1205 assert!(inv.verify(&page, &compacted).is_ok());
1206 }
1207
1208 #[test]
1209 fn system_context_invariant_fails_without_pointer() {
1210 let inv = SystemContextInvariant;
1211 let page = TypedPage::new(
1212 PageType::SystemContext,
1213 PageOrigin::System {
1214 key: "persona".into(),
1215 },
1216 200,
1217 Arc::from("[Persona context]\nsome persona info"),
1218 None,
1219 );
1220 let compacted = CompactedPage {
1221 body: Arc::from("This is a paraphrase of persona info"),
1222 tokens: 10,
1223 };
1224 let result = inv.verify(&page, &compacted);
1225 assert!(result.is_err());
1226 let violations = result.unwrap_err();
1227 assert!(violations.iter().any(|v| v.missing_field == "pointer"));
1228 }
1229
1230 #[test]
1233 fn registry_covers_all_page_types() {
1234 let reg = InvariantRegistry::default();
1235 for pt in [
1236 PageType::ToolOutput,
1237 PageType::ConversationTurn,
1238 PageType::MemoryExcerpt,
1239 PageType::SystemContext,
1240 ] {
1241 assert!(reg.get(pt).is_some(), "missing invariant for {pt:?}");
1242 }
1243 }
1244
1245 #[test]
1246 fn registry_returns_correct_page_type() {
1247 let reg = InvariantRegistry::default();
1248 assert_eq!(
1249 reg.get(PageType::ToolOutput).unwrap().page_type(),
1250 PageType::ToolOutput
1251 );
1252 assert_eq!(
1253 reg.get(PageType::SystemContext).unwrap().page_type(),
1254 PageType::SystemContext
1255 );
1256 }
1257
1258 #[test]
1261 fn enforce_ok_for_valid_system_pointer() {
1262 let reg = InvariantRegistry::default();
1263 let page = TypedPage::new(
1264 PageType::SystemContext,
1265 PageOrigin::System {
1266 key: "persona".into(),
1267 },
1268 50,
1269 Arc::from("[Persona context]\nrules"),
1270 None,
1271 );
1272 let compacted = CompactedPage {
1273 body: Arc::from("[system-ptr:blake3:aabbccdd11223344]"),
1274 tokens: 3,
1275 };
1276 assert!(reg.enforce(&page, &compacted).is_ok());
1277 }
1278
1279 #[test]
1280 fn enforce_err_for_paraphrased_system_context() {
1281 let reg = InvariantRegistry::default();
1282 let page = TypedPage::new(
1283 PageType::SystemContext,
1284 PageOrigin::System {
1285 key: "persona".into(),
1286 },
1287 50,
1288 Arc::from("[Persona context]\nrules"),
1289 None,
1290 );
1291 let compacted = CompactedPage {
1292 body: Arc::from("The persona says to be helpful."),
1293 tokens: 7,
1294 };
1295 let result = reg.enforce(&page, &compacted);
1296 assert!(result.is_err());
1297 assert!(
1298 result
1299 .unwrap_err()
1300 .iter()
1301 .any(|v| v.missing_field == "pointer")
1302 );
1303 }
1304
1305 #[test]
1306 fn enforce_ok_for_conversation_turn_with_role() {
1307 let reg = InvariantRegistry::default();
1308 let page = TypedPage::new(
1309 PageType::ConversationTurn,
1310 PageOrigin::Turn {
1311 message_id: "42".into(),
1312 },
1313 30,
1314 Arc::from("Hello from user"),
1315 None,
1316 );
1317 let compacted = CompactedPage {
1318 body: Arc::from("user asked about Rust"),
1319 tokens: 5,
1320 };
1321 assert!(reg.enforce(&page, &compacted).is_ok());
1322 }
1323
1324 #[test]
1327 fn memory_excerpt_invariant_passes_when_label_present() {
1328 let inv = MemoryExcerptInvariant;
1329 let label = "semantic_recall";
1330 let page = TypedPage::new(
1331 PageType::MemoryExcerpt,
1332 PageOrigin::Excerpt {
1333 source_label: label.into(),
1334 },
1335 80,
1336 Arc::from("[semantic recall]\n- [user] hello"),
1337 None,
1338 );
1339 let compacted = CompactedPage {
1340 body: Arc::from(format!("Summary from {label}: user greeted")),
1341 tokens: 6,
1342 };
1343 assert!(inv.verify(&page, &compacted).is_ok());
1344 }
1345
1346 #[test]
1347 fn memory_excerpt_invariant_fails_when_label_missing() {
1348 let inv = MemoryExcerptInvariant;
1349 let page = TypedPage::new(
1350 PageType::MemoryExcerpt,
1351 PageOrigin::Excerpt {
1352 source_label: "graph_facts".into(),
1353 },
1354 80,
1355 Arc::from("[known facts]\n- Alice works at Acme"),
1356 None,
1357 );
1358 let compacted = CompactedPage {
1359 body: Arc::from("Alice is employed somewhere"),
1360 tokens: 5,
1361 };
1362 let result = inv.verify(&page, &compacted);
1363 assert!(result.is_err());
1364 assert!(
1365 result
1366 .unwrap_err()
1367 .iter()
1368 .any(|v| v.missing_field == "source_label")
1369 );
1370 }
1371
1372 #[test]
1373 fn memory_excerpt_invariant_passes_for_non_excerpt_origin() {
1374 let inv = MemoryExcerptInvariant;
1375 let page = TypedPage::new(
1376 PageType::MemoryExcerpt,
1377 PageOrigin::System {
1378 key: "digests".into(),
1379 },
1380 40,
1381 Arc::from("[system]"),
1382 None,
1383 );
1384 let compacted = CompactedPage {
1385 body: Arc::from("anything"),
1386 tokens: 1,
1387 };
1388 assert!(inv.verify(&page, &compacted).is_ok());
1389 }
1390
1391 #[test]
1394 fn conversation_turn_invariant_passes_with_role_word() {
1395 let inv = ConversationTurnInvariant;
1396 let page = TypedPage::new(
1397 PageType::ConversationTurn,
1398 PageOrigin::Turn {
1399 message_id: "1".into(),
1400 },
1401 20,
1402 Arc::from("Hello world"),
1403 None,
1404 );
1405 for body in &["user: hi", "assistant replied", "system note"] {
1406 let compacted = CompactedPage {
1407 body: Arc::from(*body),
1408 tokens: 2,
1409 };
1410 assert!(inv.verify(&page, &compacted).is_ok(), "body={body}");
1411 }
1412 }
1413
1414 #[test]
1415 fn conversation_turn_invariant_fails_without_role_word() {
1416 let inv = ConversationTurnInvariant;
1417 let page = TypedPage::new(
1418 PageType::ConversationTurn,
1419 PageOrigin::Turn {
1420 message_id: "2".into(),
1421 },
1422 20,
1423 Arc::from("some turn content"),
1424 None,
1425 );
1426 let compacted = CompactedPage {
1427 body: Arc::from("content was summarized"),
1428 tokens: 3,
1429 };
1430 let result = inv.verify(&page, &compacted);
1431 assert!(result.is_err());
1432 assert!(
1433 result
1434 .unwrap_err()
1435 .iter()
1436 .any(|v| v.missing_field == "role")
1437 );
1438 }
1439
1440 #[tokio::test]
1443 async fn audit_sink_jsonl_roundtrip() {
1444 let dir = tempfile::tempdir().unwrap();
1445 let path = dir.path().join("audit.jsonl");
1446
1447 let sink = CompactionAuditSink::open(&path, 64, None).await.unwrap();
1448 let record = CompactedPageRecord {
1449 ts: "2026-04-19T00:00:00Z".into(),
1450 turn_id: "1".into(),
1451 page_id: "blake3:aabbccdd".into(),
1452 page_type: PageType::ToolOutput,
1453 origin: PageOrigin::ToolPair {
1454 tool_name: "shell".into(),
1455 },
1456 original_tokens: 100,
1457 compacted_tokens: 20,
1458 fidelity_level: "structured_summary_v1".into(),
1459 invariant_version: 1,
1460 provider_name: "test".into(),
1461 violations: vec![],
1462 classification_fallback: false,
1463 };
1464 sink.send(record);
1465
1466 drop(sink);
1468 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1470
1471 let contents = std::fs::read_to_string(&path).unwrap();
1472 assert!(!contents.is_empty(), "audit file should not be empty");
1473 let parsed: serde_json::Value = serde_json::from_str(contents.trim()).unwrap();
1474 assert_eq!(parsed["page_type"], "tool_output");
1475 assert_eq!(parsed["turn_id"], "1");
1476 assert_eq!(parsed["provider_name"], "test");
1477 }
1478
1479 #[tokio::test]
1480 async fn audit_sink_drop_counter_increments_when_full() {
1481 let dir = tempfile::tempdir().unwrap();
1482 let path = dir.path().join("audit_full.jsonl");
1483
1484 let sink = CompactionAuditSink::open(&path, 1, None).await.unwrap();
1486
1487 let make_record = || CompactedPageRecord {
1488 ts: "2026-04-19T00:00:00Z".into(),
1489 turn_id: "x".into(),
1490 page_id: "blake3:00".into(),
1491 page_type: PageType::ConversationTurn,
1492 origin: PageOrigin::Turn {
1493 message_id: "0".into(),
1494 },
1495 original_tokens: 10,
1496 compacted_tokens: 5,
1497 fidelity_level: "semantic_summary_v1".into(),
1498 invariant_version: 1,
1499 provider_name: "test".into(),
1500 violations: vec![],
1501 classification_fallback: false,
1502 };
1503
1504 for _ in 0..10 {
1506 sink.send(make_record());
1507 }
1508
1509 assert!(
1510 sink.dropped_count() > 0,
1511 "expected at least one dropped record"
1512 );
1513 }
1514
1515 #[tokio::test]
1516 async fn audit_sink_flush_does_not_panic() {
1517 let dir = tempfile::tempdir().unwrap();
1518 let path = dir.path().join("audit_flush.jsonl");
1519 let sink = CompactionAuditSink::open(&path, 16, None).await.unwrap();
1520 sink.flush().await;
1522 }
1523
1524 #[test]
1527 fn classify_with_role_system_flag_overrides_fallback() {
1528 assert_eq!(
1529 classify_with_role("You are a helpful assistant.", true),
1530 PageType::SystemContext
1531 );
1532 }
1533
1534 #[test]
1535 fn classify_with_role_prefix_wins_over_system_flag() {
1536 assert_eq!(
1537 classify_with_role("[tool_output] exit_code: 0", false),
1538 PageType::ToolOutput
1539 );
1540 }
1541
1542 #[test]
1543 fn classify_with_role_false_still_falls_back_to_conversation_turn() {
1544 assert_eq!(
1545 classify_with_role("random prose without markers", false),
1546 PageType::ConversationTurn
1547 );
1548 }
1549
1550 #[test]
1553 fn tool_output_json_structural_check_passes_when_key_preserved() {
1554 let inv = ToolOutputInvariant;
1555 let original_body = r#"{"exit_code": 0, "stdout": "ok"}"#;
1556 let page = TypedPage::new(
1557 PageType::ToolOutput,
1558 PageOrigin::ToolPair {
1559 tool_name: "shell".into(),
1560 },
1561 50,
1562 Arc::from(original_body),
1563 Some(SchemaHint::Json),
1564 );
1565 let compacted = CompactedPage {
1567 body: Arc::from("shell exit_code: 0, stdout was ok"),
1568 tokens: 8,
1569 };
1570 assert!(inv.verify(&page, &compacted).is_ok());
1571 }
1572
1573 #[test]
1574 fn tool_output_json_structural_check_fails_when_no_key_preserved() {
1575 let inv = ToolOutputInvariant;
1576 let original_body = r#"{"some_field": "value", "other_field": 42}"#;
1577 let page = TypedPage::new(
1578 PageType::ToolOutput,
1579 PageOrigin::ToolPair {
1580 tool_name: "my_tool".into(),
1581 },
1582 50,
1583 Arc::from(original_body),
1584 Some(SchemaHint::Json),
1585 );
1586 let compacted = CompactedPage {
1588 body: Arc::from("my_tool exit_status: 0 completed successfully"),
1589 tokens: 7,
1590 };
1591 let result = inv.verify(&page, &compacted);
1592 assert!(result.is_err());
1593 let violations = result.unwrap_err();
1594 assert!(
1595 violations
1596 .iter()
1597 .any(|v| v.missing_field == "structural_key")
1598 );
1599 }
1600
1601 #[tokio::test]
1604 async fn audit_sink_capacity_zero_does_not_panic() {
1605 let dir = tempfile::tempdir().unwrap();
1606 let path = dir.path().join("cap0.jsonl");
1607 let sink = CompactionAuditSink::open(&path, 0, None).await.unwrap();
1609 sink.flush().await;
1610 }
1611
1612 #[tokio::test]
1613 async fn audit_sink_open_with_supervisor_registers_task() {
1614 use tokio_util::sync::CancellationToken;
1615 use zeph_common::TaskSupervisor;
1616
1617 let dir = tempfile::tempdir().unwrap();
1618 let path = dir.path().join("audit_sup.jsonl");
1619 let cancel = CancellationToken::new();
1620 let supervisor = TaskSupervisor::new(cancel.clone());
1621
1622 let sink = CompactionAuditSink::open(&path, 64, Some(&supervisor))
1623 .await
1624 .unwrap();
1625 sink.flush().await;
1626
1627 tokio::time::sleep(std::time::Duration::from_millis(10)).await;
1629
1630 let names: Vec<String> = supervisor
1631 .snapshot()
1632 .into_iter()
1633 .map(|s| s.name.to_string())
1634 .collect();
1635 assert!(
1636 names.iter().any(|n| n == "context.audit_sink"),
1637 "supervisor must have a task named 'context.audit_sink', got: {names:?}"
1638 );
1639
1640 cancel.cancel();
1641 }
1642
1643 #[test]
1646 fn classify_with_role_non_ascii_body_does_not_panic() {
1647 let cjk = "你好世界".repeat(20); let emoji = "🦀".repeat(30); let mixed = "abc🦀中文".repeat(15);
1652
1653 let _ = classify_with_role(&cjk, false);
1655 let _ = classify_with_role(&emoji, false);
1656 let _ = classify_with_role(&mixed, false);
1657 }
1658}