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;