Skip to main content

zeph_context/
typed_page.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! Typed page classification and minimum-fidelity invariants for context compaction.
5//!
6//! Every context segment entering the assembler is tagged with a [`PageType`] and
7//! wrapped in a [`TypedPage`]. The [`PageInvariant`] trait declares the fidelity
8//! contract enforced at every compaction boundary.
9//!
10//! Classification is deterministic and side-effect free — no I/O, no LLM calls.
11//!
12//! # Architecture
13//!
14//! This module lives in `zeph-context` to keep classification logic co-located with
15//! the assembler. No dependency on `zeph-memory` is introduced here.
16//!
17//! # Feature flag
18//!
19//! All typed-page functionality is gated behind the
20//! `[memory.compaction.typed_pages] enabled = true` config key. When disabled the
21//! assembler falls back to the legacy untyped path without behaviour change.
22
23use 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// ── PageType ──────────────────────────────────────────────────────────────────
31
32/// Classification of a context segment for compaction purposes.
33///
34/// The variant determines which [`PageInvariant`] is enforced and what shape the
35/// compacted summary must have.
36///
37/// # Invariant
38///
39/// Every [`TypedPage`] carries exactly one `PageType`. Unclassifiable segments
40/// default to [`PageType::ConversationTurn`] (see [`classify`]).
41#[non_exhaustive]
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
43#[serde(rename_all = "snake_case")]
44pub enum PageType {
45    /// A tool request/response pair sourced from memory or the current turn.
46    ToolOutput,
47    /// A user or assistant message that does not carry a tool role.
48    ConversationTurn,
49    /// Cross-session context, past summaries, or graph-fact recall injections.
50    MemoryExcerpt,
51    /// Session digest, persona, skill instructions, or compression guidelines.
52    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// ── PageOrigin ────────────────────────────────────────────────────────────────
67
68/// Provenance of a [`TypedPage`], serialised into audit records.
69#[derive(Debug, Clone, Serialize, Deserialize)]
70#[serde(tag = "kind", rename_all = "snake_case")]
71#[non_exhaustive]
72pub enum PageOrigin {
73    /// Tool request/response pair.
74    ToolPair {
75        /// Name of the tool that produced this output.
76        tool_name: String,
77    },
78    /// User or assistant conversation turn.
79    Turn {
80        /// Opaque message identifier (numeric message id as string).
81        message_id: String,
82    },
83    /// Injected from memory (cross-session, summary, graph-facts, etc.).
84    Excerpt {
85        /// Human-readable label identifying the memory source.
86        source_label: String,
87    },
88    /// Session-level system context (persona, skills, digest).
89    System {
90        /// Logical key for this system context block (e.g. `"persona"`, `"skills"`).
91        key: String,
92    },
93}
94
95// ── SchemaHint ────────────────────────────────────────────────────────────────
96
97/// Body format hint for [`PageType::ToolOutput`] pages.
98///
99/// Used by the invariant to select the correct structured-summary prompt.
100#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
101#[serde(rename_all = "snake_case")]
102#[non_exhaustive]
103pub enum SchemaHint {
104    /// Body is valid JSON (object or array).
105    Json,
106    /// Body is UTF-8 text (log lines, prose, etc.).
107    Text,
108    /// Body is a unified diff.
109    Diff,
110    /// Body is a tab- or comma-separated table.
111    Table,
112    /// Body is non-UTF-8 binary data.
113    Binary,
114}
115
116// ── PageId ────────────────────────────────────────────────────────────────────
117
118/// Stable content-addressed identifier for a [`TypedPage`].
119///
120/// Computed as BLAKE3 over `page_type_tag || origin_tag || body_bytes`, encoded
121/// as lowercase hex (first 16 bytes = 32 hex chars). The same input always
122/// produces the same `PageId`, enabling deduplication across turns.
123///
124/// # Semantics
125///
126/// `PageId` is a **content hash**: identical source bytes (same page type, same
127/// origin key, same body) always produce the same id. This means that the same
128/// tool output appearing in two different turns produces the same `PageId`.
129/// Callers that need per-turn provenance must use `turn_id` from the audit record
130/// — `PageId` is for deduplication, not for uniqueness across turns.
131#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
132pub struct PageId(pub String);
133
134impl PageId {
135    /// Compute a [`PageId`] from the page type, origin key, and body bytes.
136    #[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        // Use first 16 bytes (128 bits) — sufficient for deduplication purposes.
146        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// ── TypedPage ─────────────────────────────────────────────────────────────────
156
157/// A classified context segment ready for invariant-aware compaction.
158///
159/// `TypedPage` is the unit of work passed to compaction boundaries. The
160/// [`PageId`] is content-stable: the same source bytes always produce the same
161/// id, enabling the compactor to skip already-compacted pages.
162#[derive(Debug, Clone)]
163pub struct TypedPage {
164    /// Stable content-addressed identifier.
165    pub page_id: PageId,
166    /// Classification determining which invariant applies.
167    pub page_type: PageType,
168    /// Provenance of this page (for audit records).
169    pub origin: PageOrigin,
170    /// Token count of the original body.
171    pub tokens: u32,
172    /// Body text shared across potential clones.
173    pub body: Arc<str>,
174    /// Body format hint (populated for `ToolOutput` only; `None` otherwise).
175    pub schema_hint: Option<SchemaHint>,
176}
177
178impl TypedPage {
179    /// Construct a new [`TypedPage`], computing its [`PageId`] from content.
180    #[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// ── FidelityContract ──────────────────────────────────────────────────────────
211
212/// The set of fields that must be present in a compacted page.
213///
214/// Returned by [`PageInvariant::minimum_fidelity`] and checked by
215/// [`PageInvariant::verify`] after summarization.
216#[derive(Debug, Clone)]
217pub struct FidelityContract {
218    /// Human-readable label for this contract version (e.g. `"structured_summary_v1"`).
219    pub fidelity_level: &'static str,
220    /// Schema version integer included in audit records.
221    pub invariant_version: u8,
222    /// Fields that must appear in the compacted body text.
223    pub required_fields: &'static [&'static str],
224}
225
226// ── FidelityViolation ─────────────────────────────────────────────────────────
227
228/// Describes why an invariant check failed after compaction.
229///
230/// A violation is a hard error: the compacted page is dropped and an audit
231/// record with `violations` is emitted.
232#[derive(Debug, Clone, Serialize)]
233pub struct FidelityViolation {
234    /// The field or property that was expected but missing.
235    pub missing_field: String,
236    /// Human-readable explanation of the violation.
237    pub detail: String,
238}
239
240// ── CompactedPage ─────────────────────────────────────────────────────────────
241
242/// The output of a compaction attempt, passed to [`PageInvariant::verify`].
243#[derive(Debug, Clone)]
244pub struct CompactedPage {
245    /// The summarized body text produced by the compaction provider.
246    pub body: Arc<str>,
247    /// Token count of the compacted body.
248    pub tokens: u32,
249}
250
251// ── PageInvariant trait ───────────────────────────────────────────────────────
252
253/// Minimum-fidelity contract for a single [`PageType`].
254///
255/// Implementors declare what a compacted page must contain ([`minimum_fidelity`])
256/// and verify that the actual output honours the contract ([`verify`]).
257///
258/// The trait is object-safe so implementations can be stored in a
259/// `HashMap<PageType, Box<dyn PageInvariant>>` registry.
260///
261/// # Contract
262///
263/// - [`verify`] MUST NOT perform I/O or call an LLM.
264/// - A failed [`verify`] means the compacted page is dropped — it is NEVER
265///   injected in degraded form.
266///
267/// [`minimum_fidelity`]: PageInvariant::minimum_fidelity
268/// [`verify`]: PageInvariant::verify
269pub trait PageInvariant: Send + Sync {
270    /// The page type this invariant governs.
271    fn page_type(&self) -> PageType;
272
273    /// Return the fidelity contract required for a given page.
274    fn minimum_fidelity(&self, page: &TypedPage) -> FidelityContract;
275
276    /// Verify that `compacted` satisfies the fidelity contract derived from `original`.
277    ///
278    /// # Errors
279    ///
280    /// Returns a non-empty [`Vec<FidelityViolation>`] when one or more required
281    /// fields are absent from the compacted body.
282    fn verify(
283        &self,
284        original: &TypedPage,
285        compacted: &CompactedPage,
286    ) -> Result<(), Vec<FidelityViolation>>;
287}
288
289// ── Per-type invariant implementations ───────────────────────────────────────
290
291/// Invariant for [`PageType::ToolOutput`] pages.
292///
293/// The compacted body must contain the tool name, an exit/status indicator,
294/// and at least one structural key from the original output.
295pub 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        // For binary pages the body marker is injected by the compactor, skip field checks.
317        if original.schema_hint == Some(SchemaHint::Binary) {
318            return Ok(());
319        }
320
321        let mut violations = Vec::new();
322
323        // The compacted body must reference the tool name.
324        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        // The compacted body must contain at least one exit / status indicator.
336        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        // For JSON-schema tool outputs, verify that at least one top-level JSON
350        // field name from the original body is present in the compacted body
351        // (FR-003: structural keys must be preserved, not just exit-status markers).
352        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
373/// Check that at least one top-level JSON key from `original` appears in `compacted`.
374///
375/// Parses `original` as a JSON object and returns `true` when any top-level key
376/// string is a substring of `compacted`. Returns `true` (no violation) when
377/// `original` is not a valid JSON object — the caller already checked schema hint.
378fn 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
391/// Invariant for [`PageType::ConversationTurn`] pages.
392///
393/// The compacted body must preserve a role indicator and some meaningful content.
394pub 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
427/// Invariant for [`PageType::MemoryExcerpt`] pages.
428///
429/// The compacted body must retain the source label and a message id reference.
430pub 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
464/// Invariant for [`PageType::SystemContext`] pages.
465///
466/// System context MUST NOT be paraphrased. Compaction replaces it with a
467/// pointer record; any body other than the pointer prefix is a violation.
468pub struct SystemContextInvariant;
469
470/// Pointer prefix that the compactor writes for `SystemContext` pages.
471pub 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
504// ── InvariantRegistry ─────────────────────────────────────────────────────────
505
506/// Registry mapping each [`PageType`] to its [`PageInvariant`] implementation.
507///
508/// Built once and shared via `Arc` so tests can swap in a mock registry.
509///
510/// # Examples
511///
512/// ```
513/// use zeph_context::typed_page::{InvariantRegistry, PageType};
514///
515/// let reg = InvariantRegistry::default();
516/// let inv = reg.get(PageType::ToolOutput).unwrap();
517/// assert_eq!(inv.page_type(), PageType::ToolOutput);
518/// ```
519pub 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    /// Look up the invariant for a given [`PageType`].
539    ///
540    /// Always returns `Some` for the four built-in variants.
541    #[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    /// Verify that `compacted` satisfies the invariant for `original` at a compaction boundary.
552    ///
553    /// This is the primary entry point for the compactor — it wraps `verify()` in a
554    /// `tracing::info_span!` per NFR-009 so every compaction boundary is observable.
555    ///
556    /// Returns `Ok(())` when the invariant is satisfied, or the violation list on failure.
557    ///
558    /// # Errors
559    ///
560    /// Propagates [`FidelityViolation`]s from the registered invariant implementation.
561    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// ── Classification helpers ────────────────────────────────────────────────────
588
589/// Classify a context segment by examining well-known prefix markers.
590///
591/// Classification is deterministic and performs no I/O. When the input does not
592/// match any known prefix the function defaults to [`PageType::ConversationTurn`]
593/// and logs at `WARN` level per FR-008.
594///
595/// The function emits a `context.compaction.typed_page.classify` span per NFR-009
596/// so every classification is observable in traces.
597///
598/// | Source marker | Assigned [`PageType`] |
599/// |---|---|
600/// | Starts with `[tool_output]` or `[tool:` | [`PageType::ToolOutput`] |
601/// | Starts with `[cross-session context]`, `[semantic recall]`, `[known facts]`, `[conversation summaries]` | [`PageType::MemoryExcerpt`] |
602/// | Starts with `[Persona context]`, `[Past experience]`, `[Memory summary]`, `[system` | [`PageType::SystemContext`] |
603/// | Everything else | [`PageType::ConversationTurn`] |
604///
605/// # Examples
606///
607/// ```
608/// use zeph_context::typed_page::{classify, PageType};
609///
610/// assert_eq!(classify("[tool_output] exit_code: 0"), PageType::ToolOutput);
611/// assert_eq!(classify("[cross-session context]\nsome recall"), PageType::MemoryExcerpt);
612/// assert_eq!(classify("[Persona context]\nfact"), PageType::SystemContext);
613/// assert_eq!(classify("Hello, world!"), PageType::ConversationTurn);
614/// ```
615#[must_use]
616pub fn classify(body: &str) -> PageType {
617    classify_with_role(body, false)
618}
619
620/// Classify a context segment, with an explicit `is_system_role` hint.
621///
622/// When `is_system_role` is `true` the segment is classified as
623/// [`PageType::SystemContext`] without prefix matching, preventing arbitrary
624/// system messages injected by the assembler from silently falling back to
625/// `ConversationTurn` (Key Invariant: "`SystemContext` pages are never paraphrased").
626///
627/// Use this variant when the caller has access to the message `Role`.
628///
629/// # Examples
630///
631/// ```
632/// use zeph_context::typed_page::{classify_with_role, PageType};
633///
634/// // A plain system message without a known prefix is still SystemContext.
635/// assert_eq!(classify_with_role("You are a helpful assistant.", true), PageType::SystemContext);
636/// // Role hint does not override ToolOutput prefix detection.
637/// assert_eq!(classify_with_role("[tool_output] exit_code: 0", false), PageType::ToolOutput);
638/// ```
639#[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    // Use the same prefix constants as the assembler for consistency.
650    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    // When the caller signals Role::System, classify as SystemContext even if
689    // the body does not start with a known prefix.  This prevents system
690    // context injected by the assembler (e.g. plain instructions, directives)
691    // from being eligible for paraphrase.
692    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/// Detect [`SchemaHint`] for a [`PageType::ToolOutput`] body.
704///
705/// Returns [`SchemaHint::Binary`] when the body is not valid UTF-8 (detected via
706/// presence of replacement characters) or when the caller passes `is_binary =
707/// true`. JSON detection is heuristic (starts with `{` or `[`).
708#[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    // Simple table heuristic: first line contains multiple tab or pipe separators.
725    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// ── Audit record ──────────────────────────────────────────────────────────────
733
734/// One JSONL audit record emitted per compacted page (FR-007).
735///
736/// Written to `[memory.compaction.typed_pages] audit_path` by the audit sink
737/// before the compacted context is handed to the LLM.
738#[derive(Debug, Serialize)]
739pub struct CompactedPageRecord {
740    /// ISO-8601 timestamp when the compaction occurred.
741    pub ts: String,
742    /// Opaque turn identifier (agent turn counter as string).
743    pub turn_id: String,
744    /// Stable content-addressed page identifier.
745    pub page_id: String,
746    /// Page classification.
747    pub page_type: PageType,
748    /// Serialised page origin.
749    pub origin: PageOrigin,
750    /// Token count of the original page.
751    pub original_tokens: u32,
752    /// Token count of the compacted page.
753    pub compacted_tokens: u32,
754    /// Fidelity level label from the invariant contract.
755    pub fidelity_level: String,
756    /// Schema version integer.
757    pub invariant_version: u8,
758    /// Provider name used for summarization.
759    pub provider_name: String,
760    /// Fidelity violations encountered (empty on success).
761    pub violations: Vec<FidelityViolation>,
762    /// `true` when classification fell back to `ConversationTurn`.
763    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
764    pub classification_fallback: bool,
765}
766
767// ── Batch assertions ──────────────────────────────────────────────────────────
768
769/// A failed batch-level compaction assertion.
770#[derive(Debug, Clone, Serialize)]
771pub struct BatchViolation {
772    /// Short label for the assertion that failed.
773    pub assertion: String,
774    /// Human-readable explanation.
775    pub detail: String,
776}
777
778/// Batch-level compaction assertions for typed-page enforcement.
779///
780/// Unlike per-page [`PageInvariant`] which checks one page against its compacted form,
781/// batch assertions verify aggregate properties of the entire summary against the set
782/// of classified pages that were sent to the LLM.
783///
784/// Violations are observational — they never block compaction. They are logged and
785/// emitted to the audit trail.
786///
787/// # Examples
788///
789/// ```
790/// use zeph_context::typed_page::BatchAssertions;
791///
792/// let assertions = BatchAssertions {
793///     tool_names_in_batch: vec!["shell".to_string()],
794///     has_memory_excerpt: false,
795///     excerpt_labels: vec![],
796/// };
797/// // Summary that mentions the tool — all assertions pass.
798/// let violations = assertions.check("shell ran and exited 0");
799/// assert!(violations.is_empty());
800/// ```
801#[derive(Debug, Clone, Default)]
802pub struct BatchAssertions {
803    /// Tool names collected from `ToolOutput` pages in the batch.
804    pub tool_names_in_batch: Vec<String>,
805    /// Whether any `MemoryExcerpt` pages were in the batch.
806    pub has_memory_excerpt: bool,
807    /// Source labels from `MemoryExcerpt` pages.
808    pub excerpt_labels: Vec<String>,
809}
810
811impl BatchAssertions {
812    /// Check the summary against batch-level assertions.
813    ///
814    /// Returns a list of assertion failures (empty = all pass). Failures are never fatal.
815    #[must_use]
816    pub fn check(&self, summary: &str) -> Vec<BatchViolation> {
817        let mut violations = Vec::new();
818
819        // At least one tool name from the batch must appear in the summary.
820        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 memory excerpts were present, at least one source label should appear.
838        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
858// ── TypedPagesState ───────────────────────────────────────────────────────────
859
860/// Shared runtime state for typed-page compaction, created once at agent startup.
861///
862/// Bundles the invariant registry and optional audit sink so they can be shared
863/// via `Arc` across compaction calls without per-call allocation.
864pub struct TypedPagesState {
865    /// Invariant registry shared across all compaction calls.
866    pub registry: InvariantRegistry,
867    /// Optional JSONL audit sink. `None` when audit is disabled.
868    pub audit_sink: Option<CompactionAuditSink>,
869    /// Whether enforcement is `Active` (pointer-replace `SystemContext` + batch assertions).
870    /// `false` = `Observe` mode (classify and audit only, no behavioral change).
871    pub is_active: bool,
872}
873
874// ── Audit command ─────────────────────────────────────────────────────────────
875
876/// Internal command sent through the audit sink channel.
877enum AuditCommand {
878    /// Write a compaction record.
879    Record(CompactedPageRecord),
880    /// Flush all preceding records and signal completion via the oneshot.
881    Flush(tokio::sync::oneshot::Sender<()>),
882}
883
884// ── Audit sink ────────────────────────────────────────────────────────────────
885
886/// Async bounded-mpsc audit sink for compaction records.
887///
888/// The sink serialises [`CompactedPageRecord`] values to a JSONL file via a
889/// background writer task, mirroring the `zeph-tools` audit pattern. Dropped
890/// records (when the channel is full) are counted and logged.
891///
892/// # Invariant
893///
894/// [`CompactionAuditSink::flush`] sends a rendezvous sentinel through the channel
895/// and awaits the writer task's confirmation with a 100 ms timeout. Records accepted
896/// into the channel before `flush` is called are guaranteed to be written before the
897/// flush responder fires.
898///
899/// # Examples
900///
901/// ```no_run
902/// use zeph_context::typed_page::CompactionAuditSink;
903/// use std::path::Path;
904///
905/// # async fn example() {
906/// let sink = CompactionAuditSink::open(Path::new(".local/audit/compaction.jsonl"), 256, None)
907///     .await
908///     .unwrap();
909/// # }
910/// ```
911#[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    /// Open a new audit sink writing to `path`.
919    ///
920    /// `capacity` is the bounded channel depth; records dropped when full are counted
921    /// in the internal drop counter and logged at WARN.
922    ///
923    /// When `supervisor` is `Some`, the background writer task is registered as
924    /// `"context.audit_sink"` for lifecycle management. When `None`, the task is
925    /// spawned directly (test environments only).
926    ///
927    /// # Errors
928    ///
929    /// Returns an error when `path` cannot be opened for appending.
930    #[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            // Flush remaining bytes when channel closes.
973            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); // EXEMPT: supervisor=None fallback (test environments only)
985        }
986
987        Ok(Self { tx, drop_counter })
988    }
989
990    /// Send a record to the audit sink.
991    ///
992    /// If the channel is full the record is dropped and the drop counter is incremented.
993    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    /// Flush all pending records with bounded 100 ms timeout.
1012    ///
1013    /// Sends a `Flush` sentinel through the same channel as records, so ordering is
1014    /// preserved — the writer task responds only after all preceding records are written.
1015    /// If the writer task does not respond within 100 ms, the flush times out silently.
1016    #[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    /// Number of records dropped due to a full channel.
1025    #[must_use]
1026    pub fn dropped_count(&self) -> u64 {
1027        self.drop_counter.load(std::sync::atomic::Ordering::Relaxed)
1028    }
1029}
1030
1031// ── Tests ─────────────────────────────────────────────────────────────────────
1032
1033#[cfg(test)]
1034mod tests {
1035    use super::*;
1036
1037    // ── PageId ────────────────────────────────────────────────────────────────
1038
1039    #[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    // ── classify ──────────────────────────────────────────────────────────────
1060
1061    #[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    // ── detect_schema_hint ────────────────────────────────────────────────────
1100
1101    #[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    // ── ToolOutputInvariant ───────────────────────────────────────────────────
1126
1127    #[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    // ── SystemContextInvariant ────────────────────────────────────────────────
1188
1189    #[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    // ── InvariantRegistry ─────────────────────────────────────────────────────
1231
1232    #[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    // ── InvariantRegistry::enforce ────────────────────────────────────────────
1259
1260    #[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    // ── MemoryExcerptInvariant ────────────────────────────────────────────────
1325
1326    #[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    // ── ConversationTurnInvariant ─────────────────────────────────────────────
1392
1393    #[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    // ── CompactionAuditSink ───────────────────────────────────────────────────
1441
1442    #[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 the sink to close the channel and let the writer task flush.
1467        drop(sink);
1468        // Give the writer task time to finish.
1469        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        // Capacity 1: first send fills the channel, subsequent sends are dropped.
1485        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        // Send enough records to guarantee overflow.
1505        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        // flush on an empty sink must not panic or deadlock.
1521        sink.flush().await;
1522    }
1523
1524    // ── classify_with_role ────────────────────────────────────────────────────
1525
1526    #[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    // ── check_json_structural_key (via ToolOutputInvariant) ───────────────────
1551
1552    #[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        // Compacted body references "exit_code" and "shell".
1566        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        // Compacted body references tool name and status but none of the JSON keys.
1587        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    // ── Regression: F1 — capacity=0 must not panic ────────────────────────────
1602
1603    #[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        // capacity=0 used to panic in tokio::sync::mpsc::channel(0); must clamp to 1.
1608        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        // Give the supervisor time to register the task before checking.
1628        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    // ── Regression: F3 — non-ASCII body must not panic on prefix slice ────────
1644
1645    #[test]
1646    fn classify_with_role_non_ascii_body_does_not_panic() {
1647        // CJK and emoji span multiple bytes; a naive &body[..80] would panic at a
1648        // mid-character byte boundary. classify_with_role must not panic for any input.
1649        let cjk = "你好世界".repeat(20); // 80+ bytes, 4 bytes each
1650        let emoji = "🦀".repeat(30); // 120+ bytes, 4 bytes each
1651        let mixed = "abc🦀中文".repeat(15);
1652
1653        // None of these must panic:
1654        let _ = classify_with_role(&cjk, false);
1655        let _ = classify_with_role(&emoji, false);
1656        let _ = classify_with_role(&mixed, false);
1657    }
1658}