Skip to main content

velesdb_memory/
context.rs

1//! The deterministic context compiler (EPIC-P-070).
2//!
3//! Classifies, deduplicates, and packs caller-supplied context fragments
4//! under a token budget — **no LLM, no network, no clock**: the pipeline is a
5//! sequence of pure stages (`chunk → classify → dedup → score → pack →
6//! assemble`), so the same [`CompileRequest`](crate::context::CompileRequest)
7//! always produces the same
8//! [`CompiledContext`](crate::context::CompiledContext), byte for byte.
9//!
10//! Invariants:
11//! - **Budget**: the assembled content never exceeds the request's token
12//!   budget — packing accounts per-piece estimates plus joiner costs *priced
13//!   by the injected estimator*, which bounds the whole-text estimate for a
14//!   superadditive estimator (the default rounds every piece up).
15//! - **Provenance**: every input fragment gets exactly one
16//!   [`ContextDecision`](crate::context::ContextDecision) with a stable rule
17//!   id and a content hash; every fragment stays addressable via a
18//!   **content-addressed** `ctx://source/<hash>` handle (immune to
19//!   caller-id collisions) — hashed over the text for a text fragment, over
20//!   the raw decoded media bytes for a media fragment (see
21//!   `Analysis::handle_hash`: captions are typically blank, so a
22//!   caption-keyed handle would collide every captionless image).
23//! - **Nothing critical is silently lost**: content that cannot fit becomes
24//!   a [`RetrievalHandle`](crate::context::RetrievalHandle); losing
25//!   preserve-classified content raises
26//!   [`CompiledContext::risk`](crate::context::CompiledContext::risk) to
27//!   [`FidelityRisk::High`](crate::context::FidelityRisk::High); a critical
28//!   fragment is never sacrificed to near-deduplication, and a duplicate of
29//!   a twin that did not emit verbatim keeps its own handle and risk.
30//!
31//! Memory-backed fragment selection, persisted working contexts, and
32//! compilation events layer on top in the `persistence`-gated bridge
33//! (US-002); MCP and Node expose the same types unchanged (US-003).
34
35mod budget;
36pub mod chunk;
37mod classify;
38mod dedup;
39pub mod estimator;
40/// Adapter-side I/O pre-pass for `path`-referenced context fragments
41/// (V2b-1): resolves `ContextFragment::path` into `content` under a strict,
42/// short-circuiting security pipeline, BEFORE the request reaches the pure
43/// compiler core. Not compiled for `wasm32` — there is no local filesystem
44/// to read from a WASM host; see [`crate::error::MemoryError::IngestDisabled`]
45/// for what a `path` fragment does there instead.
46#[cfg(not(target_arch = "wasm32"))]
47pub mod ingest;
48pub mod insights;
49mod log_normalize;
50// `pub(crate)`, not private: the memory bridge (`service::memory_bridge`,
51// physically stored under `context/` but logically a sibling module of
52// `context`, see `service.rs`) decodes media bytes itself (US-009, PR2) to
53// derive a deterministic placeholder embedding for a stored media source.
54pub(crate) mod media;
55pub mod model;
56/// The `suggest_budget` MCP tool's static model→window table (V2a-3 quick
57/// win). No dependency on anything else in the pipeline — a pure lookup.
58pub mod model_windows;
59pub(crate) mod provenance;
60mod relevance;
61/// Deterministic transcript segmentation for the `compile_transcript` MCP
62/// tool (V2b-2): splits a raw agent-session transcript into turns and, within
63/// each turn, into code/log/body sub-segments — pure, zero-regex, zero-clock,
64/// so the same transcript + policy always segments byte-identically. See
65/// [`segment::segment_transcript`].
66pub mod segment;
67/// The binding-side glue over [`segment`]: one implementation of
68/// "transcript in, `CompileRequest` + audit trail out", relayed by the
69/// Node, Python and WASM bindings instead of copied into each.
70pub mod transcript_bridge;
71/// The id wire contract (decimal-string `u64`) shared by every JS-facing
72/// binding of these types — one source of truth for [`wire::ID_KEYS`]
73/// instead of a Node/WASM copy each.
74pub mod wire;
75
76pub use chunk::{chunk_text, ChunkBoundary, ChunkPolicy, TextChunk};
77pub use estimator::{DynTokenEstimator, HeuristicEstimator, TokenEstimator};
78#[cfg(not(target_arch = "wasm32"))]
79pub use ingest::IngestRoots;
80pub use insights::{CompilationInsights, ModelPricing, PricingTable};
81pub use model::{
82    CompilePolicy, CompileRequest, CompiledContext, CompiledSection, ContextAction,
83    ContextDecision, ContextDecisionRef, ContextFact, ContextFragment, ContextSavings,
84    ContextSource, ContextWarning, FidelityRisk, ImportanceWeights, LoadedWorkingContext, MediaRef,
85    MemoryScope, RetrievalHandle, SectionKind, SourceReference, WorkingContext,
86    WorkingContextIndex, WorkingContextSession,
87};
88pub use model_windows::{model_window, suggest_token_budget, SuggestedBudget};
89pub use relevance::DeterministicReranker;
90pub use segment::{
91    segment_transcript, SegmentFormat, SegmentKind, SegmentationOutcome, SegmentationPolicy,
92    TranscriptSegment,
93};
94pub use transcript_bridge::{
95    build_transcript_compile_request, SegmentInfo, SegmentationReport, TranscriptCompileInput,
96};
97
98use std::collections::BTreeMap;
99
100use crate::error::MemoryError;
101use crate::id::stable_id;
102use crate::limits;
103
104use budget::PackItem;
105use classify::RuleMatch;
106use dedup::{DupKind, Duplicate};
107
108/// The stable, content-addressed id of a fragment whose caller supplied none
109/// — the crate's one id scheme (FNV-1a 64), also used as every decision's
110/// content hash and as the tail of every `ctx://source/<hash>` handle.
111#[must_use]
112pub fn fragment_id(content: &str) -> u64 {
113    stable_id(content)
114}
115
116/// The deterministic context compiler. Build one with a policy, optionally
117/// inject an estimator and a pricing table, then [`compile`](Self::compile).
118///
119/// ```rust
120/// use velesdb_memory::context::{
121///     CompilePolicy, CompileRequest, ContextCompiler, ContextFragment,
122/// };
123///
124/// let compiler = ContextCompiler::new(CompilePolicy::default());
125/// let out = compiler
126///     .compile(&CompileRequest {
127///         query: "deploy status".to_owned(),
128///         fragments: vec![ContextFragment {
129///             id: None,
130///             content: "The deploy pipeline is green.".to_owned(),
131///             path: None,
132///             kind: None,
133///             priority: None,
134///             metadata: None,
135///             media: None,
136///         }],
137///         project: None,
138///         target_model: None,
139///         token_budget: 1_000,
140///         memory_scope: None,
141///         policy: None,
142///     })
143///     .expect("a generous budget compiles");
144/// assert!(out.content.contains("deploy pipeline"));
145/// ```
146pub struct ContextCompiler {
147    policy: CompilePolicy,
148    estimator: DynTokenEstimator,
149    pricing: Option<PricingTable>,
150}
151
152impl ContextCompiler {
153    /// A compiler over `policy`, with the default char-ratio estimator and
154    /// no pricing (insights then report tokens only).
155    #[must_use]
156    pub fn new(policy: CompilePolicy) -> Self {
157        Self {
158            policy,
159            estimator: Box::new(HeuristicEstimator),
160            pricing: None,
161        }
162    }
163
164    /// Replace the token estimator (e.g. a model-exact tokenizer).
165    #[must_use]
166    pub fn with_estimator(mut self, estimator: DynTokenEstimator) -> Self {
167        self.estimator = estimator;
168        self
169    }
170
171    /// Inject a versioned pricing table so insights also report estimated
172    /// cost savings for the request's target model.
173    #[must_use]
174    pub fn with_pricing(mut self, pricing: PricingTable) -> Self {
175        self.pricing = Some(pricing);
176        self
177    }
178
179    /// The policy this compilation actually runs under: the request's
180    /// override when present, this compiler's otherwise. The memory bridge
181    /// reads it to honor the storage/event opt-outs.
182    pub(crate) fn effective_policy<'a>(&'a self, request: &'a CompileRequest) -> &'a CompilePolicy {
183        request.policy.as_ref().unwrap_or(&self.policy)
184    }
185
186    /// Compile `request` into a budgeted, fully-audited context.
187    ///
188    /// # Errors
189    ///
190    /// [`MemoryError::ContextOverLimit`] when the request exceeds a
191    /// [`crate::limits`] cap (fragment count or single-fragment size),
192    /// [`MemoryError::MetadataTooLarge`] when a fragment's `metadata` exceeds
193    /// [`crate::limits::MAX_METADATA_BYTES`], and
194    /// [`MemoryError::ContextBudget`] when the token budget minus the
195    /// policy's response reserve leaves no room for any context.
196    pub fn compile(&self, request: &CompileRequest) -> Result<CompiledContext, MemoryError> {
197        let compiled = self.compile_raw(request)?;
198        Ok(apply_slim(compiled, self.effective_policy(request)))
199    }
200
201    /// [`Self::compile`] without the [`CompilePolicy::slim_response`]
202    /// post-processing: the memory bridge needs the FULL `decisions` to
203    /// annotate memory provenance and recompute `warnings` (both can change
204    /// a pulled fragment's `relevance`/`reason` after this returns) before
205    /// slimming happens — every other caller should use [`Self::compile`].
206    pub(crate) fn compile_raw(
207        &self,
208        request: &CompileRequest,
209    ) -> Result<CompiledContext, MemoryError> {
210        let policy = self.effective_policy(request);
211        let usable = validate(request, policy)?;
212        let analyses = analyze(request, policy, self.estimator.as_ref());
213        let items = pack_items(&analyses, policy, usable, self.estimator.as_ref());
214        let taken = budget::pack(&items, usable, &self.estimator);
215        let emissions = emissions(&items, &taken);
216        Ok(self.finish(request, &analyses, &emissions))
217    }
218
219    /// Assemble the output, decisions, insights, and risk.
220    fn finish(
221        &self,
222        request: &CompileRequest,
223        analyses: &[Analysis],
224        emissions: &BTreeMap<usize, Emission>,
225    ) -> CompiledContext {
226        let sections = sections(analyses, emissions);
227        let content = sections
228            .iter()
229            .map(|section| section.content.as_str())
230            .collect::<Vec<_>>()
231            .join(budget::JOINER);
232        let decisions: Vec<ContextDecision> = analyses
233            .iter()
234            .map(|analysis| decision(analysis, analyses, emissions))
235            .collect();
236        let insights = self.insights(request, analyses, &decisions, emissions, &content);
237        let warnings = warnings_for(&decisions);
238        CompiledContext {
239            retrieval_handles: retrieval_handles(analyses, &decisions),
240            sources: analyses
241                .iter()
242                .filter(|analysis| analysis.dup.is_none())
243                .map(|analysis| {
244                    provenance::source_for(analysis.fragment_id, analysis.handle_hash())
245                })
246                .collect(),
247            risk: decisions
248                .iter()
249                .map(|decision| decision.risk)
250                .max()
251                .unwrap_or_default(),
252            content,
253            sections,
254            decisions,
255            insights,
256            warnings,
257        }
258    }
259
260    /// Token accounting, with cost figures only when pricing knows the model.
261    fn insights(
262        &self,
263        request: &CompileRequest,
264        analyses: &[Analysis],
265        decisions: &[ContextDecision],
266        emissions: &BTreeMap<usize, Emission>,
267        content: &str,
268    ) -> CompilationInsights {
269        let estimator = self.estimator.as_ref();
270        let tokens_in: u64 = analyses
271            .iter()
272            .map(|analysis| analysis.tokens)
273            .fold(0, u64::saturating_add);
274        // `content` already carries every emitted fragment's TEXT — for a
275        // media fragment (US-009, PR1) that means its caption only, since
276        // raw media bytes are never turned into packed text (see `pieces`).
277        // The image's own cost has to be added on top, but only for media
278        // that actually made it into the output (an emissions entry exists)
279        // — an externalized or superseded image contributed nothing and
280        // must not appear here either (see `pack_items`).
281        let media_tokens_out: u64 = analyses
282            .iter()
283            .filter(|analysis| emissions.contains_key(&analysis.seq))
284            .filter_map(|analysis| analysis.media.as_ref())
285            .map(|media| media.image_tokens)
286            .fold(0, u64::saturating_add);
287        let tokens_out = estimator.estimate(content).saturating_add(media_tokens_out);
288        let tokens_saved = tokens_in.saturating_sub(tokens_out);
289        let mut insights = CompilationInsights {
290            tokens_in,
291            tokens_out,
292            tokens_saved,
293            tokens_saved_by_rule: saved_by_rule(analyses, decisions, emissions, estimator),
294            ..CompilationInsights::default()
295        };
296        let cost = request.target_model.as_deref().and_then(|model| {
297            // The request's own table wins (it is the only channel wire
298            // callers — MCP, Node — have); the builder-injected one is the
299            // Rust-embedder fallback.
300            let pricing = self
301                .effective_policy(request)
302                .pricing
303                .as_ref()
304                .or(self.pricing.as_ref())?;
305            let micros = pricing.cost_micros(model, tokens_saved)?;
306            Some((micros, pricing.currency.clone(), pricing.version.clone()))
307        });
308        if let Some((micros, currency, version)) = cost {
309            insights.estimated_cost_saved_micros = Some(micros);
310            insights.currency = Some(currency);
311            insights.pricing_version = Some(version);
312        }
313        insights
314    }
315}
316
317/// Everything the pipeline derived about one input fragment. Borrows the
318/// request (the pipeline never mutates fragments), so a compile at the size
319/// caps does not double the corpus in memory.
320struct Analysis<'a> {
321    /// Input position.
322    seq: usize,
323    /// Caller id, or the content-derived stable id.
324    fragment_id: u64,
325    /// FNV-1a hash of the original content (computed once, reused by ids,
326    /// dedup, and handles).
327    content_hash: u64,
328    /// The original text, borrowed from the request.
329    original: &'a str,
330    /// Estimated tokens of the original (computed once, reused by insights,
331    /// handles, and savings attribution).
332    tokens: u64,
333    /// Classification outcome.
334    rule: RuleMatch,
335    /// Lexical relevance to the query.
336    relevance: f32,
337    /// Caller priority (default 0).
338    priority: u8,
339    /// Set when this fragment duplicates an earlier one it may safely be
340    /// dropped for (see [`retain_safe_duplicates`]).
341    dup: Option<Duplicate>,
342    /// Only set for `abstract.log_dedup`-classified fragments: the
343    /// collapsed single piece, and whether
344    /// [`CompilePolicy::normalize_log_timestamps`] actually changed the
345    /// grouping (ventilated into the decision `reason`). Computed once here
346    /// so [`pieces`] and [`decision`] never redo the line-scan.
347    abstract_collapse: Option<(String, bool)>,
348    /// Set when the fragment carries inline media (US-009, PR1): its
349    /// dedup identity and precomputed image token cost, computed once here
350    /// (decoding is not free) and reused by dedup, packing, and insights.
351    media: Option<media::MediaAnalysis>,
352    /// Set when a LATER fragment in the same request shares this one's
353    /// `kind == "screenshot"` and `metadata.target` value (US-009, PR2 —
354    /// see [`classify::screenshot_supersession`]): excluded from packing
355    /// entirely regardless of budget (see [`pack_items`]) and routed to
356    /// [`superseded_screenshot_verdict`] instead of the ordinary
357    /// pack-outcome verdicts. `analysis.rule` is left untouched (still
358    /// whatever [`classify::classify`] returned) — only this flag steers
359    /// packing and the decision; nothing else needs to know why.
360    superseded: bool,
361}
362
363impl Analysis<'_> {
364    /// The hash every `ctx://source/<hash>` handle (and thus every bridge
365    /// storage slot) for this fragment is minted from. **Media identity is
366    /// the raw decoded BYTES** ([`media::MediaAnalysis::raw_hash`]), exactly
367    /// like PR1's dedup — never the caption text: captions are typically
368    /// blank, and keying on them would collide every captionless image onto
369    /// one handle (serving arbitrary wrong bytes back). Two different
370    /// images therefore always get two different handles; byte-identical
371    /// images share one handle and resolve the same stored bytes (with the
372    /// kept instance's caption — divergent duplicate captions do not
373    /// survive, same as PR1's dedup semantics). Text fragments keep the
374    /// content hash, byte-identical to every pre-PR2 handle.
375    fn handle_hash(&self) -> u64 {
376        self.media
377            .as_ref()
378            .map_or(self.content_hash, |media| media.raw_hash)
379    }
380}
381
382/// A media fragment's total precomputed token cost: the image alone (from
383/// [`media::MediaAnalysis::image_tokens`]) plus its caption's own (usually
384/// tiny, often zero for a blank caption) text cost. Shared by [`analyze`]
385/// (feeds [`Analysis::tokens`]) and [`pieces`] (feeds the packed piece's
386/// cost) so the two can never drift apart — the same total is what gets
387/// budgeted and what gets reported as "emitted" once packed.
388fn media_fragment_tokens(
389    media: &media::MediaAnalysis,
390    caption: &str,
391    estimator: &dyn TokenEstimator,
392) -> u64 {
393    media
394        .image_tokens
395        .saturating_add(estimator.estimate(caption))
396}
397
398/// What actually got emitted for one packed fragment.
399struct Emission {
400    /// The emitted text (a prefix of the fragment's pieces, concatenated).
401    text: String,
402    /// Pieces taken / pieces available.
403    taken: usize,
404    /// Total pieces the fragment was cut into.
405    total: usize,
406}
407
408impl Emission {
409    /// Whether the fragment's pieces were all emitted.
410    fn is_full(&self) -> bool {
411        self.taken == self.total
412    }
413}
414
415/// Reject requests over the [`crate::limits`] caps and compute the usable
416/// budget (`clamped budget − reserve`).
417fn validate(request: &CompileRequest, policy: &CompilePolicy) -> Result<u64, MemoryError> {
418    // The pure core never resolves `path` (V2b-1): that is an adapter-side
419    // I/O pre-pass (`context::ingest::resolve_fragments`) that clears the
420    // field on success. A `path` still set here means either no adapter ran
421    // (e.g. a binding with no ingest support, such as the WASM build) or
422    // ingestion is disabled — both report the same explicit error rather
423    // than silently compiling an empty-content fragment.
424    if request.fragments.iter().any(|f| f.path.is_some()) {
425        return Err(MemoryError::IngestDisabled);
426    }
427    if request.fragments.len() > limits::MAX_FRAGMENTS {
428        return Err(MemoryError::ContextOverLimit(format!(
429            "{} fragments exceed the cap of {}",
430            request.fragments.len(),
431            limits::MAX_FRAGMENTS
432        )));
433    }
434    if let Some(oversized) = request
435        .fragments
436        .iter()
437        .find(|fragment| fragment.content.len() > limits::MAX_FRAGMENT_BYTES)
438    {
439        return Err(MemoryError::ContextOverLimit(format!(
440            "a fragment of {} bytes exceeds the cap of {} bytes",
441            oversized.content.len(),
442            limits::MAX_FRAGMENT_BYTES
443        )));
444    }
445    for fragment in &request.fragments {
446        let Some(metadata) = fragment.metadata.as_ref() else {
447            continue;
448        };
449        let bytes = limits::metadata_bytes(metadata);
450        if bytes > limits::MAX_METADATA_BYTES {
451            return Err(MemoryError::MetadataTooLarge {
452                bytes,
453                max: limits::MAX_METADATA_BYTES,
454            });
455        }
456    }
457    validate_media(&request.fragments)?;
458    let budget = limits::clamp_token_budget(request.token_budget);
459    let usable = budget.saturating_sub(policy.response_reserve_tokens);
460    if usable == 0 {
461        return Err(MemoryError::ContextBudget {
462            budget,
463            reserve: policy.response_reserve_tokens,
464        });
465    }
466    Ok(usable)
467}
468
469/// Reject a fragment whose media payload violates
470/// [`limits::MAX_MEDIA_BYTES`] or is not well-formed base64 — checked eagerly
471/// here, before any decoding/hashing/estimation downstream, so a malformed
472/// payload never reaches the pipeline (fail fast, same `INVALID_PARAMS`
473/// shape as every other cap in [`validate`]).
474fn validate_media(fragments: &[ContextFragment]) -> Result<(), MemoryError> {
475    let mut total_media_bytes: usize = 0;
476    for (seq, fragment) in fragments.iter().enumerate() {
477        let Some(media_ref) = &fragment.media else {
478            continue;
479        };
480        total_media_bytes = total_media_bytes.saturating_add(media_ref.bytes_b64.len());
481        if total_media_bytes > limits::MAX_TOTAL_MEDIA_BYTES {
482            return Err(MemoryError::ContextOverLimit(format!(
483                "total media payload exceeds the request cap of {} base64 bytes",
484                limits::MAX_TOTAL_MEDIA_BYTES
485            )));
486        }
487        if media_ref.bytes_b64.len() > limits::MAX_MEDIA_BYTES {
488            return Err(MemoryError::ContextOverLimit(format!(
489                "fragment #{seq} media payload of {} base64 bytes exceeds the cap of {} bytes",
490                media_ref.bytes_b64.len(),
491                limits::MAX_MEDIA_BYTES
492            )));
493        }
494        if !media::is_valid_base64(&media_ref.bytes_b64) {
495            return Err(MemoryError::ContextOverLimit(format!(
496                "fragment #{seq} media payload is not valid base64"
497            )));
498        }
499    }
500    Ok(())
501}
502
503/// Run classification, relevance scoring, and duplicate detection over the
504/// input order, hashing and estimating each fragment exactly once.
505fn analyze<'a>(
506    request: &'a CompileRequest,
507    policy: &CompilePolicy,
508    estimator: &dyn TokenEstimator,
509) -> Vec<Analysis<'a>> {
510    let contents: Vec<&str> = request
511        .fragments
512        .iter()
513        .map(|fragment| fragment.content.as_str())
514        .collect();
515    // Decode/analyze media exactly once per fragment (decoding is not
516    // free), reused below both to feed dedup's media namespace and to build
517    // each Analysis's own `media` field.
518    let media_analyses: Vec<Option<media::MediaAnalysis>> = request
519        .fragments
520        .iter()
521        .map(|fragment| fragment.media.as_ref().map(media::analyze))
522        .collect();
523    let media_hashes: Vec<Option<u64>> = media_analyses
524        .iter()
525        .map(|analysis| analysis.as_ref().map(|analysis| analysis.raw_hash))
526        .collect();
527    // Whole-batch pass, symmetric to `dedup::find_duplicates` below: needs
528    // every fragment's `kind` + `metadata.target` at once, which a per-
529    // fragment `classify::classify` call cannot see. Computed *before*
530    // dedup so the media namespace can re-anchor off a superseded fragment
531    // (see `dedup::find_duplicates`'s doc) instead of anchoring dedup on a
532    // screenshot that supersession has already excluded from packing.
533    let superseded_flags = classify::screenshot_supersession(&request.fragments);
534    let duplicates = dedup::find_duplicates(
535        &contents,
536        policy.near_dup_dedup,
537        &media_hashes,
538        &superseded_flags,
539    );
540    let query_terms = relevance::terms(&request.query);
541    let mut analyses: Vec<Analysis<'a>> = request
542        .fragments
543        .iter()
544        .zip(duplicates)
545        .zip(media_analyses)
546        .enumerate()
547        .map(|(seq, ((fragment, dup), media_analysis))| {
548            let content_hash = stable_id(&fragment.content);
549            let rule = classify::classify(fragment, policy);
550            let abstract_collapse = (rule.action == ContextAction::Abstract).then(|| {
551                classify::collapse_repeated_lines(
552                    &fragment.content,
553                    policy.normalize_log_timestamps,
554                )
555            });
556            let tokens = media_analysis.as_ref().map_or_else(
557                || estimator.estimate(&fragment.content),
558                |media| media_fragment_tokens(media, &fragment.content, estimator),
559            );
560            // A caller can opt out via `disabled_rules`, exactly like every
561            // other named rule — even though this one is not a `RULES` row.
562            let superseded = superseded_flags[seq]
563                && !policy
564                    .disabled_rules
565                    .iter()
566                    .any(|disabled| disabled == classify::SCREENSHOT_SUPERSEDED_RULE_ID);
567            Analysis {
568                seq,
569                fragment_id: fragment.id.unwrap_or(content_hash),
570                content_hash,
571                original: &fragment.content,
572                tokens,
573                rule,
574                relevance: relevance::lexical_relevance(&query_terms, &fragment.content),
575                priority: fragment.priority.unwrap_or(0),
576                dup,
577                abstract_collapse,
578                media: media_analysis,
579                superseded,
580            }
581        })
582        .collect();
583    retain_safe_duplicates(&mut analyses);
584    analyses
585}
586
587/// Keep a duplicate mark only when dropping the fragment loses nothing:
588/// the kept twin must be classified to emit **verbatim** (Preserve or Cache
589/// — an abstracted twin would collapse the duplicate's content), and a
590/// *critical* fragment is never sacrificed to near-deduplication (its bytes
591/// differ from the twin's, and its own classification demands them).
592fn retain_safe_duplicates(analyses: &mut [Analysis<'_>]) {
593    for index in 0..analyses.len() {
594        let Some(dup) = analyses[index].dup else {
595            continue;
596        };
597        let twin_verbatim = matches!(
598            analyses[dup.kept_seq].rule.action,
599            ContextAction::Preserve | ContextAction::Cache
600        );
601        let critical_near = dup.kind == DupKind::Near && analyses[index].rule.critical;
602        if !twin_verbatim || critical_near {
603            analyses[index].dup = None;
604        }
605    }
606}
607
608/// Build the packing input for every non-duplicate fragment: abstracted
609/// fragments emit their collapsed form as one piece, everything else is cut
610/// into budget-sized chunks.
611fn pack_items(
612    analyses: &[Analysis],
613    policy: &CompilePolicy,
614    usable: u64,
615    estimator: &dyn TokenEstimator,
616) -> Vec<PackItem> {
617    let chunk_policy = effective_chunk_policy(policy, usable, estimator);
618    analyses
619        .iter()
620        // A superseded screenshot (US-009, PR2) is never attempted, budget
621        // or no budget — see `Analysis::superseded`.
622        .filter(|analysis| analysis.dup.is_none() && !analysis.superseded)
623        .map(|analysis| PackItem {
624            seq: analysis.seq,
625            critical: analysis.rule.critical,
626            priority: analysis.priority,
627            relevance: analysis.relevance,
628            // Query-independent selection tier (issue #1455): see
629            // `budget::selection_order`.
630            cache: analysis.rule.action == ContextAction::Cache,
631            pieces: pieces(analysis, &chunk_policy, estimator),
632        })
633        .collect()
634}
635
636/// The emission pieces of one fragment.
637///
638/// A media fragment (US-009, PR1) is always exactly one atomic, all-or-
639/// nothing piece — never passed to [`chunk_text`], mirroring the
640/// `abstract.log_dedup` case below: packing can take it whole or not at all,
641/// never a byte-range prefix, so an image can never be cut mid-stream. Its
642/// text is only the caption (raw media bytes never become packable "piece"
643/// text); its cost is the precomputed [`media_fragment_tokens`] total, so
644/// packing never re-derives a media fragment's cost from `estimator.estimate`
645/// over an empty or near-empty caption.
646fn pieces(
647    analysis: &Analysis,
648    chunk_policy: &ChunkPolicy,
649    estimator: &dyn TokenEstimator,
650) -> Vec<budget::Piece> {
651    if let Some(media) = &analysis.media {
652        let cost = media_fragment_tokens(media, analysis.original, estimator);
653        return vec![budget::Piece {
654            text: analysis.original.to_owned(),
655            cost: Some(cost),
656        }];
657    }
658    if let Some((collapsed, _normalized)) = &analysis.abstract_collapse {
659        return vec![budget::Piece {
660            text: collapsed.clone(),
661            cost: None,
662        }];
663    }
664    chunk_text(analysis.original, chunk_policy)
665        .into_iter()
666        .map(|chunk| budget::Piece {
667            text: chunk.text,
668            cost: None,
669        })
670        .collect()
671}
672
673/// Lower bound on the pipeline's effective chunk size, regardless of budget
674/// or caller policy. Guards against memory-amplification: without a floor, a
675/// tiny `token_budget` (or a tiny caller-supplied `max_chunk_bytes`) would
676/// drive the ceiling toward one byte and explode a large fragment into one
677/// heap `String` per byte. At 256 bytes the per-piece `String` overhead is
678/// under 10 %, so pieces stay bounded by ~`input_bytes / 256` — no
679/// amplification beyond the already-capped input size ([`crate::limits`]).
680const MIN_CHUNK_BYTES: usize = 256;
681
682/// The chunk policy the pipeline actually cuts with: the ceiling tracks the
683/// usable budget (sized via the estimator's bytes-per-token hint) but is
684/// **floored at [`MIN_CHUNK_BYTES`]** so neither a tiny budget nor a tiny
685/// caller-supplied `max_chunk_bytes` can drive it toward a byte (a
686/// memory-amplification `DoS`). A budget too small to hold a floored piece
687/// simply externalizes everything, which is the correct outcome anyway.
688/// **Overlap is forced to zero** — pipeline pieces are emitted by
689/// concatenation, and an overlap prefix would duplicate every seam in
690/// content reported as verbatim; overlap stays meaningful only for the
691/// standalone [`chunk_text`] API. The byte ceiling is a *hint*: every piece
692/// is still measured by the injected estimator during packing.
693fn effective_chunk_policy(
694    policy: &CompilePolicy,
695    usable: u64,
696    estimator: &dyn TokenEstimator,
697) -> ChunkPolicy {
698    let budget_bytes = usize::try_from(usable.saturating_mul(estimator.bytes_per_token_hint()))
699        .unwrap_or(usize::MAX);
700    ChunkPolicy {
701        max_chunk_bytes: policy
702            .chunk
703            .max_chunk_bytes
704            .min(budget_bytes)
705            .max(MIN_CHUNK_BYTES),
706        overlap_bytes: 0,
707        boundary: policy.chunk.boundary,
708    }
709}
710
711/// Materialize what each packed fragment emits, keyed by `seq`. A fragment
712/// with no pieces at all (empty content) is kept here with `taken == total
713/// == 0` — trivially fully emitted, since there is nothing to lose — rather
714/// than dropped as "took none of what was offered", which is reserved for a
715/// fragment that had pieces and the budget could not fit any of them.
716fn emissions(items: &[PackItem], taken: &[usize]) -> BTreeMap<usize, Emission> {
717    items
718        .iter()
719        .zip(taken.iter().copied())
720        .filter(|&(item, count)| count > 0 || item.pieces.is_empty())
721        .map(|(item, count)| {
722            (
723                item.seq,
724                Emission {
725                    text: item.pieces[..count]
726                        .iter()
727                        .map(|piece| piece.text.as_str())
728                        .collect(),
729                    taken: count,
730                    total: item.pieces.len(),
731                },
732            )
733        })
734        .collect()
735}
736
737/// The output blocks: the cache-marked prefix first, then the body, both in
738/// input order.
739fn sections(analyses: &[Analysis], emissions: &BTreeMap<usize, Emission>) -> Vec<CompiledSection> {
740    let mut result = Vec::new();
741    for kind in [SectionKind::Cache, SectionKind::Body] {
742        let mut blocks: Vec<&str> = Vec::new();
743        let mut ids: Vec<u64> = Vec::new();
744        for analysis in analyses {
745            let cache = analysis.rule.action == ContextAction::Cache;
746            let wanted = (kind == SectionKind::Cache) == cache;
747            // Skip empty emissions: a trivially-emitted empty fragment
748            // (taken == total == 0) still gets its own decision, but must
749            // contribute no block — otherwise `join(JOINER)` would wrap it in
750            // joiners the packer never accounted for, breaking the budget
751            // invariant once more than one empty fragment is present.
752            if let Some(emission) = emissions
753                .get(&analysis.seq)
754                .filter(|emission| wanted && !emission.text.is_empty())
755            {
756                blocks.push(&emission.text);
757                ids.push(analysis.fragment_id);
758            }
759        }
760        if !blocks.is_empty() {
761            result.push(CompiledSection {
762                kind,
763                content: blocks.join(budget::JOINER),
764                fragment_ids: ids,
765            });
766        }
767    }
768    result
769}
770
771/// The auditable decision for one fragment.
772fn decision(
773    analysis: &Analysis,
774    all: &[Analysis],
775    emissions: &BTreeMap<usize, Emission>,
776) -> ContextDecision {
777    let emission = emissions.get(&analysis.seq);
778    let (action, rule_id, risk, reason, handle) = match (&analysis.dup, emission) {
779        (Some(dup), _) => dup_verdict(analysis, *dup, &all[dup.kept_seq], emissions),
780        // Checked before the emission-based arms: a superseded screenshot
781        // (US-009, PR2) is excluded from packing entirely (see
782        // `pack_items`), so `emission` is always `None` here anyway — this
783        // arm exists to give it its own rule id and reason rather than
784        // falling into the generic `externalized_verdict` below.
785        (None, _) if analysis.superseded => superseded_screenshot_verdict(analysis),
786        (None, Some(emission)) if emission.is_full() => full_verdict(analysis),
787        (None, Some(emission)) => partial_verdict(analysis, emission),
788        // A media fragment's single atomic piece is always taken whole or
789        // not at all (see `pieces`), so a missing emission for one means
790        // "did not fit" — never "took none of what was offered" from a
791        // multi-piece fragment. Media externalizes exactly like text
792        // (US-009, PR2): the memory bridge persists the bytes behind the
793        // handle this mints (see `MemoryService::retrieve_context_source`).
794        (None, None) => externalized_verdict(analysis),
795    };
796    ContextDecision {
797        fragment_id: analysis.fragment_id,
798        content_hash: analysis.content_hash,
799        action,
800        rule_id,
801        relevance: analysis.relevance,
802        risk,
803        reason,
804        memory_id: None,
805        handle,
806    }
807}
808
809/// The decision fields shared by every verdict builder.
810type Verdict = (ContextAction, String, FidelityRisk, String, Option<String>);
811
812/// The fidelity risk of content that did not make it fully into the output:
813/// **High** when the classification marked it critical (its loss matters),
814/// **Medium** otherwise. The single source of this policy — shared by the
815/// duplicate, partial, and externalized verdicts.
816fn critical_risk(critical: bool) -> FidelityRisk {
817    if critical {
818        FidelityRisk::High
819    } else {
820        FidelityRisk::Medium
821    }
822}
823
824/// A duplicate: dropped, and honest about whether its content actually
825/// survived. If the kept twin emitted fully the risk is low; if the twin was
826/// truncated or externalized the duplicate's content is *not* in the prompt,
827/// so the decision carries the elevated risk and stays machine-addressable
828/// through its own content-addressed handle.
829fn dup_verdict(
830    analysis: &Analysis,
831    dup: Duplicate,
832    twin: &Analysis,
833    emissions: &BTreeMap<usize, Emission>,
834) -> Verdict {
835    let (rule_id, variant) = match dup.kind {
836        DupKind::Exact => ("drop.duplicate", "exact duplicate"),
837        DupKind::Near => ("drop.near_duplicate", "near-duplicate"),
838    };
839    let twin_full = emissions.get(&twin.seq).is_some_and(Emission::is_full);
840    if twin_full {
841        // Media dedup keys on the image bytes alone: the twin carries the
842        // image, but a caption that differs from the twin's does NOT
843        // survive — say so instead of claiming full survival.
844        let caption_diverges = analysis.media.is_some() && analysis.original != twin.original;
845        let reason = if caption_diverges {
846            format!(
847                "{variant} of fragment #{} — image survives through it; this fragment's differing caption does not",
848                dup.kept_seq
849            )
850        } else {
851            format!(
852                "{variant} of fragment #{} — content survives through it",
853                dup.kept_seq
854            )
855        };
856        return (
857            ContextAction::Drop,
858            rule_id.to_owned(),
859            FidelityRisk::Low,
860            reason,
861            Some(provenance::handle_for(analysis.handle_hash())),
862        );
863    }
864    // Media dedup is otherwise unremarkable here: the twin's bytes were not
865    // fully packed either, but (US-009, PR2) the memory bridge now persists
866    // every non-duplicate fragment's original — media included — so a
867    // duplicate's own handle resolves exactly like a text duplicate's.
868    (
869        ContextAction::Drop,
870        rule_id.to_owned(),
871        critical_risk(analysis.rule.critical),
872        format!(
873            "{variant} of fragment #{} — but that twin was not fully emitted — recover via the handle",
874            dup.kept_seq
875        ),
876        Some(provenance::handle_for(analysis.handle_hash())),
877    )
878}
879
880/// A screenshot the whole-batch pass reclassified
881/// `retrieve.screenshot_superseded` (US-009, PR2 — see
882/// [`classify::screenshot_supersession`]): excluded from packing entirely,
883/// regardless of budget (see [`pack_items`]), because a LATER fragment in
884/// the same request already carries the current state of the same
885/// `metadata.target`. Always gets a resolvable handle — the memory bridge
886/// stores every non-duplicate fragment's original, media included.
887fn superseded_screenshot_verdict(analysis: &Analysis) -> Verdict {
888    (
889        ContextAction::Retrieve,
890        classify::SCREENSHOT_SUPERSEDED_RULE_ID.to_owned(),
891        FidelityRisk::Medium,
892        classify::SCREENSHOT_SUPERSEDED_REASON.to_owned(),
893        Some(provenance::handle_for(analysis.handle_hash())),
894    )
895}
896
897/// Fully emitted: the classification rule's action stands.
898fn full_verdict(analysis: &Analysis) -> Verdict {
899    let risk = if analysis.rule.action == ContextAction::Abstract {
900        FidelityRisk::Medium
901    } else {
902        FidelityRisk::Low
903    };
904    (
905        analysis.rule.action,
906        analysis.rule.id.to_owned(),
907        risk,
908        reason_with_normalization(analysis),
909        None,
910    )
911}
912
913/// Partially emitted: a chunk prefix is in, the rest stays retrievable.
914fn partial_verdict(analysis: &Analysis, emission: &Emission) -> Verdict {
915    (
916        analysis.rule.action,
917        analysis.rule.id.to_owned(),
918        critical_risk(analysis.rule.critical),
919        format!(
920            "{} — packed {}/{} chunks, the rest stays retrievable",
921            reason_with_normalization(analysis),
922            emission.taken,
923            emission.total
924        ),
925        Some(provenance::handle_for(analysis.handle_hash())),
926    )
927}
928
929/// The rule's base reason, with a mention of timestamp normalization
930/// appended when [`CompilePolicy::normalize_log_timestamps`] actually merged
931/// lines for this fragment (see [`Analysis::abstract_collapse`]) — an
932/// auditor asking "why did this log collapse the way it did?" sees the
933/// normalization in the same `reason` string as the rule that fired.
934fn reason_with_normalization(analysis: &Analysis) -> String {
935    match &analysis.abstract_collapse {
936        Some((_, true)) => {
937            format!(
938                "{} — timestamps normalized before collapsing",
939                analysis.rule.reason
940            )
941        }
942        _ => analysis.rule.reason.to_owned(),
943    }
944}
945
946/// Not emitted at all: externalized behind a retrieval handle.
947fn externalized_verdict(analysis: &Analysis) -> Verdict {
948    (
949        ContextAction::Retrieve,
950        "budget.externalize".to_owned(),
951        critical_risk(analysis.rule.critical),
952        format!(
953            "did not fit the budget ({}); retrievable via its handle",
954            analysis.rule.reason
955        ),
956        Some(provenance::handle_for(analysis.handle_hash())),
957    )
958}
959
960/// Relevance floor a [`ContextAction::Retrieve`] decision must clear to
961/// produce a [`ContextWarning`] (V2a-2 quick win). Chosen so the two
962/// existing compile goldens (neither carries a `Retrieve` decision) stay
963/// byte-identical; recalibrate against a wider corpus if warnings prove too
964/// noisy or too quiet in practice.
965const WARNING_RELEVANCE_THRESHOLD: f32 = 0.35;
966
967/// The warnings computed from `decisions` (V2a-2 quick win): every
968/// [`ContextAction::Retrieve`] decision at or above
969/// [`WARNING_RELEVANCE_THRESHOLD`]. Shared by [`ContextCompiler::finish`]
970/// (the pre-memory-annotation value) and the memory bridge, which
971/// recomputes it AFTER `annotate_memory_provenance` may have rewritten a
972/// pulled fragment's `relevance`/`reason` — a warning must never quote a
973/// stale value.
974///
975/// Deliberately scoped to `Retrieve` only, and that scope is NARROWER than
976/// "everything that was lost". The old justification here — "every `Drop` is
977/// a byte-identical duplicate whose content survives through its kept twin"
978/// — is false, and `dup_verdict` contradicts it in its own reason strings: a
979/// media duplicate whose caption diverges loses that caption, and a
980/// duplicate whose twin was itself not fully emitted loses the remainder.
981/// A partially packed `Preserve` and an `Abstract` are real losses too, and
982/// neither appears here either.
983///
984/// So the floor is a NOISE control, not a completeness claim: an empty
985/// `warnings` does not mean nothing was lost. `decisions` stays the
986/// exhaustive record and `risk` the cheap summary. [`ContextWarning`] and
987/// every published tool description now say this instead of promising the
988/// shortcut (#1703 DC-4).
989pub(crate) fn warnings_for(decisions: &[ContextDecision]) -> Vec<ContextWarning> {
990    decisions
991        .iter()
992        .filter(|decision| {
993            decision.action == ContextAction::Retrieve
994                && decision.relevance >= WARNING_RELEVANCE_THRESHOLD
995        })
996        .map(|decision| ContextWarning {
997            fragment_id: decision.fragment_id,
998            action: decision.action,
999            relevance: decision.relevance,
1000            reason: decision.reason.clone(),
1001        })
1002        .collect()
1003}
1004
1005/// Apply [`CompilePolicy::slim_response`]: empty `sections`/`decisions`,
1006/// leaving `content`/`insights`/`risk`/`warnings`/`sources`/`retrieval_handles`
1007/// untouched. Split out of [`ContextCompiler::compile`] so the memory bridge
1008/// can call it as its own LAST step, after annotating memory provenance and
1009/// recomputing `warnings` on the full `decisions` ([`ContextCompiler::compile_raw`]).
1010pub(crate) fn apply_slim(mut compiled: CompiledContext, policy: &CompilePolicy) -> CompiledContext {
1011    if policy.slim_response {
1012        compiled.sections.clear();
1013        compiled.decisions.clear();
1014    }
1015    compiled
1016}
1017
1018/// The handles of every fully externalized fragment, in decision order.
1019fn retrieval_handles(analyses: &[Analysis], decisions: &[ContextDecision]) -> Vec<RetrievalHandle> {
1020    analyses
1021        .iter()
1022        .zip(decisions)
1023        .filter(|(_, decision)| decision.action == ContextAction::Retrieve)
1024        .map(|(analysis, _)| RetrievalHandle {
1025            handle: provenance::handle_for(analysis.handle_hash()),
1026            fragment_id: analysis.fragment_id,
1027            estimated_tokens: analysis.tokens,
1028        })
1029        .collect()
1030}
1031
1032/// Tokens actually reflected in the output for one fragment. For an
1033/// ordinary fragment this is the injected estimator over whatever prefix of
1034/// pieces was emitted (unchanged pre-media behavior). For a media fragment
1035/// (US-009, PR1) packing is atomic (see `pieces`): an emission's mere
1036/// presence already means the *whole* precomputed cost (image +
1037/// caption — [`media_fragment_tokens`], the same total [`Analysis::tokens`]
1038/// holds) was spent, never a partial text-estimate of the caption alone —
1039/// which would misreport a fully preserved image as almost entirely
1040/// "saved" whenever its caption happens to be blank.
1041fn emitted_tokens(
1042    analysis: &Analysis,
1043    emissions: &BTreeMap<usize, Emission>,
1044    estimator: &dyn TokenEstimator,
1045) -> u64 {
1046    let Some(emission) = emissions.get(&analysis.seq) else {
1047        return 0;
1048    };
1049    if analysis.media.is_some() {
1050        analysis.tokens
1051    } else {
1052        estimator.estimate(&emission.text)
1053    }
1054}
1055
1056/// Attribute saved tokens to the rule that saved them. A fully emitted
1057/// verbatim fragment saves nothing, so every attribution comes from drops,
1058/// abstractions, externalizations, and partial packs — and the per-rule map
1059/// reconciles with the total up to joiner effects.
1060fn saved_by_rule(
1061    analyses: &[Analysis],
1062    decisions: &[ContextDecision],
1063    emissions: &BTreeMap<usize, Emission>,
1064    estimator: &dyn TokenEstimator,
1065) -> BTreeMap<String, u64> {
1066    let mut by_rule = BTreeMap::new();
1067    for (analysis, decision) in analyses.iter().zip(decisions) {
1068        let emitted = emitted_tokens(analysis, emissions, estimator);
1069        let saved = analysis.tokens.saturating_sub(emitted);
1070        if saved > 0 {
1071            *by_rule.entry(decision.rule_id.clone()).or_insert(0) += saved;
1072        }
1073    }
1074    by_rule
1075}
1076
1077#[cfg(test)]
1078#[path = "context/media_pipeline_tests.rs"]
1079mod media_pipeline_tests;
1080
1081#[cfg(test)]
1082#[path = "chunk_policy_tests.rs"]
1083mod chunk_policy_tests;
1084
1085/// #1703 DC-4. The published descriptions used to say that checking
1086/// `decisions` by hand was "only needed when `warnings` is non-empty". These
1087/// tests NAME that promise so it cannot be re-made silently: they pin the
1088/// shapes that are real losses AND produce no warning, which is exactly what
1089/// made the old shortcut false.
1090///
1091/// They are deliberately written against [`warnings_for`] rather than a full
1092/// compilation: the claim is about the filter, and a fixture that had to
1093/// drive a packer into a partial `Preserve` would pin the packer's tuning
1094/// instead of the contract.
1095#[cfg(test)]
1096#[path = "warning_completeness_tests.rs"]
1097mod warning_completeness_tests;